diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index 949c10bd16..448259ed9d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -63,6 +63,7 @@ import org.springframework.util.Assert; * * @author Gary Russell * @author Artem Bilan + * @author Urs Keller * * @since 3.0.2 * @@ -281,8 +282,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem } } else { - KafkaInboundGateway.this.logger.debug("Converter returned a null message for: " - + record); + KafkaInboundGateway.this.logger.debug("Converter returned a null message for: " + record); } } @@ -345,7 +345,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem @Override public boolean open(RetryContext context, RetryCallback callback) { - if (KafkaInboundGateway.this.recoveryCallback != null) { + if (KafkaInboundGateway.this.retryTemplate != null) { attributesHolder.set(context); } return true; @@ -354,6 +354,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem @Override public void close(RetryContext context, RetryCallback callback, Throwable throwable) { + attributesHolder.remove(); } 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 039d522124..d35da81c7d 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 @@ -69,6 +69,7 @@ import org.springframework.util.Assert; * @author Marius Bogoevici * @author Gary Russell * @author Artem Bilan + * @author Urs Keller * */ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport implements OrderlyShutdownCapable, @@ -472,7 +473,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo @Override public boolean open(RetryContext context, RetryCallback callback) { - if (KafkaMessageDrivenChannelAdapter.this.recoveryCallback != null) { + if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) { attributesHolder.set(context); } return true; @@ -481,6 +482,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo @Override public void close(RetryContext context, RetryCallback callback, Throwable throwable) { + attributesHolder.remove(); } @@ -527,7 +529,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo @Override public boolean open(RetryContext context, RetryCallback callback) { - if (KafkaMessageDrivenChannelAdapter.this.recoveryCallback != null) { + if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) { attributesHolder.set(context); } return true; @@ -536,6 +538,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo @Override public void close(RetryContext context, RetryCallback callback, Throwable throwable) { + attributesHolder.remove(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index 36d2e180c4..ff615a804f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -60,7 +60,10 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; import org.springframework.retry.backoff.NoBackOffPolicy; +import org.springframework.retry.listener.RetryListenerSupport; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; @@ -68,6 +71,7 @@ import org.springframework.retry.support.RetryTemplate; * * @author Gary Russell * @author Artem Bilan + * @author Urs Keller * * @since 3.0.2 * @@ -86,9 +90,11 @@ public class InboundGatewayTests { private static String topic6 = "testTopic6"; + private static String topic7 = "testTopic7"; + @ClassRule public static EmbeddedKafkaRule embeddedKafka = - new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); + new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6, topic7); @Rule public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace().categories("org.apache.kafka.clients", @@ -229,7 +235,7 @@ public class InboundGatewayTests { Message failed = ((MessagingException) em.getPayload()).getFailedMessage(); assertThat(failed).isNotNull(); assertThat(failed.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA)) - .isSameAs(em.getHeaders().get(KafkaHeaders.RAW_DATA)); + .isSameAs(em.getHeaders().get(KafkaHeaders.RAW_DATA)); MessageChannel reply = (MessageChannel) em.getHeaders().getReplyChannel(); MessageHeaders headers = failed.getHeaders(); reply.send(MessageBuilder.withPayload("ERROR").copyHeaders(headers).build()); @@ -335,4 +341,62 @@ public class InboundGatewayTests { gateway.stop(); } + @Test + public void testInboundRetryErrorRecoverWithoutRecocveryCallback() throws Exception { + EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); + Map consumerProps = KafkaTestUtils.consumerProps("replyHandler4", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer consumer = cf2.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic7); + + Map props = KafkaTestUtils.consumerProps("test4", "false", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + ContainerProperties containerProps = new ContainerProperties(topic7); + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic7); + KafkaInboundGateway gateway = new KafkaInboundGateway<>(container, template); + MessageChannel out = new DirectChannel() { + + @Override + protected boolean doSend(Message message, long timeout) { + throw new RuntimeException("intended"); + } + + }; + gateway.setRequestChannel(out); + gateway.setBeanFactory(mock(BeanFactory.class)); + gateway.setReplyTimeout(30_000); + RetryTemplate retryTemplate = new RetryTemplate(); + SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(); + retryPolicy.setMaxAttempts(5); + retryTemplate.setRetryPolicy(retryPolicy); + retryTemplate.setBackOffPolicy(new NoBackOffPolicy()); + final CountDownLatch retryCountLatch = new CountDownLatch(retryPolicy.getMaxAttempts()); + retryTemplate.registerListener(new RetryListenerSupport() { + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + retryCountLatch.countDown(); + } + }); + gateway.setRetryTemplate(retryTemplate); + gateway.afterPropertiesSet(); + gateway.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + template.sendDefault(0, 1487694048607L, 1, "foo"); + + assertThat(retryCountLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + gateway.stop(); + consumer.close(); + } + } 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 9d3af6c38e..1181723efa 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 @@ -85,6 +85,9 @@ 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.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.listener.RetryListenerSupport; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; @@ -94,6 +97,7 @@ import org.springframework.retry.support.RetryTemplate; * @author Artem Bilan * @author Biju Kunjummen * @author Cameron Mayfield + * @author Urs Keller * * @since 2.0 * @@ -114,7 +118,7 @@ public class MessageDrivenAdapterTests { @ClassRule public static EmbeddedKafkaRule embeddedKafkaRule = - new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); + new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); @@ -256,6 +260,61 @@ public class MessageDrivenAdapterTests { adapter.stop(); } + + /** + * the recovery callback is not mandatory, if not set and retries are exhausted the last throwable is rethrown + * to the consumer. + */ + @Test + public void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception { + Map props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + ContainerProperties containerProps = new ContainerProperties(topic6); + 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(2); + retryTemplate.setRetryPolicy(retryPolicy); + final CountDownLatch retryCountLatch = new CountDownLatch(retryPolicy.getMaxAttempts()); + retryTemplate.registerListener(new RetryListenerSupport() { + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + retryCountLatch.countDown(); + } + }); + adapter.setRetryTemplate(retryTemplate); + + adapter.afterPropertiesSet(); + adapter.start(); + ContainerTestUtils.waitForAssignment(container, 2); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic6); + template.sendDefault(1, "foo"); + + assertThat(retryCountLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + adapter.stop(); + pf.destroy(); + } + @Test public void testInboundRecordNoRetryRecover() { Map props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka); @@ -294,11 +353,11 @@ public class MessageDrivenAdapterTests { MessageHeaders headers = received.getHeaders(); assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNotNull(); assertThat(headers.get(IntegrationMessageHeaderAccessor.SOURCE_DATA)) - .isSameAs(headers.get(KafkaHeaders.RAW_DATA)); + .isSameAs(headers.get(KafkaHeaders.RAW_DATA)); Message originalMessage = ((ErrorMessage) received).getOriginalMessage(); assertThat(originalMessage).isNotNull(); assertThat(originalMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA)) - .isSameAs(headers.get(KafkaHeaders.RAW_DATA)); + .isSameAs(headers.get(KafkaHeaders.RAW_DATA)); headers = originalMessage.getHeaders(); assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1); assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic5);