GH-2239: RetryableTopic Refactoring #2328

garyrussell · 2022-07-05T20:58:32Z

Use a Task Scheduler to resume partitions, avoiding the need for timing adjustment.

cherry-pick to 2.9.x

I will make a What's New commit after the review/merge because it will certainly conflict.

I have left the commits separate to avoid you having to re-review what you have already looked at (if you so desire).

cc/ @tomazfernandes

tomazfernandes

Just a couple nits, LGTM as is, thanks!

tomazfernandes · 2022-07-05T21:21:51Z

...main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -510,29 +508,11 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 	}

 	private RetryTopicConfigurer getRetryTopicConfigurer() {
-		bootstrapRetryTopicIfNecessary();
 		return this.beanFactory.containsBean("internalRetryTopicConfigurer")


We can remove these String bean names and leave only the RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME

tomazfernandes · 2022-07-05T21:23:23Z

spring-kafka/src/main/java/org/springframework/kafka/listener/BackOffHandler.java

 */
-@FunctionalInterface


Just curious: why remove the annotation?

Thanks for pointing this out - originally I added the new method as abstract but I changed it to default later.

Since it is a new interface, I think I'll make the other method default too; while the methods are related, we shouldn't force the user to implement both if they only want the new one.

tomazfernandes · 2022-07-05T22:04:59Z

...g-kafka/src/main/java/org/springframework/kafka/listener/ContainerPausingBackOffHandler.java

@@ -51,4 +53,9 @@ public void onNextBackOff(@Nullable MessageListenerContainer container, Exceptio
 		}
 	}

+	@Override
+	public void onNextBackOff(MessageListenerContainer container, TopicPartition partition, long nextBackOff) {


Given the interface states container is @Nullable, maybe a null check is in order here?

It's @Nullable on the other method, but not this one.

tomazfernandes · 2022-07-05T22:08:19Z

Thanks @garyrussell, this looks great!

There are a few other deprecations that I'm not sure if you missed or just decided not to deal with in this PR, such as:

This part of the RetryableTopicAnnotationProcessor class can be removed, and with that we have no further use for RetryTopicInternalBeanNames and RetryTopicInternalBeanNamesTests.
We can also remove the configure methods from ListenerContainerFactoryConfigurer, and kill some other logic there, such as isSetContainerProperties and related logic.
Also, a couple of lines with String bean names in ListenerContainerFactoryResolver, and a couple @SupressWarnings in RetryTopicConfigurer.

Of course, these shouldn't change much overall, and can be part of a different PR or left for 3.0, so no worries if you prefer keeping them.

tomazfernandes · 2022-07-06T15:41:10Z

Looked at the new deprecations removal and LGTM, thanks! All this slimming down will definitely make maintenance easier going forward.

artembilan

I left some review, but I guess all of that is optional.
And may be revised in other issues/PR.

Thank you for a great overhaul! 😄

artembilan · 2022-07-06T16:51:46Z

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

-This way you can tune the precision and performance for the retry topics if you need to.
-
-NOTE: You can have separate `ListenerContainerFactory` instances for the main and retry topics - this way you can have different settings to better suit your needs, such as having a higher polling timeout setting for the main topics and a lower one for the retry topics.
+Starting with version 2.9, it is no longer necessary to tune the precision because a task scheduler is used to resume the partition and wake the consumer, if necessary.


Why to say this if the rest of the paragraph is just removed?
May be remove the whole paragraph altogether?..

artembilan · 2022-07-06T16:56:24Z

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

+    protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
+        // Use the new 2.9 mechanism to avoid re-fetching the same records after a pause
+        customizersConfigurer.customizeErrorHandler(eh -> {
+            ((DefaultErrorHandler) eh).setSeekAfterError(false);


Looks like there is some inconsistency in the API:

The CustomizersConfigurer has customizeErrorHandler(Consumer<CommonErrorHandler> errorHandlerCustomizer).

The ListenerContainerFactoryConfigurer does:

DefaultErrorHandler errorHandler = createDefaultErrorHandlerInstance(deadLetterPublishingRecoverer); ... this.errorHandlerCustomizer.accept(errorHandler);

Maybe the first one has to be changed to always accept a DefaultErrorHandler instead?..

I see you have fixed the target API for this setErrorHandlerCustomizer(Consumer<DefaultErrorHandler> errorHandlerCustomizer). Therefore this cast is redundant in this doc.

artembilan · 2022-07-06T16:59:30Z

...main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

-			this.beanFactory.getBean("internalRetryTopicBootstrapper",
-					org.springframework.kafka.retrytopic.RetryTopicBootstrapper.class).bootstrapRetryTopic();
-		}
+		return this.beanFactory.getBean(RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME,


Shouldn't we cache the result of this call?
I see that getRetryTopicConfigurer() is called for every single @KafkaListener in the project...

artembilan · 2022-07-06T17:12:44Z

...ka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java

 			try {
-				return this.beanFactory.getBean(RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME,
-						KafkaOperations.class);
+				return this.beanFactory.getBean(DEFAULT_SPRING_BOOT_KAFKA_TEMPLATE_NAME, KafkaOperations.class);


This is semantic cycle with Spring Boot.
Why just plain getBean(KafkaOperations.class) is not enough?

spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerContainerRegistry.java

artembilan · 2022-07-06T17:30:28Z

...rc/main/java/org/springframework/kafka/listener/ContainerPartitionPausingBackOffManager.java

+		getListenerContainerFromContext(context).pausePartition(topicPartition);
+		this.backOffHandler.onNextBackOff(getListenerContainerFromContext(context), topicPartition, backOffTime);
+		throw new KafkaBackoffException(String.format("Partition %s from topic %s is not ready for consumption, " +
+				"backing off for approx. %s millis.", context.getTopicPartition().partition(),


Looks like you have already extracted a topicPartition variable...

.../java/org/springframework/kafka/listener/ContainerPartitionPausingBackOffManagerFactory.java

...ng-kafka/src/main/java/org/springframework/kafka/listener/ListenerContainerPauseService.java

...kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicSchedulerWrapper.java

New back of manager (and factory) that uses a task scheduler to resume the paused partitions. Revert change to deprecated PartitionPausingBackoffManager. Log resume.

Also fix unrelated race in EKIT. Only allow one `RetryTemplateConfigurationSupport` bean.

Resolves spring-projects#2239

tomazfernandes · 2022-07-06T22:48:57Z

...main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -508,8 +510,16 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 	}

 	private RetryTopicConfigurer getRetryTopicConfigurer() {
-		return this.beanFactory.getBean(RetryTopicBeanNames.RETRY_TOPIC_CONFIGURER_BEAN_NAME,
-				RetryTopicConfigurer.class);
+		try {


Isn't this missing something like if (this.topicConfigurer == null)?

doh. Thanks.

artembilan

Just one minor nit-pick for the doc.

Thanks

artembilan · 2022-07-07T13:59:56Z

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

+    protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
+        // Use the new 2.9 mechanism to avoid re-fetching the same records after a pause
+        customizersConfigurer.customizeErrorHandler(eh -> {
+            ((DefaultErrorHandler) eh).setSeekAfterError(false);


I see you have fixed the target API for this setErrorHandlerCustomizer(Consumer<DefaultErrorHandler> errorHandlerCustomizer). Therefore this cast is redundant in this doc.

artembilan · 2022-07-07T14:49:31Z

Cherry-picked to 2.9.x as b88cbf9 after some minor conflict resolutions.

tomazfernandes approved these changes Jul 5, 2022

View reviewed changes

artembilan requested changes Jul 6, 2022

View reviewed changes

garyrussell force-pushed the GH-2239 branch 2 times, most recently from f9cbe4e to 35eb5d0 Compare July 6, 2022 20:53

garyrussell added 7 commits July 6, 2022 17:16

spring-projectsGH-2239: Replace PartitionPausingBackOffManager

9acdbde

New back of manager (and factory) that uses a task scheduler to resume the paused partitions. Revert change to deprecated PartitionPausingBackoffManager. Log resume.

Remove legacy code.

9f869b6

Also fix unrelated race in EKIT. Only allow one `RetryTemplateConfigurationSupport` bean.

Fix static var.

7951aa1

Docs.

386512a

Resolves spring-projects#2239

More docs.

f48a168

Remove more dead/deprecated code.

e5e3a84

Address PR Comments.

529fd45

tomazfernandes reviewed Jul 6, 2022

View reviewed changes

Fix RetryTopicConfigurer bean retrieval.

69f881b

garyrussell force-pushed the GH-2239 branch from 35eb5d0 to 69f881b Compare July 7, 2022 13:06

artembilan requested changes Jul 7, 2022

View reviewed changes

Remove unnecessary casts in doc.

bff0183

artembilan merged commit 37910b6 into spring-projects:main Jul 7, 2022

garyrussell deleted the GH-2239 branch July 7, 2022 14:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GH-2239: RetryableTopic Refactoring #2328

GH-2239: RetryableTopic Refactoring #2328

garyrussell commented Jul 5, 2022

tomazfernandes left a comment

tomazfernandes Jul 5, 2022

tomazfernandes Jul 5, 2022

garyrussell Jul 6, 2022

tomazfernandes Jul 5, 2022

garyrussell Jul 6, 2022

tomazfernandes commented Jul 5, 2022

tomazfernandes commented Jul 6, 2022

artembilan left a comment

artembilan Jul 6, 2022

artembilan Jul 6, 2022

artembilan Jul 7, 2022

artembilan Jul 6, 2022

artembilan Jul 6, 2022

artembilan Jul 6, 2022

tomazfernandes Jul 6, 2022

garyrussell Jul 7, 2022

artembilan left a comment

artembilan Jul 7, 2022

artembilan commented Jul 7, 2022

		*/
		@FunctionalInterface

GH-2239: RetryableTopic Refactoring #2328

GH-2239: RetryableTopic Refactoring #2328

Conversation

garyrussell commented Jul 5, 2022

tomazfernandes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tomazfernandes commented Jul 5, 2022

tomazfernandes commented Jul 6, 2022

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan commented Jul 7, 2022