diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java index 0db6167370..a8ac83b3ff 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-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. @@ -54,6 +54,9 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback"); return builder.getBeanDefinition(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java index b3638eac92..b59a0c8e68 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java @@ -40,50 +40,55 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd @Override protected AbstractBeanDefinition parseConsumer(final Element element, final ParserContext parserContext) { - final BeanDefinitionBuilder kafkaProducerMessageHandlerBuilder = + final BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaProducerMessageHandler.class); final String kafkaTemplateBeanName = element.getAttribute("kafka-template"); - kafkaProducerMessageHandlerBuilder.addConstructorArgReference(kafkaTemplateBeanName); + builder.addConstructorArgReference(kafkaTemplateBeanName); BeanDefinition topicExpressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic", "topic-expression", parserContext, element, false); if (topicExpressionDef != null) { - kafkaProducerMessageHandlerBuilder.addPropertyValue("topicExpression", topicExpressionDef); + builder.addPropertyValue("topicExpression", topicExpressionDef); } BeanDefinition messageKeyExpressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("message-key", "message-key-expression", parserContext, element, false); if (messageKeyExpressionDef != null) { - kafkaProducerMessageHandlerBuilder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef); + builder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef); } BeanDefinition partitionIdExpressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("partition-id", "partition-id-expression", parserContext, element, false); if (partitionIdExpressionDef != null) { - kafkaProducerMessageHandlerBuilder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef); + builder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef); } - IntegrationNamespaceUtils.setValueIfAttributeDefined(kafkaProducerMessageHandlerBuilder, element, "sync"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "sync"); BeanDefinition sendTimeoutExpressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("send-timeout", "send-timeout-expression", parserContext, element, false); if (sendTimeoutExpressionDef != null) { - kafkaProducerMessageHandlerBuilder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef); + builder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef); } BeanDefinition timestampExpressionDef = IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("timestamp-expression", element); if (timestampExpressionDef != null) { - kafkaProducerMessageHandlerBuilder.addPropertyValue("timestampExpression", timestampExpressionDef); + builder.addPropertyValue("timestampExpression", timestampExpressionDef); } - return kafkaProducerMessageHandlerBuilder.getBeanDefinition(); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-failure-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel", + "outputChannel"); + + return builder.getBeanDefinition(); } } diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd index 8116abc983..a50008d7f6 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd @@ -149,6 +149,33 @@ + + + + + + + + + + + + + + + + + + + + + @@ -226,6 +253,35 @@ + + + + A retry template for retrying deliveries; an 'error-channel' is not allowed + when a retry template is provided; configure a 'recovery-callback' such as an + 'ErrorMessageSendingRecoverer' when using a retry template. + + + + + + + + + + + + Used in conjunction with a 'retry-template'; in most cases this will be + an 'ErrorMessageSendingRecoverer'. Omitting this element will cause an + exception to be thrown to the listener container after retries are exhausted. + + + + + + + + + @@ -236,4 +292,19 @@ + + + + + + + + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml index 101789a50d..460e80dc53 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml @@ -22,7 +22,9 @@ channel="nullChannel" message-converter="messageConverter" payload-type="java.lang.String" - error-channel="errorChannel" /> + error-message-strategy="ems" + retry-template="retryTemplate" + recovery-callback="recoveryCallback" /> + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java index 0f15b47764..020f1cb53b 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java @@ -30,6 +30,7 @@ import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.ListenerMode; +import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.integration.test.util.TestUtils; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.listener.KafkaMessageListenerContainer; @@ -37,6 +38,7 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; import org.springframework.kafka.listener.config.ContainerProperties; +import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetryTemplate; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -61,13 +63,21 @@ public class KafkaMessageDrivenChannelAdapterParserTests { @Autowired private KafkaMessageDrivenChannelAdapter kafkaBatchListener; + @Autowired + private ErrorMessageStrategy ems; + + @Autowired + private RetryTemplate retryTemplate; + + @Autowired + private RecoveryCallback recoveryCallback; + @Test public void testKafkaMessageDrivenChannelAdapterParser() throws Exception { assertThat(this.kafkaListener.isAutoStartup()).isFalse(); assertThat(this.kafkaListener.isRunning()).isFalse(); assertThat(this.kafkaListener.getPhase()).isEqualTo(100); assertThat(TestUtils.getPropertyValue(this.kafkaListener, "outputChannel")).isSameAs(this.nullChannel); - assertThat(TestUtils.getPropertyValue(this.kafkaListener, "errorChannel")).isSameAs(this.errorChannel); KafkaMessageListenerContainer container = TestUtils.getPropertyValue(this.kafkaListener, "messageListenerContainer", KafkaMessageListenerContainer.class); @@ -78,6 +88,9 @@ public class KafkaMessageDrivenChannelAdapterParserTests { .isEqualTo(String.class); assertThat(TestUtils.getPropertyValue(this.kafkaListener, "batchListener.fallbackType")) .isEqualTo(String.class); + assertThat(TestUtils.getPropertyValue(this.kafkaListener, "errorMessageStrategy")).isSameAs(this.ems); + assertThat(TestUtils.getPropertyValue(this.kafkaListener, "retryTemplate")).isSameAs(this.retryTemplate); + assertThat(TestUtils.getPropertyValue(this.kafkaListener, "recoveryCallback")).isSameAs(this.recoveryCallback); } @Test diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml index 6f6b0c2319..42c07fffb9 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml @@ -20,6 +20,9 @@ partition-id-expression="'2'" send-timeout-expression="1000" timestamp-expression="T(System).currentTimeMillis()" + error-message-strategy="ems" + send-failure-channel="failures" + send-success-channel="successes" > @@ -45,4 +48,10 @@ partition-id="#{0}" send-timeout="500" /> + + + + + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index 9b418f64e3..4fc614d234 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -74,6 +74,13 @@ public class KafkaOutboundAdapterParserTests { assertThat(TestUtils.getPropertyValue(messageHandler, "timestampExpression.expression")) .isEqualTo("T(System).currentTimeMillis()"); + assertThat(TestUtils.getPropertyValue(messageHandler, "errorMessageStrategy")) + .isSameAs(this.appContext.getBean("ems")); + assertThat(TestUtils.getPropertyValue(messageHandler, "sendFailureChannel")) + .isSameAs(this.appContext.getBean("failures")); + assertThat(TestUtils.getPropertyValue(messageHandler, "outputChannel")) + .isSameAs(this.appContext.getBean("successes")); + messageHandler = this.appContext.getBean("kafkaOutboundChannelAdapter2.handler", KafkaProducerMessageHandler.class); assertThat(messageHandler).isNotNull(); @@ -81,14 +88,11 @@ public class KafkaOutboundAdapterParserTests { assertThat(TestUtils.getPropertyValue(messageHandler, "sync", Boolean.class)).isFalse(); assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.literalValue")).isEqualTo("500"); - - } @Test public void testSyncMode() { - @SuppressWarnings("resource") MockProducer mockProducer = new MockProducer(false, new IntegerSerializer(), new StringSerializer()) {