Async suspend function listener not automatically acknowledged?

[rbraeunlich]

Hi everyone,
I was wondering about an error in my application (not part of my question). While trying to simplyfy my code, e.g. by using a suspend function in my listener, which is now being supported, I tried to find out if I have to do the ack by myself. I checked the documentation and also found this earlier question: #3290
This indicates that I don't have to do the ack myself. Nevertheless, when debugging my integration tests I could see that the first message never was acknowledged. When doing it myself, all tests pass.
The critical part seems to be this line:
https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java#L553

My suspend function doesn't return anything and therefore the adapter doesn't do the ack for me. The documentation only states "return types include CompletableFuture, Mono and Kotlin suspend functions". Not that my suspend function has to return anything.
Additionally, if I was to return a kotlinx.coroutines.Job or kotlinx.coroutines.Deferred the helper in org.springframework.kafka.listener.adapter.AdapterUtils#isAsyncReply wouldn't recognize them because it only checks for Mono and CompletableFuture.

So, I wonder if the documentation here needs to be more specific or if the code is wrong.

Cheers,
Ronny

[sobychacko]

@rbraeunlich, this may be a bug that we are not checking for the suspend return type in the framework. Could you set up a small, simple application where we can see this issue and debug it? Thanks!

[artembilan]

Yes. That's a bug.
We really just miss to see if the listener method is a suspend function.
That HandlerAdapter logic:

this.asyncReplies = AdapterUtils.isAsyncReply(invokerHandlerMethod.getMethod().getReturnType());

has to be fixed like:

this.asyncReplies =
				AdapterUtils.isAsyncReply(invokerHandlerMethod.getMethod().getReturnType())
						|| KotlinDetector.isSuspendingFunction(invokerHandlerMethod.getMethod());

Raising an issue from this discussion...

[rbraeunlich]

Did you create an issue @artembilan ?

[sobychacko]

Here is the issue: #3740
and the PR that was merged: #3742