From abdc2e1b4f95a7be589aa5e930f89cf19227e5ff Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 8 May 2019 12:58:47 -0400 Subject: [PATCH 1/2] Auto-configure Kafka listener container with rebalance listener This commit associates a `ConsumerAwareRebalanceListener` to the auto-configured listener container factory if a single instance is found in the context. See gh-16755 --- ...fkaListenerContainerFactoryConfigurer.java | 13 ++++++++++ .../KafkaAnnotationDrivenConfiguration.java | 8 ++++++- .../kafka/KafkaAutoConfigurationTests.java | 24 +++++++++++++++++++ 3 files changed, 44 insertions(+), 1 deletion(-) diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java index 1631f5980d..e5775b6b61 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java @@ -25,6 +25,7 @@ import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AfterRollbackProcessor; import org.springframework.kafka.listener.BatchErrorHandler; +import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.support.converter.MessageConverter; @@ -53,6 +54,8 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { private AfterRollbackProcessor afterRollbackProcessor; + private ConsumerAwareRebalanceListener rebalanceListener; + /** * Set the {@link KafkaProperties} to use. * @param properties the properties @@ -111,6 +114,15 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { this.afterRollbackProcessor = afterRollbackProcessor; } + /** + * Set the {@link ConsumerAwareRebalanceListener} to use. + * @param rebalanceListener the rebalance listener. + * @since 2.2 + */ + void setRebalanceListener(ConsumerAwareRebalanceListener rebalanceListener) { + this.rebalanceListener = rebalanceListener; + } + /** * Configure the specified Kafka listener container factory. The factory can be * further tuned and default settings can be overridden. @@ -160,6 +172,7 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig); map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal); map.from(this.transactionManager).to(container::setTransactionManager); + map.from(this.rebalanceListener).to(container::setConsumerRebalanceListener); } } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java index 34e78548c9..5f8b009803 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java @@ -29,6 +29,7 @@ import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AfterRollbackProcessor; import org.springframework.kafka.listener.BatchErrorHandler; +import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.support.converter.BatchMessageConverter; import org.springframework.kafka.support.converter.BatchMessagingMessageConverter; @@ -63,6 +64,8 @@ class KafkaAnnotationDrivenConfiguration { private final AfterRollbackProcessor afterRollbackProcessor; + private final ConsumerAwareRebalanceListener rebalanceListener; + KafkaAnnotationDrivenConfiguration(KafkaProperties properties, ObjectProvider messageConverter, ObjectProvider batchMessageConverter, @@ -70,7 +73,8 @@ class KafkaAnnotationDrivenConfiguration { ObjectProvider> kafkaTransactionManager, ObjectProvider errorHandler, ObjectProvider batchErrorHandler, - ObjectProvider> afterRollbackProcessor) { + ObjectProvider> afterRollbackProcessor, + ObjectProvider rebalanceListener) { this.properties = properties; this.messageConverter = messageConverter.getIfUnique(); this.batchMessageConverter = batchMessageConverter.getIfUnique( @@ -80,6 +84,7 @@ class KafkaAnnotationDrivenConfiguration { this.errorHandler = errorHandler.getIfUnique(); this.batchErrorHandler = batchErrorHandler.getIfUnique(); this.afterRollbackProcessor = afterRollbackProcessor.getIfUnique(); + this.rebalanceListener = rebalanceListener.getIfUnique(); } @Bean @@ -95,6 +100,7 @@ class KafkaAnnotationDrivenConfiguration { configurer.setErrorHandler(this.errorHandler); configurer.setBatchErrorHandler(this.batchErrorHandler); configurer.setAfterRollbackProcessor(this.afterRollbackProcessor); + configurer.setRebalanceListener(this.rebalanceListener); return configurer; } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java index 935af69698..585675c265 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java @@ -55,6 +55,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaAdmin; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AfterRollbackProcessor; +import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.ContainerProperties.AckMode; import org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler; @@ -674,6 +675,18 @@ public class KafkaAutoConfigurationTests { }); } + @Test + public void testConcurrentKafkaListenerContainerFactoryWithCustomRebalanceListener() { + this.contextRunner.withUserConfiguration(RebalanceListenerConfiguration.class) + .run((context) -> { + ConcurrentKafkaListenerContainerFactory factory = context + .getBean(ConcurrentKafkaListenerContainerFactory.class); + assertThat(factory.getContainerProperties()) + .hasFieldOrPropertyWithValue("consumerRebalanceListener", + context.getBean("rebalanceListener")); + }); + } + @Test public void testConcurrentKafkaListenerContainerFactoryWithKafkaTemplate() { this.contextRunner.run((context) -> { @@ -749,6 +762,17 @@ public class KafkaAutoConfigurationTests { } + @Configuration(proxyBeanMethods = false) + protected static class RebalanceListenerConfiguration { + + @Bean + public ConsumerAwareRebalanceListener rebalanceListener() { + return new ConsumerAwareRebalanceListener() { + }; + } + + } + @Configuration(proxyBeanMethods = false) @EnableKafkaStreams protected static class EnableKafkaStreamsConfiguration { From 74208bb1a7d593759d52280be7748815b90741b4 Mon Sep 17 00:00:00 2001 From: Stephane Nicoll Date: Tue, 21 May 2019 09:49:34 +0200 Subject: [PATCH 2/2] Polish "Auto-configure Kafka listener container with rebalance listener" Closes gh-16755 --- ...fkaListenerContainerFactoryConfigurer.java | 22 +++++++++---------- .../KafkaAnnotationDrivenConfiguration.java | 12 +++++----- .../kafka/KafkaAutoConfigurationTests.java | 3 +-- .../main/asciidoc/spring-boot-features.adoc | 5 +++-- 4 files changed, 21 insertions(+), 21 deletions(-) diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java index e5775b6b61..78d614f999 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java @@ -48,14 +48,14 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { private KafkaAwareTransactionManager transactionManager; + private ConsumerAwareRebalanceListener rebalanceListener; + private ErrorHandler errorHandler; private BatchErrorHandler batchErrorHandler; private AfterRollbackProcessor afterRollbackProcessor; - private ConsumerAwareRebalanceListener rebalanceListener; - /** * Set the {@link KafkaProperties} to use. * @param properties the properties @@ -89,6 +89,15 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { this.transactionManager = transactionManager; } + /** + * Set the {@link ConsumerAwareRebalanceListener} to use. + * @param rebalanceListener the rebalance listener. + * @since 2.2 + */ + void setRebalanceListener(ConsumerAwareRebalanceListener rebalanceListener) { + this.rebalanceListener = rebalanceListener; + } + /** * Set the {@link ErrorHandler} to use. * @param errorHandler the error handler @@ -114,15 +123,6 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { this.afterRollbackProcessor = afterRollbackProcessor; } - /** - * Set the {@link ConsumerAwareRebalanceListener} to use. - * @param rebalanceListener the rebalance listener. - * @since 2.2 - */ - void setRebalanceListener(ConsumerAwareRebalanceListener rebalanceListener) { - this.rebalanceListener = rebalanceListener; - } - /** * Configure the specified Kafka listener container factory. The factory can be * further tuned and default settings can be overridden. diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java index 5f8b009803..89dca80442 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java @@ -58,33 +58,33 @@ class KafkaAnnotationDrivenConfiguration { private final KafkaAwareTransactionManager transactionManager; + private final ConsumerAwareRebalanceListener rebalanceListener; + private final ErrorHandler errorHandler; private final BatchErrorHandler batchErrorHandler; private final AfterRollbackProcessor afterRollbackProcessor; - private final ConsumerAwareRebalanceListener rebalanceListener; - KafkaAnnotationDrivenConfiguration(KafkaProperties properties, ObjectProvider messageConverter, ObjectProvider batchMessageConverter, ObjectProvider> kafkaTemplate, ObjectProvider> kafkaTransactionManager, + ObjectProvider rebalanceListener, ObjectProvider errorHandler, ObjectProvider batchErrorHandler, - ObjectProvider> afterRollbackProcessor, - ObjectProvider rebalanceListener) { + ObjectProvider> afterRollbackProcessor) { this.properties = properties; this.messageConverter = messageConverter.getIfUnique(); this.batchMessageConverter = batchMessageConverter.getIfUnique( () -> new BatchMessagingMessageConverter(this.messageConverter)); this.kafkaTemplate = kafkaTemplate.getIfUnique(); this.transactionManager = kafkaTransactionManager.getIfUnique(); + this.rebalanceListener = rebalanceListener.getIfUnique(); this.errorHandler = errorHandler.getIfUnique(); this.batchErrorHandler = batchErrorHandler.getIfUnique(); this.afterRollbackProcessor = afterRollbackProcessor.getIfUnique(); - this.rebalanceListener = rebalanceListener.getIfUnique(); } @Bean @@ -97,10 +97,10 @@ class KafkaAnnotationDrivenConfiguration { configurer.setMessageConverter(messageConverterToUse); configurer.setReplyTemplate(this.kafkaTemplate); configurer.setTransactionManager(this.transactionManager); + configurer.setRebalanceListener(this.rebalanceListener); configurer.setErrorHandler(this.errorHandler); configurer.setBatchErrorHandler(this.batchErrorHandler); configurer.setAfterRollbackProcessor(this.afterRollbackProcessor); - configurer.setRebalanceListener(this.rebalanceListener); return configurer; } diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java index 585675c265..5c1c385904 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java @@ -767,8 +767,7 @@ public class KafkaAutoConfigurationTests { @Bean public ConsumerAwareRebalanceListener rebalanceListener() { - return new ConsumerAwareRebalanceListener() { - }; + return mock(ConsumerAwareRebalanceListener.class); } } diff --git a/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc b/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc index 15550b8e3c..728d4a4caa 100644 --- a/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc +++ b/spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc @@ -6158,8 +6158,9 @@ The following component creates a listener endpoint on the `someTopic` topic: ---- If a `KafkaTransactionManager` bean is defined, it is automatically associated to the -container factory. Similarly, if a `ErrorHandler` or `AfterRollbackProcessor` bean is -defined, it is automatically associated to the default factory. +container factory. Similarly, if a `ErrorHandler`, `AfterRollbackProcessor` or +`ConsumerAwareRebalanceListener` bean is defined, it is automatically associated to the +default factory. Depending on the listener type, a `RecordMessageConverter` or `BatchMessageConverter` bean is associated to the default factory. If only a `RecordMessageConverter` bean is present