GH-3012: Non-blocking retries support @KafkaListener on class level #3105
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
@Wzy19930507 This PR is very large with many changes, making the review harder on our side. Can you split the PR into multiple ones? That way, it is easy for us to review; maybe you can isolate the components needed as a prerequisite for the feature into a separate one. And when that is merged, perform the feature changes in another one. However, if it is difficult for you to split the PR, please let us know, and we will try our best to review this as is. |
Good advice, let me try! And please allow me keep this PR, until complete prerequisite feature. |
Wzy19930507
added a commit
to Wzy19930507/spring-kafka
that referenced
this pull request
Mar 8, 2024
* Support process `@RetryableTopic` from Class in `RetryTopicConfigurationProvider`. * Support process `@DltHandler` when `@RetryableTopic` annotated on the Class in `RetryableTopicAnnotationProcessor`. part of spring-projects#3105
Wzy19930507
added a commit
to Wzy19930507/spring-kafka
that referenced
this pull request
Mar 8, 2024
* Support process `@RetryableTopic` from Class in `RetryTopicConfigurationProvider`. * Support process `@DltHandler` when `@RetryableTopic` annotated on the Class in `RetryableTopicAnnotationProcessor`. part of spring-projects#3105
sobychacko
pushed a commit
that referenced
this pull request
Mar 8, 2024
* `@RetryableTopic` support annotated on Class. * Support process `@RetryableTopic` from Class in `RetryTopicConfigurationProvider`. * Support process `@DltHandler` when `@RetryableTopic` annotated on the Class in `RetryableTopicAnnotationProcessor`. * add java doc * add a snippet in what-new.adoc and retry-config.adoc part of #3105 and this contributes to fixing #3105 eventually
Wzy19930507
added a commit
to Wzy19930507/spring-kafka
that referenced
this pull request
Mar 10, 2024
* `EndpointCustomizerFactory` adaptor `MultiMethodKafkaListenerEndpoint`. * `RetryTopicConfigurer.processAndRegisterEndpoint` support `@KafkaListener` on a class. * Add new class `EndpointHandlerMultiMethod` to handler multi method for retrying endpoints. part 2 of spring-projects#3105
Wzy19930507
added a commit
to Wzy19930507/spring-kafka
that referenced
this pull request
Mar 10, 2024
* `EndpointCustomizerFactory` adaptor `MultiMethodKafkaListenerEndpoint`. * `RetryTopicConfigurer.processAndRegisterEndpoint` support `@KafkaListener` on a class. * Add new class `EndpointHandlerMultiMethod` to handler multi method for retrying endpoints. part2 of spring-projects#3105 and this contributes to fixing spring-projects#3105 eventually
sobychacko
pushed a commit
that referenced
this pull request
Mar 13, 2024
* @RetryableTopic support for KafkaListener annotation on class part 2 * `EndpointCustomizerFactory` adaptor `MultiMethodKafkaListenerEndpoint`. * `RetryTopicConfigurer.processAndRegisterEndpoint` support `@KafkaListener` on a class. * Add new class `EndpointHandlerMultiMethod` to handler multi method for retrying endpoints. * Deprecated `EndpointCustomizerFactory.addSuffixesAndMethod`. * Document public API changes in `whats-new.adoc`(or javadoc). part2 of #3105 and this contributes to fixing #3105 eventually
Wzy19930507
force-pushed
the
GH-3012_class_level_non_blocking_retry
branch
from
18115b4
to
854962a
Compare
sobychacko
requested changes
Mar 15, 2024
| @@ -7,6 +7,8 @@ Version 2.9 changed the mechanism to bootstrap infrastructure beans; see xref:re | |||
| Achieving non-blocking retry / dlt functionality with Kafka usually requires setting up extra topics and creating and configuring the corresponding listeners. | |||
| Since 2.7 Spring for Apache Kafka offers support for that via the `@RetryableTopic` annotation and `RetryTopicConfiguration` class to simplify that bootstrapping. | |||
|
|
|||
| Since 3.2 Non-Blocking Retries support xref:kafka/receiving-messages/class-level-kafkalistener.adoc[@KafkaListener on a Class]. | |||
There was a problem hiding this comment.
Rewrite to something like - Since 3.2, Spring for Apache Kafka supports non-blocking retries... (with the link).
| @@ -28,6 +28,21 @@ public void processMessage(MyPojo message) { | |||
| } | |||
| ---- | |||
|
|
|||
| Since 3.2, `@RetryableTopic` support @KafkaListener on a class would be: | |||
There was a problem hiding this comment.
@RetryableTopic support for
| Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass, | ||
| (ReflectionUtils.MethodFilter) method -> | ||
| AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null); | ||
| final List<Method> multiMethods = new ArrayList<>(methodsWithHandler); |
There was a problem hiding this comment.
No need to use final in local variables.
| @@ -165,6 +165,19 @@ | |||
| * }</code> | |||
| * }</code> | |||
| *</pre> | |||
| * <p> Since 3.2 , {@link org.springframework.kafka.annotation.RetryableTopic} annotation support | |||
| ExistingRetryTopicClassLevelIntegrationTests.MAIN_TOPIC_WITH_PARTITION_INFO, | ||
| ExistingRetryTopicClassLevelIntegrationTests.RETRY_TOPIC_WITH_PARTITION_INFO}, partitions = 4) | ||
| @TestPropertySource(properties = "two.attempts=2") | ||
| public class ExistingRetryTopicClassLevelIntegrationTests { |
There was a problem hiding this comment.
We can leave the access level at default (package-protected).
|
|
||
| } | ||
|
|
||
| public static class ShouldRetryOnlyByTopicException extends RuntimeException { |
| } | ||
| } | ||
|
|
||
| public static class ShouldSkipBothRetriesException extends RuntimeException { |
There was a problem hiding this comment.
public? same for other similar classes in this test class.
| RetryTopicClassLevelIntegrationTests.TWO_LISTENERS_TOPIC, | ||
| RetryTopicClassLevelIntegrationTests.MANUAL_TOPIC }) | ||
| @TestPropertySource(properties = { "five.attempts=5", "kafka.template=customKafkaTemplate"}) | ||
| public class RetryTopicClassLevelIntegrationTests { |
| } | ||
|
|
||
| @SuppressWarnings("serial") | ||
| public static class MyRetryException extends RuntimeException { |
| } | ||
|
|
||
| @Configuration | ||
| public static class KafkaProducerConfig { |
There was a problem hiding this comment.
Thanks for your review, fix it.
…n class level Fixes: spring-projects#3012 * Non-blocking retries support `@KafkaListener` on class level, include `@RetryableTopic`, `@DltHandler`, `RetryTopicConfiguration`, `RetryTopicConfigurationSupport`. * Doc and test Non-blocking retries support @KafkaListener on class level.
Wzy19930507
force-pushed
the
GH-3012_class_level_non_blocking_retry
branch
from
854962a
to
fe24063
Compare
|
Thanks, @Wzy19930507, for this excellent new feature. Just a reminder: limit the commit message titles to only 50 or fewer characters. Commit message descriptions can be as long and descriptive as you want. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Fixes: #3012
@KafkaListeneron class level, include@RetryableTopic,@DltHandler,RetryTopicConfiguration,RetryTopicConfigurationSupport.@KafkaListeneron class level.Part 1: Support process annotation
@RetryableTopicand@DltHandlerwhen@RetryableTopicannotated on class, see #3107.Part 2:
RetryTopicConfigurer#processAndRegisterEndpointsupportMultiMethodKafkaListenerEndpoint, see #3112.Part 3:
@EnableKafkaintegration feature that non-blocking retries support@KafkaListeneron class level, see #3105.