GH-195: Add XML Schema support for new components

Resolves https://github.com/spring-projects/spring-integration-kafka/issues/195

- gateways
- message source

Polishing

Don't start adapters in parser tests

Fix XML filename for test

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.BeanMetadataElement;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the inbound channel adapter.
+ *
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+public class KafkaInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
+
+	@Override
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	@Override
+	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaMessageSource.class);
+		builder.addConstructorArgReference(element.getAttribute("consumer-factory"));
+		String attribute = element.getAttribute("ack-factory");
+		if (StringUtils.hasText(attribute)) {
+			builder.addConstructorArgReference(attribute);
+		}
+		builder.addConstructorArgValue(element.getAttribute("topics"));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "client-id");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "group-id");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "raw-header", "rawMessageHeader");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "rebalance-listener");
+		return builder.getBeanDefinition();
+	}
+
+}

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.integration.config.xml.AbstractInboundGatewayParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
+
+/**
+ * Inbound gateway parser.
+ *
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+public class KafkaInboundGatewayParser extends AbstractInboundGatewayParser {
+
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return KafkaInboundGateway.class;
+	}
+
+	@Override
+	protected void doPostProcess(BeanDefinitionBuilder builder, Element element) {
+		builder.addConstructorArgReference(element.getAttribute("listener-container"));
+		builder.addConstructorArgReference(element.getAttribute("kafka-template"));
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback");
+	}
+
+	@Override
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	@Override
+	protected boolean isEligibleAttribute(String attributeName) {
+		return super.isEligibleAttribute(attributeName)
+				&& !attributeName.equals("listener-container")
+				&& !attributeName.equals("kafka-template");
+	}
+
+}

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

@@ -32,6 +32,9 @@ public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 	public void init() {
 		registerBeanDefinitionParser("outbound-channel-adapter", new KafkaOutboundChannelAdapterParser());
 		registerBeanDefinitionParser("message-driven-channel-adapter", new KafkaMessageDrivenChannelAdapterParser());
+		registerBeanDefinitionParser("outbound-gateway", new KafkaOutboundGatewayParser());
+		registerBeanDefinitionParser("inbound-gateway", new KafkaInboundGatewayParser());
+		registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser());
 	}
 
 }

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

@@ -18,12 +18,10 @@ package org.springframework.integration.kafka.config.xml;
 
 import org.w3c.dom.Element;
 
-import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
-import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
 
 /**
@@ -43,50 +41,7 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd
 		final BeanDefinitionBuilder builder =
 								BeanDefinitionBuilder.genericBeanDefinition(KafkaProducerMessageHandler.class);
 
-		final String kafkaTemplateBeanName = element.getAttribute("kafka-template");
-
-		builder.addConstructorArgReference(kafkaTemplateBeanName);
-
-		BeanDefinition topicExpressionDef =
-				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic", "topic-expression",
-						parserContext, element, false);
-		if (topicExpressionDef != null) {
-			builder.addPropertyValue("topicExpression", topicExpressionDef);
-		}
-
-		BeanDefinition messageKeyExpressionDef =
-				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("message-key",
-						"message-key-expression", parserContext, element, false);
-		if (messageKeyExpressionDef != null) {
-			builder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef);
-		}
-
-		BeanDefinition partitionIdExpressionDef =
-				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("partition-id",
-						"partition-id-expression", parserContext, element, false);
-		if (partitionIdExpressionDef != null) {
-			builder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef);
-		}
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "sync");
-
-		BeanDefinition sendTimeoutExpressionDef =
-				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("send-timeout",
-						"send-timeout-expression", parserContext, element, false);
-		if (sendTimeoutExpressionDef != null) {
-			builder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef);
-		}
-
-		BeanDefinition timestampExpressionDef =
-				IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("timestamp-expression", element);
-
-		if (timestampExpressionDef != null) {
-			builder.addPropertyValue("timestampExpression", timestampExpressionDef);
-		}
-
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-failure-channel");
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel");
-
+		KafkaParsingUtils.commonOutboundProperties(element, parserContext, builder);
 		return builder.getBeanDefinition();
 	}
 

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
+import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
+
+/**
+ * Parser for the outbound gateway.
+ *
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+public class KafkaOutboundGatewayParser extends AbstractConsumerEndpointParser {
+
+	@Override
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	@Override
+	protected String getInputChannelAttributeName() {
+		return "request-channel";
+	}
+
+	@Override
+	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(KafkaProducerMessageHandler.class);
+		KafkaParsingUtils.commonOutboundProperties(element, parserContext, builder);
+		return builder;
+	}
+
+}

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+
+/**
+ * Utilities to assist with parsing XML.
+ *
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+public final class KafkaParsingUtils {
+
+	private KafkaParsingUtils() {
+		super();
+	}
+
+	public static void commonOutboundProperties(final Element element, final ParserContext parserContext,
+			final BeanDefinitionBuilder builder) {
+
+		final String kafkaTemplateBeanName = element.getAttribute("kafka-template");
+		builder.addConstructorArgReference(kafkaTemplateBeanName);
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-failure-channel");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "send-success-channel");
+
+		BeanDefinition topicExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("topic", "topic-expression",
+						parserContext, element, false);
+		if (topicExpressionDef != null) {
+			builder.addPropertyValue("topicExpression", topicExpressionDef);
+		}
+
+		BeanDefinition messageKeyExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("message-key",
+						"message-key-expression", parserContext, element, false);
+		if (messageKeyExpressionDef != null) {
+			builder.addPropertyValue("messageKeyExpression", messageKeyExpressionDef);
+		}
+
+		BeanDefinition partitionIdExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("partition-id",
+						"partition-id-expression", parserContext, element, false);
+		if (partitionIdExpressionDef != null) {
+			builder.addPropertyValue("partitionIdExpression", partitionIdExpressionDef);
+		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "sync");
+
+		BeanDefinition sendTimeoutExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("send-timeout",
+						"send-timeout-expression", parserContext, element, false);
+		if (sendTimeoutExpressionDef != null) {
+			builder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef);
+		}
+
+		BeanDefinition timestampExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("timestamp-expression", element);
+
+		if (timestampExpressionDef != null) {
+			builder.addPropertyValue("timestampExpression", timestampExpressionDef);
+		}
+	}
+
+}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java

@@ -112,6 +112,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 	 */
 	public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer<K, V> messageListenerContainer,
 			ListenerMode mode) {
+
 		Assert.notNull(messageListenerContainer, "messageListenerContainer is required");
 		Assert.isNull(messageListenerContainer.getContainerProperties().getMessageListener(),
 				"Container must not already have a listener");

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -50,6 +50,7 @@ import org.springframework.integration.endpoint.Pausable;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.KafkaMessageHeaders;
@@ -94,8 +95,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	private final Supplier<Duration> minTimeoutProvider =
 			() -> Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT));
 
-	private final Log logger = LogFactory.getLog(getClass());
-
 	private final ConsumerFactory<K, V> consumerFactory;
 
 	private final KafkaAckCallbackFactory<K, V> ackCallbackFactory;
@@ -118,6 +117,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private ConsumerRebalanceListener rebalanceListener;
 
+	private ConsumerAwareRebalanceListener consumerAwareRebalanceListener;
+
 	private boolean rawMessageHeader;
 
 	private Duration commitTimeout;
@@ -145,6 +146,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 		Assert.notNull(consumerFactory, "'consumerFactory' must not be null");
 		Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null");
+		Assert.isTrue(topics != null && topics.length > 0, "At least one topic is required");
 		this.consumerFactory = fixOrRejectConsumerFactory(consumerFactory);
 		this.ackCallbackFactory = ackCallbackFactory;
 		this.topics = topics;
@@ -223,6 +225,9 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	 */
 	public void setRebalanceListener(ConsumerRebalanceListener rebalanceListener) {
 		this.rebalanceListener = rebalanceListener;
+		if (rebalanceListener instanceof ConsumerAwareRebalanceListener) {
+			this.consumerAwareRebalanceListener = (ConsumerAwareRebalanceListener) rebalanceListener;
+		}
 	}
 
 	@Override
@@ -377,6 +382,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	protected void createConsumer() {
 		synchronized (this.consumerMonitor) {
 			this.consumer = this.consumerFactory.createConsumer(this.groupId, this.clientId, null);
+			boolean isConsumerAware = this.consumerAwareRebalanceListener != null;
 			this.consumer.subscribe(Arrays.asList(this.topics), new ConsumerRebalanceListener() {
 
 				@Override
@@ -385,7 +391,11 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 					if (KafkaMessageSource.this.logger.isInfoEnabled()) {
 						KafkaMessageSource.this.logger.info("Partitions revoked: " + partitions);
 					}
-					if (KafkaMessageSource.this.rebalanceListener != null) {
+					if (isConsumerAware) {
+						KafkaMessageSource.this.consumerAwareRebalanceListener.onPartitionsRevokedAfterCommit(
+								KafkaMessageSource.this.consumer, partitions);
+					}
+					else if (KafkaMessageSource.this.rebalanceListener != null) {
 						KafkaMessageSource.this.rebalanceListener.onPartitionsRevoked(partitions);
 					}
 				}
@@ -397,7 +407,11 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 					if (KafkaMessageSource.this.logger.isInfoEnabled()) {
 						KafkaMessageSource.this.logger.info("Partitions assigned: " + partitions);
 					}
-					if (KafkaMessageSource.this.rebalanceListener != null) {
+					if (isConsumerAware) {
+						KafkaMessageSource.this.consumerAwareRebalanceListener.onPartitionsAssigned(
+								KafkaMessageSource.this.consumer, partitions);
+					}
+					else if (KafkaMessageSource.this.rebalanceListener != null) {
 						KafkaMessageSource.this.rebalanceListener.onPartitionsAssigned(partitions);
 					}
 				}

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

@@ -132,8 +132,6 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	private Type replyPayloadType = Object.class;
 
-	private volatile boolean noOutputChannel;
-
 	public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
 		Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
 		this.kafkaTemplate = kafkaTemplate;
@@ -360,6 +358,9 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		String topic = this.topicExpression != null ?
 				this.topicExpression.getValue(this.evaluationContext, message, String.class)
 				: messageHeaders.get(KafkaHeaders.TOPIC, String.class);
+		if (topic == null) {
+			topic = this.kafkaTemplate.getDefaultTopic();
+		}
 
 		Assert.state(StringUtils.hasText(topic), "The 'topic' can not be empty or null");
 

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

@@ -27,155 +27,224 @@
 		</xsd:annotation>
 
 		<xsd:complexType>
-			<xsd:all>
-				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
-				<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
-							 minOccurs="0" maxOccurs="1" />
-			</xsd:all>
+			<xsd:complexContent>
+				<xsd:extension base="outboundType">
+					<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+					<xsd:attributeGroup ref="kafkaTemplate"/>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="outbound-gateway">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines the Consumer Endpoint for the KafkaProducerMessageHandler
+				that writes the contents of the Message to kafka broker and receives
+				a reply.
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="outboundType">
+					<xsd:attribute name="id" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Unique ID for this adapter.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="kafka-template" use="required" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+								Specifies the ReplyingKafkaTemplate used to publish messages and
+								receive replies. The replyTimeout is set on the template.
+							]]></xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.kafka.requestreply.ReplyingKafkaTemplate" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="request-channel" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Message Channel to which Messages should be sent in order to have them sent to Kafka.
+							</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="requires-reply" type="xsd:string" use="optional" default="true">
+						<xsd:annotation>
+							<xsd:documentation>
+								Specify whether this outbound gateway must return a non-null value. This value is
+								'true' by default, and a ReplyRequiredException will be thrown when
+								the underlying service returns a null value.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="reply-channel" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+			Message Channel to which replies should be sent after being received from Kafka.
+							</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="reply-timeout" type="xsd:string">
+							<xsd:annotation>
+								<xsd:documentation><![CDATA[
+									Allows you to specify how long this gateway will wait for
+									the reply message to be sent successfully to the reply channel
+									before throwing an exception. This attribute only applies when the
+									channel might block, for example when using a bounded queue channel that
+									is currently full.
+
+									Also, keep in mind that when sending to a DirectChannel, the
+									invocation will occur in the sender's thread. Therefore,
+									the failing of the send operation may be caused by other
+									components further downstream.
+
+									The "reply-timeout" attribute maps to the "sendTimeout" property of the
+									underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
+
+									The attribute will default, if not specified, to '-1', meaning that
+									by default, the Gateway will wait indefinitely. The value is
+									specified in milliseconds.
+								]]></xsd:documentation>
+							</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="inbound-channel-adapter">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines a Polling Channel Adapter for the
+				'org.springframework.integration.kafka.inbound.KafkaMessageSource'
+				for polling a Kafka topic.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:sequence>
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
 			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
-			<xsd:attribute name="kafka-template" type="xsd:string">
+			<xsd:attribute name="consumer-factory" use="required" type="xsd:string">
 				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the KafkaTemplate used to publish messages.
-					]]></xsd:documentation>
+					<xsd:documentation>
+						A reference to a ConsumerFactory.
+					</xsd:documentation>
 					<xsd:appinfo>
 						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.kafka.core.KafkaTemplate" />
+							<tool:expected-type type="org.springframework.core.ConsumerFactory"/>
 						</tool:annotation>
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="topic" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the name of the Kafka topic.
-								This attribute is mutually exclusive with 'topic-expression' attribute.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="topic-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the expression to determine the name of the Kafka topic
-								against the Message at runtime.
-								This attribute is mutually exclusive with 'topic' attribute.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="message-key" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the Key for the Kafka message.
-								This attribute is mutually exclusive with 'message-key-expression' attribute.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="message-key-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the expression to determine the Key for Kafka message
-								against the Message at runtime.
-								This attribute is mutually exclusive with 'message-key' attribute.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="partition-id" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the target partition for the Kafka message.
-								This attribute is mutually exclusive with 'partition-id-expression' attribute.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="partition-id-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the expression to determine the partition for Kafka message
-								against the Message at runtime.
-								This attribute is mutually exclusive with 'partition-id' attribute.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="timestamp-expression" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-								Specifies the expression to determine the timestamp for a Kafka record
-								against the Message at runtime.
-					]]></xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="sync">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-						Specifies whether to block the sending thread until the producer
-						callback has been invoked, indicating the broker has accepted the
-						message (or an exception thrown if the send fails). Default: false.
-					]]></xsd:documentation>
-				</xsd:annotation>
-				<xsd:simpleType>
-					<xsd:union memberTypes="xsd:boolean xsd:string"/>
-				</xsd:simpleType>
-			</xsd:attribute>
-			<xsd:attribute name="send-timeout">
+			<xsd:attribute name="ack-factory" use="optional" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						Specifies a timeout in milliseconds for how long the 'KafkaProducerMessageHandler'
-						should wait wait for send operation results. Defaults to 10 seconds.
-						The timeout is applied only in 'sync' mode.
+						A reference to a ConsumerFactory.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type
+								type="org.springframework.integration.kafka.inbound.KafkaMessageSource$KafkaAckCallbackFactory"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="topics" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Comma-delimited list of topic names.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="group-id" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						The 'group.id' Kafka Consumer property; overrides the value in the consumer factory.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="client-id" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						The 'client.id' Kafka Consumer property; overrides the value in the consumer factory.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="poll-timeout">
+				<xsd:annotation>
+					<xsd:documentation>
+						The time to block in poll() in milliseconds; default 50ms.
 					</xsd:documentation>
 				</xsd:annotation>
 				<xsd:simpleType>
 					<xsd:union memberTypes="xsd:int xsd:string"/>
 				</xsd:simpleType>
 			</xsd:attribute>
-			<xsd:attribute name="send-timeout-expression">
+			<xsd:attribute name="message-converter" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						Specifies an expression that is evaluated to determine a timeout in milliseconds
-						for how long the 'KafkaProducerMessageHandler'
-						should wait wait for send operation results. Defaults to 10 seconds.
-						The timeout is applied only in 'sync' mode.
+						A reference to a RecordMessageConverter; default 'MessagingMessageConverter'.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type
+								type="org.springframework.kafka.support.converter.RecordMessageConverter"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="payload-type" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						The payload type to which convert the record data; only applies if he message-converter
+						type-aware converter, such as a JSON converter.
+						Use a SpEL expression, e.g. '#{T(java.lang.String)}'.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="order">
+			<xsd:attribute name="rebalance-listener" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						Specifies the order for invocation when this endpoint is connected as a
-						subscriber to a SubscribableChannel.
+						A reference to a 'ConsumerRebalanceListener'.
+					</xsd:documentation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type
+								type="org.apache.kafka.clients.consumer.ConsumerRebalanceListener"/>
+						</tool:annotation>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="raw-header">
+				<xsd:annotation>
+					<xsd:documentation>
+						Set to true to add the raw 'ConsumerRecord' in the 'KafkaHeaders.RAW_DATA' header.
 					</xsd:documentation>
 				</xsd:annotation>
 				<xsd:simpleType>
-					<xsd:union memberTypes="xsd:int xsd:string"/>
+					<xsd:union memberTypes="xsd:int xsd:boolean"/>
 				</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>
 
@@ -186,102 +255,98 @@
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
-			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
-			<xsd:attribute name="send-timeout" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-						Maximum amount of time in milliseconds to wait when sending a message to the channel
-						if such channel may block. For example, a Queue Channel can block until space is available
-						if its maximum capacity has been reached.
-					</xsd:documentation>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="error-channel" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-						Message Channel to which error Messages should 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="listener-container" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-						An 'org.springframework.kafka.listener.AbstractMessageListenerContainer' bean reference.
-					</xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.kafka.listener.AbstractMessageListenerContainer"/>
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="message-converter" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-						An 'org.springframework.kafka.support.converter.MessageConverter' bean reference.
-						if mode = 'record' must be a 'RecordMessageConverter'; if mode = 'batch' must be
-						a `BatchMessageConverter`. Defaults to the default implementation for each mode.
-					</xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.kafka.support.converter.MessageConverter"/>
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
-			<xsd:attribute name="mode" default="record">
-				<xsd:annotation>
-					<xsd:documentation>
-						'record' or 'batch' - default 'record' - one converted ConsumerRecord per message, when
-						'batch' then the payload is a collection of converted ConsumerRecords.
-					</xsd:documentation>
-				</xsd:annotation>
-				<xsd:simpleType>
-					<xsd:union memberTypes="listenerMode xsd:string"/>
-				</xsd:simpleType>
-			</xsd:attribute>
-			<xsd:attribute name="payload-type" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation>
-						Set the payload type to convert to when using a type-aware message converter such as the
-						StringJsonMessageConverter. Fully qualified class name; defaults to 'java.lang.Object'.
-					</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:complexContent>
+				<xsd:extension base="inboundType">
+					<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+					<xsd:attribute name="mode" default="record">
+						<xsd:annotation>
+							<xsd:documentation>
+								'record' or 'batch' - default 'record' - one converted ConsumerRecord per message, when
+								'batch' then the payload is a collection of converted ConsumerRecords.
+							</xsd:documentation>
+						</xsd:annotation>
+						<xsd:simpleType>
+							<xsd:union memberTypes="listenerMode xsd:string"/>
+						</xsd:simpleType>
+					</xsd:attribute>
+					<xsd:attribute name="send-timeout" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Maximum amount of time in milliseconds to wait when sending a message to the channel
+								if such channel may block. For example, a Queue Channel can block until space is available
+								if its maximum capacity has been reached.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="inbound-gateway">
+		<xsd:annotation>
+			<xsd:documentation>
+				Defines the Message Producing Endpoint for the KafkaInboundGateway.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="inboundType">
+					<xsd:attribute name="id" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Unique ID for this gateway.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
+					<xsd:attributeGroup ref="kafkaTemplate"/>
+					<xsd:attribute name="request-channel" use="required" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Message Channel to which converted Messages should 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="request-timeout" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Maximum amount of time in milliseconds to wait when sending a message to the channel
+								if such channel may block. For example, a Queue Channel can block until space is available
+								if its maximum capacity has been reached.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="reply-channel" use="optional" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+	Message Channel where reply Messages will be expected.
+							</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="reply-timeout" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+	Used to set the receiveTimeout on the underlying MessagingTemplate instance
+	(org.springframework.integration.core.MessagingTemplate) for receiving messages
+	from the reply channel. If not specified this property will default to "1000"
+	(1 second). Only applies if the container thread hands off to another thread
+	before the reply is sent.
+							]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>
 
@@ -292,6 +357,223 @@
 		</xsd:restriction>
 	</xsd:simpleType>
 
+	<xsd:complexType name="outboundType">
+		<xsd:all>
+			<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
+			<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
+						 minOccurs="0" maxOccurs="1" />
+		</xsd:all>
+		<xsd:attribute name="topic" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the name of the Kafka topic.
+							This attribute is mutually exclusive with 'topic-expression' attribute.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="topic-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the expression to determine the name of the Kafka topic
+							against the Message at runtime.
+							This attribute is mutually exclusive with 'topic' attribute.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-key" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the Key for the Kafka message.
+							This attribute is mutually exclusive with 'message-key-expression' attribute.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-key-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the expression to determine the Key for Kafka message
+							against the Message at runtime.
+							This attribute is mutually exclusive with 'message-key' attribute.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="partition-id" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the target partition for the Kafka message.
+							This attribute is mutually exclusive with 'partition-id-expression' attribute.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="partition-id-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the expression to determine the partition for Kafka message
+							against the Message at runtime.
+							This attribute is mutually exclusive with 'partition-id' attribute.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="timestamp-expression" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the expression to determine the timestamp for a Kafka record
+							against the Message at runtime.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="sync">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Specifies whether to block the sending thread until the producer
+					callback has been invoked, indicating the broker has accepted the
+					message (or an exception thrown if the send fails). Default: false.
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="send-timeout">
+			<xsd:annotation>
+				<xsd:documentation>
+					Specifies a timeout in milliseconds for how long the 'KafkaProducerMessageHandler'
+					should wait wait for send operation results. Defaults to 10 seconds.
+					The timeout is applied only in 'sync' mode.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:int xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="send-timeout-expression">
+			<xsd:annotation>
+				<xsd:documentation>
+					Specifies an expression that is evaluated to determine a timeout in milliseconds
+					for how long the 'KafkaProducerMessageHandler'
+					should wait wait for send operation results. Defaults to 10 seconds.
+					The timeout is applied only in 'sync' mode.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="order">
+			<xsd:annotation>
+				<xsd:documentation>
+					Specifies the order for invocation when this endpoint is connected as a
+					subscriber to a SubscribableChannel.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<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:complexType name="inboundType">
+		<xsd:attribute name="error-channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Message Channel to which error Messages should 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="listener-container" use="required" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					An 'org.springframework.kafka.listener.AbstractMessageListenerContainer' bean reference.
+				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.kafka.listener.AbstractMessageListenerContainer"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="message-converter" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					An 'org.springframework.kafka.support.converter.MessageConverter' bean reference.
+					if mode = 'record' must be a 'RecordMessageConverter'; if mode = 'batch' must be
+					a `BatchMessageConverter`. Defaults to the default implementation for each mode.
+				</xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.kafka.support.converter.MessageConverter"/>
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="payload-type" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Set the payload type to convert to when using a type-aware message converter such as the
+					StringJsonMessageConverter. Fully qualified class name; defaults to 'java.lang.Object'.
+				</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:attributeGroup name="errorMessageStrategyGroup">
 		<xsd:attribute name="error-message-strategy" type="xsd:string">
 			<xsd:annotation>
@@ -307,4 +589,18 @@
 		</xsd:attribute>
 	</xsd:attributeGroup>
 
+	<xsd:attributeGroup name="kafkaTemplate">
+		<xsd:attribute name="kafka-template" use="required" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+							Specifies the KafkaTemplate used to publish messages.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="org.springframework.kafka.core.KafkaTemplate" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
 </xsd:schema>

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

@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<bean id="cf" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
+		<constructor-arg>
+			<map>
+				<entry key="bootstrap.servers" value="#{embeddedKafka.brokersAsString}"/>
+				<entry key="auto.offset.reset" value="earliest"/>
+				<entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
+				<entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
+				<entry key="max.poll.records" value="1"/>
+			</map>
+		</constructor-arg>
+	</bean>
+
+	<bean id="pf" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
+		<constructor-arg>
+			<map>
+				<entry key="bootstrap.servers" value="#{embeddedKafka.brokersAsString}"/>
+				<entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
+				<entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
+			</map>
+		</constructor-arg>
+	</bean>
+
+	<int-kafka:inbound-channel-adapter
+			channel="fromOne"
+			consumer-factory="cf"
+			topics="one"
+			group-id="groupForTopic1">
+		<int:poller fixed-delay="5000"/>
+	</int-kafka:inbound-channel-adapter>
+
+	<int:chain input-channel="fromOne">
+		<int:transformer expression="payload + headers['kafka_receivedTopic']"/>
+		<int-kafka:outbound-channel-adapter
+				kafka-template="template"
+				topic="two"/>
+	</int:chain>
+
+	<int-kafka:message-driven-channel-adapter
+			listener-container="containerTwo"
+			channel="fromTwo"/>
+
+	<int:chain input-channel="fromTwo" output-channel="lastChannel">
+		<int:transformer expression="payload + headers['kafka_receivedTopic']"/>
+		<int-kafka:outbound-gateway
+				kafka-template="replyingTemplate"/>
+		<int:transformer expression="payload + headers['kafka_receivedTopic']"/>
+	</int:chain>
+
+	<int:channel id="lastChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int-kafka:inbound-gateway
+			listener-container="containerThree"
+			kafka-template="replyingTemplate"
+			request-channel="fromThree"/>
+
+	<int:transformer input-channel="fromThree"
+					 expression="payload + headers['kafka_receivedTopic']"/>
+
+	<bean id="template" class="org.springframework.kafka.core.KafkaTemplate">
+		<constructor-arg ref="pf"/>
+	</bean>
+
+	<bean id="containerTwo" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
+		<constructor-arg ref="cf"/>
+		<constructor-arg>
+			<bean class="org.springframework.kafka.listener.ContainerProperties">
+				<constructor-arg name="topics" value="two"/>
+				<property name="groupId" value="container1"/>
+			</bean>
+		</constructor-arg>
+	</bean>
+
+	<bean id="containerThree" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
+		<constructor-arg ref="cf"/>
+		<constructor-arg>
+			<bean class="org.springframework.kafka.listener.ContainerProperties">
+				<constructor-arg name="topics" value="three"/>
+				<property name="groupId" value="container3"/>
+			</bean>
+		</constructor-arg>
+	</bean>
+
+	<bean id="replyingTemplate" class="org.springframework.kafka.requestreply.ReplyingKafkaTemplate">
+		<constructor-arg ref="pf"/>
+		<constructor-arg ref="containerFour"/>
+		<property name="defaultTopic" value="three"/>
+	</bean>
+
+	<bean id="containerFour" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
+		<constructor-arg ref="cf"/>
+		<constructor-arg>
+			<bean class="org.springframework.kafka.listener.ContainerProperties">
+				<constructor-arg name="topics" value="four"/>
+				<property name="groupId" value="container4"/>
+			</bean>
+		</constructor-arg>
+	</bean>
+
+</beans>

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+@EmbeddedKafka(topics = { "one", "two", "three", "four" })
+public class AllXmlTests {
+
+	@Autowired
+	private KafkaTemplate<String, String> template;
+
+	@Autowired
+	private PollableChannel lastChannel;
+
+	@Test
+	public void testEndToEnd() {
+		this.template.send("one", "foo");
+		Message<?> received = this.lastChannel.receive(30_000);
+		assertThat(received).isNotNull();
+		assertThat(received.getPayload()).isEqualTo("fooonetwothreefour");
+	}
+
+}

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

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
+
+	<int-kafka:inbound-channel-adapter
+			id="adapter1"
+			consumer-factory="consumerFactory"
+			ack-factory="ackFactory"
+			topics="topic1"
+			channel="inbound"
+			client-id="client"
+			group-id="group"
+			message-converter="converter"
+			payload-type="#{T(java.lang.String)}"
+			raw-header="true"
+			auto-startup="false"
+			rebalance-listener="rebal">
+		<int:poller fixed-delay="5000"/>
+	</int-kafka:inbound-channel-adapter>
+
+	<int-kafka:inbound-channel-adapter
+			id="adapter2"
+			consumer-factory="consumerFactory"
+			topics="topic1, topic2"
+			group-id="group"
+			auto-startup="false"
+			channel="inbound">
+		<int:poller fixed-delay="5000"/>
+	</int-kafka:inbound-channel-adapter>
+
+	<bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
+		<constructor-arg>
+			<map>
+				<entry key="max.poll.records" value="1"/>
+			</map>
+		</constructor-arg>
+	</bean>
+
+	<bean id="ackFactory" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg
+			value="org.springframework.integration.kafka.inbound.KafkaMessageSource$KafkaAckCallbackFactory"/>
+	</bean>
+
+	<int:channel id="inbound"/>
+
+	<bean id="converter" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.kafka.support.converter.RecordMessageConverter"/>
+	</bean>
+
+	<bean id="rebal" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.apache.kafka.clients.consumer.ConsumerRebalanceListener"/>
+	</bean>
+
+</beans>

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationContext;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class KafkaInboundChannelAdapterParserTests {
+
+	@Autowired
+	@Qualifier("adapter1.source")
+	private KafkaMessageSource<?, ?> source1;
+
+	@Autowired
+	@Qualifier("adapter2.source")
+	private KafkaMessageSource<?, ?> source2;
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void testProps() {
+		assertThat(TestUtils.getPropertyValue(this.source1, "topics")).isEqualTo(new String[] { "topic1" });
+		assertThat(TestUtils.getPropertyValue(this.source1, "consumerFactory"))
+			.isSameAs(this.context.getBean("consumerFactory"));
+		assertThat(TestUtils.getPropertyValue(this.source1, "ackCallbackFactory"))
+			.isSameAs(this.context.getBean("ackFactory"));
+		assertThat(TestUtils.getPropertyValue(this.source1, "clientId")).isEqualTo("client");
+		assertThat(TestUtils.getPropertyValue(this.source1, "groupId")).isEqualTo("group");
+		assertThat(TestUtils.getPropertyValue(this.source1, "messageConverter"))
+			.isSameAs(this.context.getBean("converter"));
+		assertThat(TestUtils.getPropertyValue(this.source1, "payloadType")).isEqualTo(String.class);
+		assertThat(TestUtils.getPropertyValue(this.source1, "rawMessageHeader", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.source1, "rebalanceListener"))
+			.isSameAs(this.context.getBean("rebal"));
+
+		assertThat(TestUtils.getPropertyValue(this.source2, "topics")).isEqualTo(new String[] { "topic1", "topic2" });
+	}
+
+}

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

@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
+
+	<int-kafka:inbound-gateway
+			id="gateway1"
+			listener-container="container1"
+			kafka-template="template"
+			auto-startup="false"
+			phase="100"
+			request-timeout="5000"
+			request-channel="nullChannel"
+			reply-channel="errorChannel"
+			reply-timeout="43"
+			message-converter="messageConverter"
+			payload-type="java.lang.String"
+			error-message-strategy="ems"
+			retry-template="retryTemplate"
+			recovery-callback="recoveryCallback"/>
+
+	<bean id="template" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.kafka.core.KafkaTemplate"/>
+	</bean>
+
+	<bean id="messageConverter" class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
+
+	<bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
+		<constructor-arg>
+			<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
+				<constructor-arg>
+					<map>
+						<entry key="" value=""/>
+					</map>
+				</constructor-arg>
+			</bean>
+		</constructor-arg>
+		<constructor-arg>
+			<bean class="org.springframework.kafka.listener.ContainerProperties">
+				<constructor-arg name="topics" value="foo"/>
+			</bean>
+		</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/KafkaInboundGatewayTests.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.kafka.listener.KafkaMessageListenerContainer;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class KafkaInboundGatewayTests {
+
+	@Autowired
+	private KafkaInboundGateway<?, ?, ?> gateway1;
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void testProps() {
+		assertThat(this.gateway1.isAutoStartup()).isFalse();
+		assertThat(this.gateway1.isRunning()).isFalse();
+		assertThat(this.gateway1.getPhase()).isEqualTo(100);
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "requestChannelName")).isEqualTo("nullChannel");
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "replyChannelName")).isEqualTo("errorChannel");
+		KafkaMessageListenerContainer<?, ?> container =
+				TestUtils.getPropertyValue(this.gateway1, "messageListenerContainer",
+						KafkaMessageListenerContainer.class);
+		assertThat(container).isNotNull();
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "listener.fallbackType"))
+				.isEqualTo(String.class);
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "errorMessageStrategy"))
+			.isSameAs(this.context.getBean("ems"));
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "retryTemplate"))
+			.isSameAs(this.context.getBean("retryTemplate"));
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "recoveryCallback"))
+			.isSameAs(this.context.getBean("recoveryCallback"));
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "messagingTemplate.sendTimeout")).isEqualTo(5000L);
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "replyTimeout")).isEqualTo(43L);
+	}
+
+}

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

@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<int-kafka:outbound-gateway
+		id="allProps"
+		error-message-strategy="ems"
+		kafka-template="template"
+		message-key-expression="'key'"
+		order="23"
+		partition-id-expression="2"
+		reply-channel="replies"
+		reply-timeout="43"
+		request-channel="requests"
+		requires-reply="false"
+		send-success-channel="successes"
+		send-failure-channel="failures"
+		send-timeout-expression="44"
+		sync="true"
+		timestamp-expression="T(System).currentTimeMillis()"
+		topic-expression="'topic'"/>
+
+	<int:channel id="requests"/>
+
+	<int:channel id="replies"/>
+
+	<int:channel id="successes"/>
+
+	<int:channel id="failures"/>
+
+	<bean id="ems" class="org.springframework.integration.kafka.config.xml.KafkaOutboundGatewayParserTests$EMS"/>
+
+	<bean id="template" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.kafka.requestreply.ReplyingKafkaTemplate"/>
+	</bean>
+
+</beans>

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.config.xml;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationContext;
+import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 3.2
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class KafkaOutboundGatewayParserTests {
+
+	@Autowired
+	@Qualifier("allProps.handler")
+	KafkaProducerMessageHandler<?, ?> messageHandler;
+
+	@Autowired
+	private ApplicationContext context;
+
+	@Test
+	public void testProps() {
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "errorMessageStrategy")).isInstanceOf(EMS.class);
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "kafkaTemplate"))
+			.isSameAs(this.context.getBean("template"));
+		assertThat(this.messageHandler.getOrder()).isEqualTo(23);
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "topicExpression.expression")).isEqualTo("'topic'");
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "messageKeyExpression.expression"))
+			.isEqualTo("'key'");
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "partitionIdExpression.expression")).isEqualTo("2");
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "sync", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendTimeoutExpression.expression")).isEqualTo("44");
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "timestampExpression.expression"))
+				.isEqualTo("T(System).currentTimeMillis()");
+
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "errorMessageStrategy"))
+				.isSameAs(this.context.getBean("ems"));
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendFailureChannel"))
+				.isSameAs(this.context.getBean("failures"));
+		assertThat(TestUtils.getPropertyValue(this.messageHandler, "sendSuccessChannel"))
+				.isSameAs(this.context.getBean("successes"));
+	}
+
+	public static class EMS extends DefaultErrorMessageStrategy {
+
+	}
+
+}

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java

@@ -403,17 +403,18 @@ public class MessageSourceTests {
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	public void testMaxPollRecords() {
-		KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()));
+		KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()),
+				"topic");
 		assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class)
 				.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1);
 		source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(
-				Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)));
+				Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)), "topic");
 		assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class)
 				.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1);
 		try {
 			new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) {
 
-			}));
+			}), "topic");
 			fail("Expected exception");
 		}
 		catch (IllegalArgumentException e) {