From 53d32c2332983679eca0699f77468a76461c20fd Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 14 Sep 2021 10:17:32 -0400 Subject: [PATCH] GH-1140: CommonErrorHandler per consumer binding (#1143) * GH-1140: CommonErrorHandler per consumer binding Setting CommonErrorHandler on consumer binding through its bean name. If present, binder will resolve this bean and assign it on the listener container. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1140 * Addressing PR review --- docs/src/main/asciidoc/overview.adoc | 6 ++ .../properties/KafkaConsumerProperties.java | 13 ++++ .../kafka/KafkaMessageChannelBinder.java | 7 ++ .../stream/binder/kafka/KafkaBinderTests.java | 66 +++++++++++++++++++ 4 files changed, 92 insertions(+) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index b78fdbc0d..fc4217f07 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -321,6 +321,12 @@ When using a transactional binder, the offset of a recovered record (e.g. when r Setting this property to `false` suppresses committing the offset of recovered record. + Default: true. +commonErrorHandlerBeanName:: +`CommonErrorHandler` bean name to use per consumer binding. +When present, this user provided `CommonErrorHandler` takes precedence over any other error handlers defined by the binder. +This is a handy way to express error handlers, if the application does not want to use a `ListenerContainerCustomizer` and then check the destination/group combination to set an error handler. ++ +Default: none. [[reset-offsets]] ==== Resetting Offsets diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index ae9ef2826..0a4d2561d 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java @@ -210,6 +210,12 @@ public class KafkaConsumerProperties { */ private boolean txCommitRecovered = true; + /** + * CommonErrorHandler bean name per consumer binding. + * @since 3.2 + */ + private String commonErrorHandlerBeanName; + /** * @return if each record needs to be acknowledged. * @@ -529,4 +535,11 @@ public class KafkaConsumerProperties { this.txCommitRecovered = txCommitRecovered; } + public String getCommonErrorHandlerBeanName() { + return commonErrorHandlerBeanName; + } + + public void setCommonErrorHandlerBeanName(String commonErrorHandlerBeanName) { + this.commonErrorHandlerBeanName = commonErrorHandlerBeanName; + } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 3155f9543..4da85be76 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -102,6 +102,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.CommonErrorHandler; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.ConsumerAwareRebalanceListener; import org.springframework.kafka.listener.ConsumerProperties; @@ -777,6 +778,12 @@ public class KafkaMessageChannelBinder extends else { kafkaMessageDrivenChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel()); } + final String commonErrorHandlerBeanName = extendedConsumerProperties.getExtension().getCommonErrorHandlerBeanName(); + if (StringUtils.hasText(commonErrorHandlerBeanName)) { + final CommonErrorHandler commonErrorHandler = getApplicationContext().getBean(commonErrorHandlerBeanName, + CommonErrorHandler.class); + messageListenerContainer.setCommonErrorHandler(commonErrorHandler); + } this.getContainerCustomizer().configure(messageListenerContainer, destination.getName(), group); this.ackModeInfo.put(destination, messageListenerContainer.getContainerProperties().getAckMode()); return kafkaMessageDrivenChannelAdapter; diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index e6ec667ba..667aeb80b 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -118,8 +118,11 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.CommonErrorHandler; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.DefaultErrorHandler; +import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; @@ -145,6 +148,7 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.MimeTypeUtils; +import org.springframework.util.backoff.FixedBackOff; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture; @@ -1343,6 +1347,68 @@ public class KafkaBinderTests extends producerBinding.unbind(); } + @Test + @SuppressWarnings("unchecked") + public void testCommonErrorHandlerBeanNameOnConsumerBinding() throws Exception { + Binder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + BindingProperties producerBindingProperties = createProducerBindingProperties( + producerProperties); + + DirectChannel moduleOutputChannel = createBindableChannel("output", + producerBindingProperties); + + CountDownLatch latch = new CountDownLatch(1); + CommonErrorHandler commonErrorHandler = new DefaultErrorHandler(new FixedBackOff(0L, 0L)) { + @Override + public void handleRemaining(Exception thrownException, List> records, + Consumer consumer, MessageListenerContainer container) { + super.handleRemaining(thrownException, records, consumer, container); + latch.countDown(); + } + }; + + ConfigurableApplicationContext context = TestUtils.getPropertyValue(binder, + "binder.applicationContext", ConfigurableApplicationContext.class); + context.getBeanFactory().registerSingleton("fooCommonErrorHandler", commonErrorHandler); + + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setMaxAttempts(2); + consumerProperties.setBackOffInitialInterval(100); + consumerProperties.setBackOffMaxInterval(150); + consumerProperties.getExtension().setCommonErrorHandlerBeanName("fooCommonErrorHandler"); + + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler(); + moduleInputChannel.subscribe(handler); + long uniqueBindingId = System.currentTimeMillis(); + Binding producerBinding = binder.bindProducer( + "retryTest." + uniqueBindingId + ".0", moduleOutputChannel, + producerProperties); + Binding consumerBinding = binder.bindConsumer( + "retryTest." + uniqueBindingId + ".0", "testGroup", moduleInputChannel, + consumerProperties); + + String testMessagePayload = "test." + UUID.randomUUID(); + Message testMessage = MessageBuilder + .withPayload(testMessagePayload.getBytes()).build(); + moduleOutputChannel.send(testMessage); + + Thread.sleep(3000); + + //Assertions for the CommonErrorHandler configured on the consumer binding (commonErrorHandlerBeanName). + assertThat(KafkaTestUtils.getPropertyValue(consumerBinding, + "lifecycle.messageListenerContainer.commonErrorHandler")).isSameAs(commonErrorHandler); + latch.await(10, TimeUnit.SECONDS); + + binderBindUnbindLatency(); + consumerBinding.unbind(); + producerBinding.unbind(); + } + + //See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/870 for motivation for this test. @Test @SuppressWarnings("unchecked")