Asynchronous server-side processing in a request/reply scenario #1189

dnitzan · 2019-08-07T12:57:43Z

Request/reply semantics requires that the server side listener method (annotated with @SendTo) return a response synchronously. There are use cases where server side request processing is asynchronous such that the listener method can't return a result immediately and must defer the response, releasing the consumer thread to handle other incoming requests. Once the result is available, an application thread would need to send it back to the client. This would be a feature similar to Servlet 3.0 Asynchronous Processing.

The text was updated successfully, but these errors were encountered:

garyrussell · 2019-08-07T14:23:30Z

We already added this for @RabbitListener last year.

However, managing the offsets for Kafka could make this tricky to implement - we wouldn't want to commit the offset until the async operation is complete.

If the next async operation completes first, and its offset committed, and then the first one fails, it's too late because its offset has already been implicitly committed.

Since Kafka message are not discretely acknowledged, going async with Kafka record consumption is really not recommended because of the risk of message loss and/or the complexity of managing offsets.

For this reason we didn't proceed here, but we can put it on the backlog and give it some more thought.

dnitzan · 2019-08-07T14:35:49Z

True, we faced the same issues building a custom request/reply mechanism on top of Spring Kafka. We eventually elected to just commit periodically (with ack-mode=time) regardless of downstream asynchronous processing and it is perfectly fine for us. If such a relaxed guarantee is sufficient, then it greatly simplifies the implementation.

garyrussell · 2019-08-07T14:55:48Z

OK; with that caveat; we'll try to get it into a future release.

dcheung2 · 2021-06-18T01:16:19Z

I think it is reasonable difficulty. But still may could be done even without a full prefect asynchronized flow.

for example.

when ackMode=BATCH
-> That container could async all the listeners within same BATCH but refer the ack until all messages in a poll() responsed

when ackMode=TIME
-> same as BATCH, but synchronize all once by time, not poll()

when ackMode=RECORD
-> No real concurrently processing. Just one message a time, but allow developers to use async API.

The real difficulty is timeout/retry management which make the state complex.

For another example,
if the first message jammed, the 2nd failed, but 3rd successes. should it ack or retry from where ?
but this is a problem business logic to worry about. not the middleware
That spring-kafka should provide some configuration.

Preparation for spring-projects#1189 Defer committing out of order offsets until the gaps are filled. Pause the consumer until all acks are acknowledged.

Preparation for #1189 Defer committing out of order offsets until the gaps are filled. Pause the consumer until all acks are acknowledged.

garyrussell · 2021-07-07T18:25:57Z

I have added support for out-of-order manual commits, which will now make implementing this possible.

Add note about increased duplicates possibility.

When pending acks are released due to a gap being "filled", wake the consumer if it is currently paused.

garyrussell · 2023-05-24T20:55:34Z

Also support Kotlin Coroutines, discussed here #2653

Wzy19930507 · 2023-12-13T05:53:14Z

Hi, @sobychacko @artembilan
may i pick it up.

sobychacko · 2023-12-13T15:15:09Z

@Wzy19930507 Certainly!

artembilan · 2023-12-13T15:20:28Z

I'm sorry. What is the plan for this issue?
Looks like Gary has added already enough functionality.
What else you see has to be fixed?

Thanks

garyrussell · 2023-12-13T15:33:16Z

What else you see has to be fixed?

Add support for Mono and Future listener method return types in the MMLA, similar to what we have in RabbitMQ.

This is now possible when out of order commits are enabled.

artembilan · 2023-12-13T15:57:06Z

Cool!
Then it sounds like a copy/paste from the: https://github.com/spring-projects/spring-amqp/blob/main/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java#L375

* Support `Mono` and `Future` * Support auto ack at async return scenario when manual commit * Support `KafkaListenerErrorHandler` * Add warn log if the container is not configured for out-of-order manual commit * Add async return test in `BatchMessagingMessageListenerAdapterTests` and `MessagingMessageListenerAdapterTests` * Add unit test async listener with `@SendTo` in `AsyncListenerTests` * Add `async-returns.adoc` and `whats-new.adoc`

* Support kotlin suspend * Add kotlin suspend test, async request/reply with `@KafkaListener` `@KafkaHandler` and `@SendTo` * Fix async-returns.adoc and async warn log in `MessagingMessageListenerAdapter` * Add dependency `org.jetbrains.kotlinx:kotlinx-coroutines-reactor:1.7.3`

* auto-detect async reply than coerce the out-of-order manual commit. * add new interface `HandlerMethodDetect` to detect handler args and return type. * add auto-detect async reply than coerce the out-of-order manual commit unit test at `@KafkaListener` `@KafkaHandler` scene. * modify async-returns.adoc

…handled Sending the result from a `KafkaListenerErrorHandler` was broken for `@KafkaHandler` because the send to expression was lost.