GH-175: XML Support for Error Handling

Resolves spring-projects/spring-integration-kafka#175

__cherry-pick to 2.3.x__

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

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

spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd

@@ -149,6 +149,33 @@
 					<xsd:union memberTypes="xsd:int xsd:string"/>
 				</xsd:simpleType>
 			</xsd:attribute>
+			<xsd:attribute name="send-failure-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Specifies the channel to which an ErrorMessage for a failed send will be sent.
+					]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel" />
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="send-success-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Specifies the channel to which message with a payload of type
+						'org.apache.kafka.clients.producer.RecordMetadata' will be sent
+						after a successful send.
+					]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.messaging.MessageChannel" />
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attributeGroup ref="errorMessageStrategyGroup" />
 		</xsd:complexType>
 	</xsd:element>
 
@@ -226,6 +253,35 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="retry-template" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						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.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.retry.support.RetryTemplate" />
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="recovery-callback" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						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.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.retry.RecoveryCallback" />
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attributeGroup ref="errorMessageStrategyGroup" />
 		</xsd:complexType>
 	</xsd:element>
 
@@ -236,4 +292,19 @@
 		</xsd:restriction>
 	</xsd:simpleType>
 
+	<xsd:attributeGroup name="errorMessageStrategyGroup">
+		<xsd:attribute name="error-message-strategy" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Specifies the Strategy for building an ErrorMessage.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.integration.support.ErrorMessageStrategy" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
 </xsd:schema>

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" />
 
 	<int-kafka:message-driven-channel-adapter
 			id="kafkaBatchListener"
@@ -71,4 +73,10 @@
 		</constructor-arg>
 	</bean>
 
+	<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy" />
+
+	<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate" />
+
+	<bean id="recoveryCallback" class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer" />
+
 </beans>

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

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"
 	>
 		<int-kafka:request-handler-advice-chain>
 			<bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice" />
@@ -45,4 +48,10 @@
 										partition-id="#{0}"
 										send-timeout="500" />
 
+	<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy" />
+
+	<int:channel id="failures" />
+
+	<int:channel id="successes" />
+
 </beans>

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<Integer, String> mockProducer =
 				new MockProducer<Integer, String>(false, new IntegerSerializer(), new StringSerializer()) {