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

Wzy19930507 · 2024-01-18T03:02:19Z

Resolves #1189

support Mono and Future
support kotlin suspend
Auto detect async reply
@SendTo for @KafkaHandler after error is handled

Refactor MessagingMessageListenerAdapter

move BatchMessagingMessageListenerAdapter#invoke and RecordMessagingMessageListenerAdapter#invoke to MessagingMessageListenerAdapter
move KafkaListenerErrorHandler to MessagingMessageListenerAdapter
add @Nullable to KafkaListenerErrorHandler

artembilan

This is great, but looks like it require more work to make it perfect.

Also see the warning in your new Kotlin test:

 w: file:///home/runner/work/spring-kafka/spring-kafka/spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinCoroutinesTests.kt:100:14 Unsafe use of a nullable receiver of type ConsumerRecord<String!, String!>?

...ng-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/async-returns.adoc

...ain/java/org/springframework/kafka/annotation/ContinuationHandlerMethodArgumentResolver.java

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AdapterUtils.java

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerMethodDetect.java

...ng-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

...va/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapterTests.java

...st/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapterTests.java

...ng-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

artembilan

This is very close to the final solution.
Please, address those my final "at a glance" reviews and I'll pull it locally to play with and possible merge.

Thanks

...va/org/springframework/kafka/listener/adapter/ContinuationHandlerMethodArgumentResolver.java

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerMethodDetect.java

artembilan · 2024-01-19T17:06:18Z

Probably some race condition.
Since coroutine is an async you might have some thread barrier to wait for the reply to be fulfilled.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AsyncRepliesAware.java

...va/org/springframework/kafka/listener/adapter/ContinuationHandlerMethodArgumentResolver.java

artembilan · 2024-01-19T17:31:38Z

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

+		catch (Exception ex) {
+		}
+		this.logger.error(t, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
+		acknowledge(acknowledgment);


Why do we ack positively in case of error?
Spring AMQP logic is like this:

protected void asyncFailure(Message request, Channel channel, Throwable t, Object source) { this.logger.error("Future, Mono, or suspend function was completed with an exception for " + request, t); try { channel.basicNack(request.getMessageProperties().getDeliveryTag(), false, ContainerUtils.shouldRequeue(this.defaultRequeueRejected, t, this.logger)); } catch (IOException e) { this.logger.error("Failed to nack message", e); } }

Don't we have something like that in Spring Kafka as well?

I'm having some trouble, function Acknowlege.nack not support out-of-order commit now.
Another Similar functions is recoverer(in particular DLT), but not sure how to implement it yet.

Could we temporary use KafkaListenerErrorHandler?

OK. I see you do use it in the handleException():

Object errorResult = this.errorHandler.handleError(message, e, consumer, acknowledgment);

Then it goes to the:

handleResult(Objects.requireNonNullElse(result, errorResult), records, acknowledgment, consumer, message);

So, probably it is too late over here to make some assumption and do this unconditional acknowledge(acknowledgment).

I mean it feel like this call has to be removed from here

Wzy19930507 · 2024-01-19T18:02:48Z

Probably some race condition.
Since coroutine is an async you might have some thread barrier to wait for the reply to be fulfilled.

After kafka client 2.4 producer use sticky partition, it random chose partition and topic default partitions is 2.

artembilan · 2024-01-19T18:11:03Z

After kafka client 2.4 producer use sticky partition, it random chose partition and topic default partitions is 2.

So, consider to configure that @EmbeddedKafka to provide just one partition per topic.

Wzy19930507 · 2024-01-19T18:12:57Z

So, consider to configure that @EmbeddedKafka to provide just one partition per topic.

Thanks, nice idea.

artembilan · 2024-01-19T18:55:25Z

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

+		catch (Exception ex) {
+		}
+		this.logger.error(t, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
+		acknowledge(acknowledgment);


OK. I see you do use it in the handleException():

Object errorResult = this.errorHandler.handleError(message, e, consumer, acknowledgment);

Then it goes to the:

handleResult(Objects.requireNonNullElse(result, errorResult), records, acknowledgment, consumer, message);

So, probably it is too late over here to make some assumption and do this unconditional acknowledge(acknowledgment).

I mean it feel like this call has to be removed from here

Wzy19930507 · 2024-01-19T19:07:27Z

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

+	protected void handleException(Object records, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
+			Message<?> message, ListenerExecutionFailedException e) {
+
+		if (this.errorHandler != null) {


I mean it feel like this call has to be removed from here

~~When this.errorHandler != null, need acknowledgment skip error record.~~

OK. The idea is to ack the record we have just handled via errorHandler.
Isn't that this error handler responsibility then?
I'm not sure though what we should do otherwise, when this.errorHandler == null...

When this.errorHandler != null, need acknowledgment skip error record.

Sorry, misdescription, When this.errorHandler == null will throw origin exception or errorHandler threw exception, need acknowledgment skip error record. at out of order commit mode, if not ack, will pause the consumer.

* move `BatchMessagingMessageListenerAdapter#invoke` and `RecordMessagingMessageListenerAdapter#invoke` to `MessagingMessageListenerAdapter` * move `KafkaListenerErrorHandler` to `MessagingMessageListenerAdapter` * add `@Nullable` to `KafkaListenerErrorHandler`

* 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.

* move class from package `annotation` to package `adapter` * re name bar,baz in BatchMessagingMessageListenerAdapterTests * poblish unit test `MessagingMessageListenerAdapterTests` and `EnableKafkaKotlinCoroutinesTests` * poblish doc async-returns.adoc and nav.adoc

* rename `HandlerMethodDetect` to `AsyncRepliesAware` * fix javadoc in `ContinuationHandlerMethodArgumentResolver`

After kafka client 2.4 producer uses sticky partition, its randomly chose partition and topic default partitions is 2, configure that @embeddedkafka to provide just one partition per topic. * javadoc in `AsyncRepliesAware` * fix test in EnableKafkaKotlinCoroutinesTests * polish adoc

* polish `HandlerAdapter` * change `InvocationResult` to record * Optimization `MessagingMessageListenerAdapter.asyncFailure`

Wzy19930507 · 2024-01-30T03:16:40Z

Sorry i do some wrong when git rebase action, should i reopen RP to keep the RP clean?

artembilan · 2024-01-30T03:25:05Z

I believe you can just do git rebase main if your main is up today.

if that doesn’t help, no worries: we will rebase and squash on merge.

we didn’t forget about you: just busy with some other urgent stuff.

thank you for updates anyway !

Wzy19930507 · 2024-01-30T07:41:30Z

I solved this problem and thanks for your patient answer.

...ng-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/async-returns.adoc

artembilan requested changes Jan 18, 2024

View reviewed changes

Wzy19930507 commented Jan 19, 2024

View reviewed changes

...ng-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java Show resolved Hide resolved

Wzy19930507 force-pushed the GH-1189 branch from 6d82f4f to 4cd372b Compare January 19, 2024 03:54

artembilan requested changes Jan 19, 2024

View reviewed changes

...va/org/springframework/kafka/listener/adapter/ContinuationHandlerMethodArgumentResolver.java Show resolved Hide resolved

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerMethodDetect.java Outdated Show resolved Hide resolved

Wzy19930507 force-pushed the GH-1189 branch from 8f1c20a to bf8bb23 Compare January 19, 2024 16:20

This comment was marked as resolved.

Sign in to view

artembilan requested changes Jan 19, 2024

View reviewed changes

Wzy19930507 commented Jan 19, 2024

View reviewed changes

Wzy19930507 requested a review from artembilan January 30, 2024 02:48

Zhiyang.Wang1 and others added 9 commits January 30, 2024 10:55

spring-projectsGH-1189: @SendTo for @KafkaHandler after error is …

412d320

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

review fix

92639b6

* rename `HandlerMethodDetect` to `AsyncRepliesAware` * fix javadoc in `ContinuationHandlerMethodArgumentResolver`

* polish DelegatingInvocableHandler and add javadoc

76a2e0b

* polish `HandlerAdapter` * change `InvocationResult` to record * Optimization `MessagingMessageListenerAdapter.asyncFailure`

Wzy19930507 force-pushed the GH-1189 branch from d61dd33 to 76a2e0b Compare January 30, 2024 07:38

artembilan requested changes Jan 31, 2024

View reviewed changes

...ng-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/async-returns.adoc Outdated Show resolved Hide resolved

Mention version in the doc for async return types

916dfaf

artembilan reviewed Jan 31, 2024

View reviewed changes

...ng-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/async-returns.adoc Outdated Show resolved Hide resolved

Fix typo for the sentence in doc

865de6a

artembilan enabled auto-merge (squash) January 31, 2024 18:26

artembilan merged commit b44d742 into spring-projects:main Jan 31, 2024
3 checks passed

Wzy19930507 deleted the GH-1189 branch February 1, 2024 02:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

Wzy19930507 commented Jan 18, 2024 •

edited

Loading

artembilan left a comment

artembilan left a comment

This comment was marked as resolved.

artembilan commented Jan 19, 2024

artembilan Jan 19, 2024

Wzy19930507 Jan 19, 2024

artembilan Jan 19, 2024

Wzy19930507 commented Jan 19, 2024

artembilan commented Jan 19, 2024

Wzy19930507 commented Jan 19, 2024 •

edited

Loading

artembilan Jan 19, 2024

Wzy19930507 Jan 19, 2024 •

edited

Loading

artembilan Jan 19, 2024

Wzy19930507 Jan 20, 2024 •

edited

Loading

Wzy19930507 commented Jan 30, 2024

artembilan commented Jan 30, 2024

Wzy19930507 commented Jan 30, 2024 •

edited

Loading

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

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

Conversation

Wzy19930507 commented Jan 18, 2024 • edited Loading

artembilan left a comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

This comment was marked as resolved.

artembilan commented Jan 19, 2024

artembilan Jan 19, 2024

Choose a reason for hiding this comment

Wzy19930507 Jan 19, 2024

Choose a reason for hiding this comment

artembilan Jan 19, 2024

Choose a reason for hiding this comment

Wzy19930507 commented Jan 19, 2024

artembilan commented Jan 19, 2024

Wzy19930507 commented Jan 19, 2024 • edited Loading

artembilan Jan 19, 2024

Choose a reason for hiding this comment

Wzy19930507 Jan 19, 2024 • edited Loading

Choose a reason for hiding this comment

artembilan Jan 19, 2024

Choose a reason for hiding this comment

Wzy19930507 Jan 20, 2024 • edited Loading

Choose a reason for hiding this comment

Wzy19930507 commented Jan 30, 2024

artembilan commented Jan 30, 2024

Wzy19930507 commented Jan 30, 2024 • edited Loading

Wzy19930507 commented Jan 18, 2024 •

edited

Loading

Wzy19930507 commented Jan 19, 2024 •

edited

Loading

Wzy19930507 Jan 19, 2024 •

edited

Loading

Wzy19930507 Jan 20, 2024 •

edited

Loading

Wzy19930507 commented Jan 30, 2024 •

edited

Loading