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 66594904e8..0fbc9b52a7 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 @@ -24,6 +24,7 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; 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.ContainerProperties; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.support.converter.MessageConverter; @@ -48,6 +49,8 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { private ErrorHandler errorHandler; + private BatchErrorHandler batchErrorHandler; + private AfterRollbackProcessor afterRollbackProcessor; /** @@ -91,6 +94,14 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { this.errorHandler = errorHandler; } + /** + * Set the {@link BatchErrorHandler} to use. + * @param batchErrorHandler the error handler + */ + public void setBatchErrorHandler(BatchErrorHandler batchErrorHandler) { + this.batchErrorHandler = batchErrorHandler; + } + /** * Set the {@link AfterRollbackProcessor} to use. * @param afterRollbackProcessor the after rollback processor @@ -124,8 +135,13 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer { map.from(this.replyTemplate).to(factory::setReplyTemplate); map.from(properties::getType).whenEqualTo(Listener.Type.BATCH) .toCall(() -> factory.setBatchListener(true)); - map.from(this.errorHandler).to(factory::setErrorHandler); map.from(this.afterRollbackProcessor).to(factory::setAfterRollbackProcessor); + if (properties.getType().equals(Listener.Type.BATCH)) { + factory.setBatchErrorHandler(this.batchErrorHandler); + } + else { + factory.setErrorHandler(this.errorHandler); + } } private void configureContainer(ContainerProperties container) { 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 f2b84a1674..34e78548c9 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 @@ -28,6 +28,7 @@ import org.springframework.kafka.config.KafkaListenerConfigUtils; 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.ErrorHandler; import org.springframework.kafka.support.converter.BatchMessageConverter; import org.springframework.kafka.support.converter.BatchMessagingMessageConverter; @@ -58,6 +59,8 @@ class KafkaAnnotationDrivenConfiguration { private final ErrorHandler errorHandler; + private final BatchErrorHandler batchErrorHandler; + private final AfterRollbackProcessor afterRollbackProcessor; KafkaAnnotationDrivenConfiguration(KafkaProperties properties, @@ -66,6 +69,7 @@ class KafkaAnnotationDrivenConfiguration { ObjectProvider> kafkaTemplate, ObjectProvider> kafkaTransactionManager, ObjectProvider errorHandler, + ObjectProvider batchErrorHandler, ObjectProvider> afterRollbackProcessor) { this.properties = properties; this.messageConverter = messageConverter.getIfUnique(); @@ -74,6 +78,7 @@ class KafkaAnnotationDrivenConfiguration { this.kafkaTemplate = kafkaTemplate.getIfUnique(); this.transactionManager = kafkaTransactionManager.getIfUnique(); this.errorHandler = errorHandler.getIfUnique(); + this.batchErrorHandler = batchErrorHandler.getIfUnique(); this.afterRollbackProcessor = afterRollbackProcessor.getIfUnique(); } @@ -88,6 +93,7 @@ class KafkaAnnotationDrivenConfiguration { configurer.setReplyTemplate(this.kafkaTemplate); configurer.setTransactionManager(this.transactionManager); configurer.setErrorHandler(this.errorHandler); + configurer.setBatchErrorHandler(this.batchErrorHandler); configurer.setAfterRollbackProcessor(this.afterRollbackProcessor); 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 48e5856061..a783cf1d45 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.KafkaTemplate; import org.springframework.kafka.listener.AfterRollbackProcessor; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.ContainerProperties.AckMode; +import org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler; import org.springframework.kafka.listener.SeekToCurrentErrorHandler; import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer; import org.springframework.kafka.support.converter.BatchMessageConverter; @@ -581,6 +582,39 @@ public class KafkaAutoConfigurationTests { }); } + @Test + public void concurrentKafkaListenerContainerFactoryInBatchModeShouldUseBatchErrorHandler() { + this.contextRunner.withUserConfiguration(BatchErrorHandlerConfiguration.class) + .withPropertyValues("spring.kafka.listener.type=batch").run((context) -> { + ConcurrentKafkaListenerContainerFactory factory = context + .getBean(ConcurrentKafkaListenerContainerFactory.class); + assertThat(KafkaTestUtils.getPropertyValue(factory, "errorHandler")) + .isSameAs(context.getBean("batchErrorHandler")); + }); + } + + @Test + public void concurrentKafkaListenerContainerFactoryInBatchModeWhenBatchErrorHandlerNotAvailableShouldBeNull() { + this.contextRunner.withPropertyValues("spring.kafka.listener.type=batch") + .run((context) -> { + ConcurrentKafkaListenerContainerFactory factory = context + .getBean(ConcurrentKafkaListenerContainerFactory.class); + assertThat(KafkaTestUtils.getPropertyValue(factory, "errorHandler")) + .isNull(); + }); + } + + @Test + public void concurrentKafkaListenerContainerFactoryInBatchModeAndSimpleErrorHandlerShouldBeNull() { + this.contextRunner.withPropertyValues("spring.kafka.listener.type=batch") + .withUserConfiguration(ErrorHandlerConfiguration.class).run((context) -> { + ConcurrentKafkaListenerContainerFactory factory = context + .getBean(ConcurrentKafkaListenerContainerFactory.class); + assertThat(KafkaTestUtils.getPropertyValue(factory, "errorHandler")) + .isNull(); + }); + } + @Test public void testConcurrentKafkaListenerContainerFactoryWithDefaultTransactionManager() { this.contextRunner @@ -660,6 +694,16 @@ public class KafkaAutoConfigurationTests { } + @Configuration(proxyBeanMethods = false) + protected static class BatchErrorHandlerConfiguration { + + @Bean + public SeekToCurrentBatchErrorHandler batchErrorHandler() { + return new SeekToCurrentBatchErrorHandler(); + } + + } + @Configuration(proxyBeanMethods = false) protected static class TransactionManagerConfiguration {