From c219a43c585c29c9771bcf34c0b4cf128a7bd875 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 17 Apr 2017 17:28:17 -0400 Subject: [PATCH] SCSt-913: Add Support for Enhanced Error Handling - bind input message, raw record to retry context - support recover without retry (recover after a single shot) Change RecoveryCallback from to Actually for backwards compatibility. Changes needed for the latest updates to Spring Integration Now that the 4.3.10.BS artifacts are in artifactory. Suppress Deprecation Warnings checkstyle and PR Comments Javadocs and TL.remove() Fix Attributes Logic Polishing - PR Comments - Enhanced error message for conversion errors. Make errorChannel and retryTemplate mutually exclusive Fix TL.remove with retry. * Polishing `build.gradle` before merging Fix TL.remove with retry. Remove mavenLocal() Conflicts: build.gradle src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java * Use `KafkaHeaders.RAW_DATA` * Use `ErrorMesage` directly * Adapt DSL test to use `ErrorMessageSendingRecoverer` --- .../KafkaMessageDrivenChannelAdapterSpec.java | 2 +- .../KafkaMessageDrivenChannelAdapter.java | 133 ++++++++++++++++-- .../kafka/outbound/package-info.java | 4 + .../RawRecordHeaderErrorMessageStrategy.java | 50 +++++++ .../kafka/support/package-info.java | 4 + .../integration/kafka/dsl/KafkaDslTests.java | 8 +- .../inbound/MessageDrivenAdapterTests.java | 111 ++++++++++++++- 7 files changed, 297 insertions(+), 15 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/package-info.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java index bfca39e845..a022441622 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java @@ -139,7 +139,7 @@ public class KafkaMessageDrivenChannelAdapterSpec recoveryCallback) { + public S recoveryCallback(RecoveryCallback recoveryCallback) { this.target.setRecoveryCallback(recoveryCallback); return _this(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 1dbd4d23cf..948aa13531 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -21,8 +21,12 @@ import java.util.List; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.springframework.core.AttributeAccessor; import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.BatchMessageListener; import org.springframework.kafka.listener.MessageListener; @@ -33,6 +37,7 @@ import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.BatchMessageConverter; import org.springframework.kafka.support.converter.ConversionException; import org.springframework.kafka.support.converter.MessageConverter; @@ -40,6 +45,9 @@ import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.messaging.Message; import org.springframework.messaging.support.ErrorMessage; import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryListener; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; @@ -56,11 +64,13 @@ import org.springframework.util.Assert; */ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport implements OrderlyShutdownCapable { + private static final ThreadLocal attributesHolder = new ThreadLocal<>(); + private final AbstractMessageListenerContainer messageListenerContainer; - private final RecordMessagingMessageListenerAdapter recordListener = new IntegrationRecordMessageListener(); + private final IntegrationRecordMessageListener recordListener = new IntegrationRecordMessageListener(); - private final BatchMessagingMessageListenerAdapter batchListener = new IntegrationBatchMessageListener(); + private final IntegrationBatchMessageListener batchListener = new IntegrationBatchMessageListener(); private final ListenerMode mode; @@ -96,6 +106,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo this.messageListenerContainer = messageListenerContainer; this.messageListenerContainer.setAutoStartup(false); this.mode = mode; + setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy()); } /** @@ -172,7 +183,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo /** * A {@link RecoveryCallback} instance for retry operation; - * if null, the exception will be thrown to the container after retries are exhausted. + * if null, the exception will be thrown to the container after retries are exhausted + * (unless an error channel is configured). * Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified. * @param recoveryCallback the recovery callback. * @since 2.0.1 @@ -211,6 +223,11 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo protected void onInit() { super.onInit(); + if (this.retryTemplate != null) { + Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is " + + "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to " + + "send an error message when retries are exhausted"); + } if (this.mode.equals(ListenerMode.record)) { MessageListener listener = this.recordListener; @@ -222,11 +239,13 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo this.ackDiscarded); listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate, this.recoveryCallback); + this.retryTemplate.registerListener(this.recordListener); } else { if (this.retryTemplate != null) { listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate, this.recoveryCallback); + this.retryTemplate.registerListener(this.recordListener); } if (this.recordFilterStrategy != null) { listener = new FilteringMessageListenerAdapter<>(listener, this.recordFilterStrategy, @@ -272,6 +291,42 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo return getPhase(); } + /** + * If there's a retry template, it will set the attributes holder via the listener. If + * there's no retry template, but there's an error channel, we create a new attributes + * holder here. If an attributes holder exists (by either method), we set the + * attributes for use by the {@link ErrorMessageStrategy}. + * @param record the record. + * @param message the message. + * @since 2.1.1 + */ + private void setAttributesIfNecessary(Object record, Message message) { + boolean needHolder = getErrorChannel() != null + && KafkaMessageDrivenChannelAdapter.this.retryTemplate == null; + boolean needAttributes = needHolder | KafkaMessageDrivenChannelAdapter.this.retryTemplate != null; + if (needHolder) { + attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null)); + } + if (needAttributes) { + AttributeAccessor attributes = attributesHolder.get(); + if (attributes != null) { + attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message); + attributes.setAttribute(KafkaHeaders.RAW_DATA, record); + } + } + } + + @Override + protected AttributeAccessor getErrorMessageAttributes(Message message) { + AttributeAccessor attributes = attributesHolder.get(); + if (attributes == null) { + return super.getErrorMessageAttributes(message); + } + else { + return attributes; + } + } + /** * The listener mode for the container, record or batch. * @since 1.2 @@ -291,7 +346,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo batch } - private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter { + private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter + implements RetryListener { IntegrationRecordMessageListener() { super(null, null); @@ -302,15 +358,21 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo Message message = null; try { message = toMessagingMessage(record, acknowledgment, consumer); + setAttributesIfNecessary(record, message); } catch (RuntimeException e) { - Exception exception = new ConversionException("Failed to convert to message for: " + record, e); - if (getErrorChannel() != null) { - getMessagingTemplate().send(getErrorChannel(), new ErrorMessage(exception)); - } + RuntimeException exception = new ConversionException("Failed to convert to message for: " + record, e); + sendErrorMessageIfNecessary(null, exception); } if (message != null) { - sendMessage(message); + try { + sendMessage(message); + } + finally { + if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) { + attributesHolder.remove(); + } + } } else { KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: " @@ -318,9 +380,30 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } } + @Override + public boolean open(RetryContext context, RetryCallback callback) { + if (KafkaMessageDrivenChannelAdapter.this.recoveryCallback != null) { + attributesHolder.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + attributesHolder.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + } - private class IntegrationBatchMessageListener extends BatchMessagingMessageListenerAdapter { + private class IntegrationBatchMessageListener extends BatchMessagingMessageListenerAdapter + implements RetryListener { IntegrationBatchMessageListener() { super(null, null); @@ -332,6 +415,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo Message message = null; try { message = toMessagingMessage(records, acknowledgment, consumer); + setAttributesIfNecessary(records, message); } catch (RuntimeException e) { Exception exception = new ConversionException("Failed to convert to message for: " + records, e); @@ -340,7 +424,14 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } } if (message != null) { - sendMessage(message); + try { + sendMessage(message); + } + finally { + if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) { + attributesHolder.remove(); + } + } } else { KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: " @@ -348,6 +439,26 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } } + @Override + public boolean open(RetryContext context, RetryCallback callback) { + if (KafkaMessageDrivenChannelAdapter.this.recoveryCallback != null) { + attributesHolder.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + attributesHolder.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + } } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java new file mode 100644 index 0000000000..d5f6b0cd54 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides outbound Spring Integration Kafka components. + */ +package org.springframework.integration.kafka.outbound; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java new file mode 100644 index 0000000000..e90727d0f0 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java @@ -0,0 +1,50 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.kafka.support; + +import java.util.Collections; +import java.util.Map; + +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; + +/** + * {@link ErrorMessageStrategy} extension that adds the raw record as + * a header to the {@link ErrorMessage}. + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 2.1.1 + * + */ +public class RawRecordHeaderErrorMessageStrategy implements ErrorMessageStrategy { + + @Override + public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor context) { + Object inputMessage = context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY); + Map headers = + Collections.singletonMap(KafkaHeaders.RAW_DATA, context.getAttribute(KafkaHeaders.RAW_DATA)); + return new ErrorMessage(throwable, headers, + inputMessage instanceof Message ? (Message) inputMessage : null); + } + +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/package-info.java new file mode 100644 index 0000000000..0b8fe48568 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides support classes. + */ +package org.springframework.integration.kafka.support; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 19ce7037c4..6c73076d0f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -38,8 +38,10 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @@ -196,7 +198,8 @@ public class KafkaDslTests { .configureListenerContainer(c -> c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL) .id("topic1ListenerContainer")) - .errorChannel("errorChannel") + .recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(), + new RawRecordHeaderErrorMessageStrategy())) .retryTemplate(new RetryTemplate()) .filterInRetry(true)) .filter(Message.class, m -> @@ -214,7 +217,8 @@ public class KafkaDslTests { KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2) .configureListenerContainer(c -> c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)) - .errorChannel("errorChannel") + .recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(), + new RawRecordHeaderErrorMessageStrategy())) .retryTemplate(new RetryTemplate()) .filterInRetry(true)) .filter(Message.class, m -> diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index b1046254a4..a4638cbd44 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -30,8 +30,11 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.junit.ClassRule; import org.junit.Test; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.ListenerMode; +import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -52,8 +55,12 @@ import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.ContainerTestUtils; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; /** * @@ -72,8 +79,12 @@ public class MessageDrivenAdapterTests { private static String topic3 = "testTopic3"; + private static String topic4 = "testTopic4"; + + private static String topic5 = "testTopic5"; + @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5); @Test public void testInboundRecord() throws Exception { @@ -160,6 +171,104 @@ public class MessageDrivenAdapterTests { adapter.stop(); } + @Test + public void testInboundRecordRetryRecover() throws Exception { + Map props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + ContainerProperties containerProps = new ContainerProperties(topic4); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + KafkaMessageDrivenChannelAdapter adapter = new KafkaMessageDrivenChannelAdapter<>(container); + MessageChannel out = new DirectChannel() { + + @Override + protected boolean doSend(Message message, long timeout) { + throw new RuntimeException("intended"); + } + + }; + adapter.setOutputChannel(out); + RetryTemplate retryTemplate = new RetryTemplate(); + SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(); + retryPolicy.setMaxAttempts(1); + retryTemplate.setRetryPolicy(retryPolicy); + QueueChannel errorChannel = new QueueChannel(); + adapter.setRecoveryCallback( + new ErrorMessageSendingRecoverer(errorChannel, new RawRecordHeaderErrorMessageStrategy())); + adapter.setRetryTemplate(retryTemplate); + adapter.afterPropertiesSet(); + adapter.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic4); + template.sendDefault(1, "foo"); + + Message received = errorChannel.receive(10000); + assertThat(received).isInstanceOf(ErrorMessage.class); + MessageHeaders headers = received.getHeaders(); + assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNotNull(); + received = ((ErrorMessage) received).getOriginalMessage(); + assertThat(received).isNotNull(); + headers = received.getHeaders(); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic4); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + + adapter.stop(); + } + + @Test + public void testInboundRecordNoRetryRecover() throws Exception { + Map props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + ContainerProperties containerProps = new ContainerProperties(topic5); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + KafkaMessageDrivenChannelAdapter adapter = new KafkaMessageDrivenChannelAdapter<>(container); + MessageChannel out = new DirectChannel() { + + @Override + protected boolean doSend(Message message, long timeout) { + throw new RuntimeException("intended"); + } + + }; + adapter.setOutputChannel(out); + QueueChannel errorChannel = new QueueChannel(); + adapter.setErrorChannel(errorChannel); + adapter.setRecoveryCallback( + new ErrorMessageSendingRecoverer(errorChannel, new RawRecordHeaderErrorMessageStrategy())); + adapter.afterPropertiesSet(); + adapter.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic5); + template.sendDefault(1, "foo"); + + Message received = errorChannel.receive(10000); + assertThat(received).isInstanceOf(ErrorMessage.class); + MessageHeaders headers = received.getHeaders(); + assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNotNull(); + received = ((ErrorMessage) received).getOriginalMessage(); + assertThat(received).isNotNull(); + headers = received.getHeaders(); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic5); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + + adapter.stop(); + } + @Test public void testInboundBatch() throws Exception { Map props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka);