diff --git a/pom.xml b/pom.xml index c9c7fef8e..fac7b912d 100644 --- a/pom.xml +++ b/pom.xml @@ -14,7 +14,7 @@ 1.7 0.10.1.1 1.1.6.RELEASE - 2.1.1.RELEASE + 2.1.2.BUILD-SNAPSHOT 1.3.0.BUILD-SNAPSHOT 1.3.4.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka-0.10.1-test/pom.xml b/spring-cloud-stream-binder-kafka-0.10.1-test/pom.xml index 4692e3887..1121b71a4 100644 --- a/spring-cloud-stream-binder-kafka-0.10.1-test/pom.xml +++ b/spring-cloud-stream-binder-kafka-0.10.1-test/pom.xml @@ -59,6 +59,17 @@ org.springframework.integration spring-integration-kafka + 2.1.2.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-core + 4.3.12.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-jmx + 4.3.12.BUILD-SNAPSHOT org.springframework.cloud diff --git a/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml b/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml index 474a94d07..6190d9b72 100644 --- a/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml +++ b/spring-cloud-stream-binder-kafka-0.10.2-test/pom.xml @@ -60,6 +60,17 @@ org.springframework.integration spring-integration-kafka + 2.1.2.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-core + 4.3.12.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-jmx + 4.3.12.BUILD-SNAPSHOT org.springframework.cloud diff --git a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java b/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java index b945fb5d7..3ff966c75 100644 --- a/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java +++ b/spring-cloud-stream-binder-kafka-0.10.2-test/src/test/java/org/springframework/cloud/stream/binder/kafka/Kafka10TestBinder.java @@ -67,6 +67,7 @@ public class Kafka10TestBinder extends AbstractKafkaTestBinder { ProducerListener producerListener = new LoggingProducerListener(); binder.setProducerListener(producerListener); AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class); + setApplicationContext(context); binder.setApplicationContext(context); binder.afterPropertiesSet(); this.setBinder(binder); diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc index a58d4b502..6d41943fc 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/index.adoc @@ -23,12 +23,15 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat = Reference Guide include::overview.adoc[] + include::dlq.adoc[] + include::metrics.adoc[] = Appendices [appendix] include::building.adoc[] + include::contributing.adoc[] // ====================================================================================== diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index c66a7eb45..7acec12f4 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -523,5 +523,15 @@ spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.c spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde ---- +[[kafka-error-channels]] +== Error Channels +Starting with _version 1.3_, the binder unconditionally sends exceptions to an error channel for each consumer destination, and can be configured to send async producer send failures to an error channel too. +See <> for more information. +The payload of the `ErrorMessage` for a send failure is a `KafkaSendFailureException` with properties: + +* `failedMessage` - the spring-messaging `Message` that failed to be sent. +* `record` - the raw `ProducerRecord` that was created from the `failedMessage` + +There is no automatic handling of these exceptions (such as sending to a <>); you can consume these exceptions with your own Spring Integration flow. diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index 14fb7e7f7..ae032e303 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -54,10 +54,23 @@ spring-kafka ${spring-kafka.version} + org.springframework.integration - spring-integration-kafka + spring-integration-core + 4.3.12.BUILD-SNAPSHOT + + org.springframework.integration + spring-integration-jmx + 4.3.12.BUILD-SNAPSHOT + + + org.springframework.integration + spring-integration-test + 4.3.12.BUILD-SNAPSHOT + + org.springframework.boot spring-boot-test 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 8dff0370a..94b9faeac 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 @@ -147,7 +147,8 @@ public class KafkaMessageChannelBinder extends @Override protected MessageHandler createProducerMessageHandler(final ProducerDestination destination, - ExtendedProducerProperties producerProperties) throws Exception { + ExtendedProducerProperties producerProperties, MessageChannel errorChannel) + throws Exception { final DefaultKafkaProducerFactory producerFB = getProducerFactory(producerProperties); Collection partitions = provisioningProvider.getPartitionsForTopic( producerProperties.getPartitionCount(), @@ -171,8 +172,12 @@ public class KafkaMessageChannelBinder extends if (this.producerListener != null) { kafkaTemplate.setProducerListener(this.producerListener); } - return new ProducerConfigurationMessageHandler(kafkaTemplate, destination.getName(), producerProperties, - producerFB); + ProducerConfigurationMessageHandler handler = new ProducerConfigurationMessageHandler(kafkaTemplate, + destination.getName(), producerProperties, producerFB); + if (errorChannel != null) { + handler.setSendFailureChannel(errorChannel); + } + return handler; } private DefaultKafkaProducerFactory getProducerFactory( @@ -413,7 +418,7 @@ public class KafkaMessageChannelBinder extends private final DefaultKafkaProducerFactory producerFactory; - private ProducerConfigurationMessageHandler(KafkaTemplate kafkaTemplate, String topic, + ProducerConfigurationMessageHandler(KafkaTemplate kafkaTemplate, String topic, ExtendedProducerProperties producerProperties, DefaultKafkaProducerFactory producerFactory) { super(kafkaTemplate); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java index 36a287514..9a9fa6373 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AbstractKafkaTestBinder.java @@ -22,12 +22,10 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; -import org.springframework.context.support.GenericApplicationContext; -import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.context.ApplicationContext; import org.springframework.integration.codec.Codec; import org.springframework.integration.codec.kryo.KryoRegistrar; import org.springframework.integration.codec.kryo.PojoCodec; -import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.tuple.TupleKryoRegistrar; import com.esotericsoftware.kryo.Kryo; @@ -40,15 +38,19 @@ import com.esotericsoftware.kryo.Registration; public abstract class AbstractKafkaTestBinder extends AbstractTestBinder, ExtendedProducerProperties> { + private ApplicationContext applicationContext; + @Override public void cleanup() { // do nothing - the rule will take care of that } - protected void addErrorChannel(GenericApplicationContext context) { - PublishSubscribeChannel errorChannel = new PublishSubscribeChannel(); - context.getBeanFactory().initializeBean(errorChannel, IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); - context.getBeanFactory().registerSingleton(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, errorChannel); + protected final void setApplicationContext(ApplicationContext context) { + this.applicationContext = context; + } + + public ApplicationContext getApplicationContext() { + return this.applicationContext; } protected static Codec getCodec() { 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 f27b5f751..5071424ff 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 @@ -58,27 +58,37 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.integration.kafka.support.KafkaSendFailureException; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.support.SendResult; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.retry.backoff.FixedBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.SettableListenableFuture; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; import kafka.utils.ZKStringSerializer$; import kafka.utils.ZkUtils; @@ -138,8 +148,7 @@ public abstract class KafkaBinderTests extends testDlqGuts(false); } - @SuppressWarnings({ "unchecked", "rawtypes" }) - public void testDlqGuts(boolean withRetry) throws Exception { + private void testDlqGuts(boolean withRetry) throws Exception { AbstractKafkaTestBinder binder = getBinder(); DirectChannel moduleOutputChannel = new DirectChannel(); DirectChannel moduleInputChannel = new DirectChannel(); @@ -167,7 +176,6 @@ public abstract class KafkaBinderTests extends ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(), "applicationContext", ApplicationContext.class); - Map beansOfType = context.getBeansOfType(MessageChannel.class); SubscribableChannel boundErrorChannel = context .getBean(producerName + ".testGroup.errors-0", SubscribableChannel.class); SubscribableChannel globalErrorChannel = context.getBean("errorChannel", SubscribableChannel.class); @@ -1722,6 +1730,68 @@ public abstract class KafkaBinderTests extends assertThat(extractEndpoint(producerBinding).isRunning()).isFalse(); } + @SuppressWarnings({ "rawtypes", "unchecked" }) + @Test + public void testProducerErrorChannel() throws Exception { + AbstractKafkaTestBinder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties()); + ExtendedProducerProperties producerProps = new ExtendedProducerProperties<>( + new KafkaProducerProperties()); + producerProps.setHeaderMode(HeaderMode.raw); + producerProps.setErrorChannelEnabled(true); + Binding producerBinding = binder.bindProducer("ec.0", moduleOutputChannel, producerProps); + final Message message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar") + .build(); + SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors", SubscribableChannel.class); + final AtomicReference> errorMessage = new AtomicReference<>(); + final CountDownLatch latch = new CountDownLatch(2); + ec.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + errorMessage.set(message); + latch.countDown(); + } + + }); + SubscribableChannel globalEc = binder.getApplicationContext() + .getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class); + globalEc.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + latch.countDown(); + } + + }); + KafkaProducerMessageHandler endpoint = TestUtils.getPropertyValue(producerBinding, "lifecycle", + KafkaProducerMessageHandler.class); + final RuntimeException fooException = new RuntimeException("foo"); + final AtomicReference sent = new AtomicReference<>(); + new DirectFieldAccessor(endpoint).setPropertyValue("kafkaTemplate", + new KafkaTemplate(mock(ProducerFactory.class)) { + + @Override + public ListenableFuture send(String topic, Object payload) { + sent.set(payload); + SettableListenableFuture future = new SettableListenableFuture<>(); + future.setException(fooException); + return future; + } + + }); + + moduleOutputChannel.send(message); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class); + assertThat(errorMessage.get().getPayload()).isInstanceOf(KafkaSendFailureException.class); + KafkaSendFailureException exception = (KafkaSendFailureException) errorMessage.get().getPayload(); + assertThat(exception.getCause()).isSameAs(fooException); + assertThat(new String(((byte[] )exception.getFailedMessage().getPayload()))).isEqualTo(message.getPayload()); + assertThat(exception.getRecord().value()).isSameAs(sent.get()); + producerBinding.unbind(); + } + @Override protected void binderBindUnbindLatency() throws InterruptedException { Thread.sleep(500);