Add XML attribute for header-mapper

* Added support for header-mapper to outbound-channel-adapter and
outbound-gateway XML

* Follow up on review

* Corrected mistake

* Corrected indentation

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java

@@ -27,6 +27,7 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
  * Utilities to assist with parsing XML.
  *
  * @author Gary Russell
+ * @author Tom van den Berge
  * @since 3.2
  *
  */
@@ -81,6 +82,8 @@ public final class KafkaParsingUtils {
 		if (timestampExpressionDef != null) {
 			builder.addPropertyValue("timestampExpression", timestampExpressionDef);
 		}
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "header-mapper");
 	}
 
 }

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

@@ -84,6 +84,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
  * @author Gary Russell
  * @author Marius Bogoevici
  * @author Biju Kunjummen
+ * @author Tom van den Berge
  *
  * @since 0.5
  */
@@ -186,6 +187,10 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		this.headerMapper = headerMapper;
 	}
 
+	public KafkaHeaderMapper getHeaderMapper() {
+		return this.headerMapper;
+	}
+
 	public KafkaTemplate<?, ?> getKafkaTemplate() {
 		return this.kafkaTemplate;
 	}

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

@@ -31,6 +31,7 @@
 				<xsd:extension base="outboundType">
 					<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
 					<xsd:attributeGroup ref="kafkaTemplate"/>
+					<xsd:attributeGroup ref="headerMapper"/>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -126,6 +127,7 @@
 								]]></xsd:documentation>
 							</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attributeGroup ref="headerMapper"/>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -640,4 +642,19 @@
 			</xsd:annotation>
 		</xsd:attribute>
 	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="headerMapper">
+		<xsd:attribute name="header-mapper" use="optional" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the HeaderMapper used map Spring message headers to or from Kafka message headers.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.kafka.support.KafkaHeaderMapper" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
 </xsd:schema>

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml

@@ -23,6 +23,7 @@
 										error-message-strategy="ems"
 										send-failure-channel="failures"
 										send-success-channel="successes"
+										header-mapper="customHeaderMapper"
 	>
 		<int-kafka:request-handler-advice-chain>
 			<bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice" />
@@ -53,5 +54,7 @@
 	<int:channel id="failures" />
 
 	<int:channel id="successes" />
+	
+	<bean id="customHeaderMapper" class="org.springframework.kafka.support.DefaultKafkaHeaderMapper" />
 
 </beans>

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java

@@ -53,6 +53,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * @author Artem Bilan
  * @author Gary Russell
  * @author Biju Kunjummen
+ * @author Tom van den Berge
  *
  * @since 0.5
  */
@@ -84,6 +85,8 @@ public class KafkaOutboundAdapterParserTests {
 				.isSameAs(this.appContext.getBean("failures"));
 		assertThat(TestUtils.getPropertyValue(messageHandler, "sendSuccessChannel"))
 				.isSameAs(this.appContext.getBean("successes"));
+		assertThat(TestUtils.getPropertyValue(messageHandler, "headerMapper"))
+				.isSameAs(this.appContext.getBean("customHeaderMapper"));
 
 		messageHandler
 				= this.appContext.getBean("kafkaOutboundChannelAdapter2.handler", KafkaProducerMessageHandler.class);
@@ -94,7 +97,6 @@ public class KafkaOutboundAdapterParserTests {
 		assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.literalValue")).isEqualTo("500");
 	}
 
-
 	@Test
 	public void testSyncMode() {
 		MockProducer<Integer, String> mockProducer =

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml

@@ -23,7 +23,9 @@
 		send-timeout-expression="44"
 		sync="true"
 		timestamp-expression="T(System).currentTimeMillis()"
-		topic-expression="'topic'"/>
+		topic-expression="'topic'"
+		header-mapper="customHeaderMapper"
+		/>
 
 	<int:channel id="requests"/>
 
@@ -39,4 +41,5 @@
 		<constructor-arg value="org.springframework.kafka.requestreply.ReplyingKafkaTemplate"/>
 	</bean>
 
+	<bean id="customHeaderMapper" class="org.springframework.kafka.support.DefaultKafkaHeaderMapper" />
 </beans>

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java

@@ -31,6 +31,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
+ * @author Tom van den Berge
  * @since 3.2
  *
  */
@@ -66,6 +67,8 @@ public class KafkaOutboundGatewayParserTests {
 				.isSameAs(this.context.getBean("failures"));
 		assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendSuccessChannel"))
 				.isSameAs(this.context.getBean("successes"));
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "headerMapper"))
+				.isSameAs(this.context.getBean("customHeaderMapper"));
 	}
 
 	public static class EMS extends DefaultErrorMessageStrategy {

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java

@@ -104,6 +104,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
  * @author Gary Russell
  * @author Biju Kunjummen
  * @author Artem Bilan
+ * @author Tom van den Berge
  *
  * @since 2.0
  */
@@ -329,6 +330,28 @@ public class KafkaProducerMessageHandlerTests {
 		producerFactory.destroy();
 	}
 
+	@Test
+	public void testOutboundWithCustomHeaderMapper() throws Exception {
+		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
+				KafkaTestUtils.producerProps(embeddedKafka));
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
+		KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setHeaderMapper(new DefaultKafkaHeaderMapper("!*"));
+		handler.afterPropertiesSet();
+
+		Message<?> message = MessageBuilder.withPayload("foo")
+				.setHeader(KafkaHeaders.TOPIC, topic1)
+				.setHeader("foo-header", "foo-header-value")
+				.build();
+		handler.handleMessage(message);
+
+		ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic1);
+		assertThat(record.headers().toArray().length).isEqualTo(0);
+
+		producerFactory.destroy();
+	}
+
 	@Test
 	public void testOutboundGateway() throws Exception {
 		ConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(