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);