diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java
index 23d8833abc..f7935a57f1 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2022 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.
@@ -27,6 +27,7 @@ import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
  * Inbound gateway parser.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -48,6 +49,9 @@ public class KafkaInboundGatewayParser extends AbstractInboundGatewayParser {
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				"on-partitions-assigned-seek-callback");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "bind-source-record");
 	}
 
 	@Override
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java
index 20560c1de6..e78f9f1dab 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -59,6 +59,12 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "record-filter-strategy");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
+				"on-partitions-assigned-seek-callback");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ack-discarded");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "filter-in-retry");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "bind-source-record");
 
 		return builder.getBeanDefinition();
 	}
diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd
index a45de95f31..1fe69766f9 100644
--- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd
+++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd
@@ -373,6 +373,46 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="record-filter-strategy" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+								Wrap a 'KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener'
+								into 'FilteringMessageListenerAdapter' with the provided 'RecordFilterStrategy'.
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.kafka.listener.adapter.RecordFilterStrategy" />
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+					<xsd:attribute name="ack-discarded">
+						<xsd:annotation>
+							<xsd:documentation>
+								A boolean flag to indicate if 'FilteringMessageListenerAdapter
+								should acknowledge discarded records or not.
+								Does not make sense if 'record-filter-strategy' isn't specified.
+							</xsd:documentation>
+						</xsd:annotation>
+						<xsd:simpleType>
+							<xsd:union memberTypes="xsd:boolean xsd:string"/>
+						</xsd:simpleType>
+					</xsd:attribute>
+					<xsd:attribute name="filter-in-retry">
+						<xsd:annotation>
+							<xsd:documentation>
+								A boolean flag to specify the order in which the filter and retry
+								operations are performed.
+								Does not make sense if only one of 'retry-template' or
+								'record-filter-strategy' is present, or none.
+								When true, the filter is called for each retry; when false, the filter is only
+								 called once for each delivery from the container.
+							</xsd:documentation>
+						</xsd:annotation>
+						<xsd:simpleType>
+							<xsd:union memberTypes="xsd:boolean xsd:string"/>
+						</xsd:simpleType>
+					</xsd:attribute>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -725,6 +765,30 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="on-partitions-assigned-seek-callback" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					A 'BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback>'
+					bean reference for seeks management.
+				]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref">
+						<tool:expected-type type="java.util.function.BiConsumer" />
+					</tool:annotation>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="bind-source-record">
+			<xsd:annotation>
+				<xsd:documentation>
+					Set to true to bind the source consumer record in the header named
+					'IntegrationMessageHeaderAccessor#SOURCE_DATA'.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:union memberTypes="xsd:boolean xsd:string"/>
+			</xsd:simpleType>
+		</xsd:attribute>
 		<xsd:attributeGroup ref="errorMessageStrategyGroup" />
 	</xsd:complexType>
 
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml
index 4340e8c8f2..52f61a9a2a 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml
@@ -19,7 +19,9 @@
 			payload-type="java.lang.String"
 			error-message-strategy="ems"
 			retry-template="retryTemplate"
-			recovery-callback="recoveryCallback"/>
+			recovery-callback="recoveryCallback"
+			bind-source-record="true"
+			on-partitions-assigned-seek-callback="onPartitionsAssignedSeekCallback"/>
 
 	<bean id="template" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="org.springframework.kafka.core.KafkaTemplate"/>
@@ -44,6 +46,10 @@
 		</constructor-arg>
 	</bean>
 
+	<bean id="onPartitionsAssignedSeekCallback" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="java.util.function.BiConsumer"/>
+	</bean>
+
 	<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy"/>
 
 	<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"/>
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java
index 6add2cd85e..0ca50bfea3 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java
@@ -30,6 +30,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -58,13 +59,16 @@ public class KafkaInboundGatewayTests {
 		assertThat(TestUtils.getPropertyValue(this.gateway1, "listener.fallbackType"))
 				.isEqualTo(String.class);
 		assertThat(TestUtils.getPropertyValue(this.gateway1, "errorMessageStrategy"))
-			.isSameAs(this.context.getBean("ems"));
+				.isSameAs(this.context.getBean("ems"));
 		assertThat(TestUtils.getPropertyValue(this.gateway1, "retryTemplate"))
-			.isSameAs(this.context.getBean("retryTemplate"));
+				.isSameAs(this.context.getBean("retryTemplate"));
 		assertThat(TestUtils.getPropertyValue(this.gateway1, "recoveryCallback"))
-			.isSameAs(this.context.getBean("recoveryCallback"));
+				.isSameAs(this.context.getBean("recoveryCallback"));
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "onPartitionsAssignedSeekCallback"))
+				.isSameAs(this.context.getBean("onPartitionsAssignedSeekCallback"));
 		assertThat(TestUtils.getPropertyValue(this.gateway1, "messagingTemplate.sendTimeout")).isEqualTo(5000L);
 		assertThat(TestUtils.getPropertyValue(this.gateway1, "replyTimeout")).isEqualTo(43L);
+		assertThat(TestUtils.getPropertyValue(this.gateway1, "bindSourceRecord", Boolean.class)).isTrue();
 	}
 
 }
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml
index 6a638129cb..90cba98070 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml
@@ -1,9 +1,9 @@
 <?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"
-	xmlns:context="http://www.springframework.org/schema/context"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	   xmlns:context="http://www.springframework.org/schema/context"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd
 		http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
 
@@ -20,7 +20,12 @@
 			payload-type="java.lang.String"
 			error-message-strategy="ems"
 			retry-template="retryTemplate"
-			recovery-callback="recoveryCallback" />
+			recovery-callback="recoveryCallback"
+			bind-source-record="true"
+			on-partitions-assigned-seek-callback="onPartitionsAssignedSeekCallback"
+			filter-in-retry="true"
+			ack-discarded="true"
+			record-filter-strategy="recordFilterStrategy"/>
 
 	<int-kafka:message-driven-channel-adapter
 			id="kafkaBatchListener"
@@ -31,7 +36,7 @@
 			channel="nullChannel"
 			mode="batch"
 			message-converter="messageConverter"
-			error-channel="errorChannel" />
+			error-channel="errorChannel"/>
 
 	<bean id="messageConverter" class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
 
@@ -40,39 +45,47 @@
 			<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
 				<constructor-arg>
 					<map>
-						<entry key="" value="" />
+						<entry key="" value=""/>
 					</map>
 				</constructor-arg>
 			</bean>
 		</constructor-arg>
 		<constructor-arg>
 			<bean class="org.springframework.kafka.listener.ContainerProperties">
-				<constructor-arg name="topics" value="foo" />
+				<constructor-arg name="topics" value="foo"/>
 			</bean>
 		</constructor-arg>
 	</bean>
 
+	<bean id="onPartitionsAssignedSeekCallback" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="java.util.function.BiConsumer"/>
+	</bean>
+
+	<bean id="recordFilterStrategy" class="org.mockito.Mockito" factory-method="mock">
+		<constructor-arg value="org.springframework.kafka.listener.adapter.RecordFilterStrategy"/>
+	</bean>
+
 	<bean id="container2" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
 		<constructor-arg>
 			<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
 				<constructor-arg>
 					<map>
-						<entry key="" value="" />
+						<entry key="" value=""/>
 					</map>
 				</constructor-arg>
 			</bean>
 		</constructor-arg>
 		<constructor-arg>
 			<bean class="org.springframework.kafka.listener.ContainerProperties">
-				<constructor-arg name="topics" value="foo" />
+				<constructor-arg name="topics" value="foo"/>
 			</bean>
 		</constructor-arg>
 	</bean>
 
-	<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy" />
+	<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy"/>
 
-	<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate" />
+	<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"/>
 
-	<bean id="recoveryCallback" class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer" />
+	<bean id="recoveryCallback" class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer"/>
 
 </beans>
diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
index 1dcf78c25c..bcc89e6dac 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
@@ -24,6 +24,7 @@ import java.util.Collections;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -43,7 +44,7 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
- * @author Artem Bilan.
+ * @author Artem Bilan
  * @author Gary Russell
  *
  * @since 5.4
@@ -52,6 +53,9 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 @DirtiesContext
 class KafkaMessageDrivenChannelAdapterParserTests {
 
+	@Autowired
+	private ApplicationContext context;
+
 	@Autowired
 	private NullChannel nullChannel;
 
@@ -92,6 +96,13 @@ class KafkaMessageDrivenChannelAdapterParserTests {
 		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);
+		assertThat(TestUtils.getPropertyValue(this.kafkaListener, "onPartitionsAssignedSeekCallback"))
+				.isSameAs(this.context.getBean("onPartitionsAssignedSeekCallback"));
+		assertThat(TestUtils.getPropertyValue(this.kafkaListener, "bindSourceRecord", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.kafkaListener, "filterInRetry", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.kafkaListener, "ackDiscarded", Boolean.class)).isTrue();
+		assertThat(TestUtils.getPropertyValue(this.kafkaListener, "recordFilterStrategy"))
+				.isSameAs(this.context.getBean("recordFilterStrategy"));
 	}
 
 	@Test