Upgrade to SI-5.0 & Port Kafka Java DSL

* Upgrade all the dependencies
* Upgrade to Gradle 3.1 (experimental)
* Make Java 8 as minimal
* Decrease test logging level to `warn`
* Document Java DSL support and some other polishing in the `README.adoc`

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java

@@ -0,0 +1,251 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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.dsl;
+
+import java.util.regex.Pattern;
+
+import org.apache.kafka.common.TopicPartition;
+
+import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
+
+/**
+ * Factory class for Apache Kafka components.
+ *
+ * @author Artem Bilan
+ * @author Nasko Vasilev
+ *
+ * @since 3.0
+ */
+public final class Kafka {
+
+	/**
+	 * Create an initial {@link KafkaProducerMessageHandlerSpec}.
+	 * @param kafkaTemplate the {@link KafkaTemplate} to use
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the Kafka09ProducerMessageHandlerSpec.
+	 */
+	public static <K, V> KafkaProducerMessageHandlerSpec<K, V>
+	outboundChannelAdapter(KafkaTemplate<K, V> kafkaTemplate) {
+		return new KafkaProducerMessageHandlerSpec<>(kafkaTemplate);
+	}
+
+	/**
+	 * Create an initial {@link KafkaProducerMessageHandlerSpec} with ProducerFactory.
+	 * @param producerFactory the {@link ProducerFactory} Java 8 Lambda.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaProducerMessageHandlerSpec.
+	 * @see <a href="https://kafka.apache.org/documentation.html#producerconfigs">Kafka Producer Configs</a>
+	 */
+	public static <K, V> KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<K, V>
+	outboundChannelAdapter(ProducerFactory<K, V> producerFactory) {
+		return new KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<>(producerFactory);
+	}
+
+	/**
+	 * Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}.
+	 * @param listenerContainer the {@link AbstractMessageListenerContainer}.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @param <A> the {@link KafkaMessageDrivenChannelAdapterSpec} extension type.
+	 * @return the Kafka09MessageDrivenChannelAdapterSpec.
+	 */
+	public static <K, V, A extends KafkaMessageDrivenChannelAdapterSpec<K, V, A>>
+	KafkaMessageDrivenChannelAdapterSpec<K, V, A> messageDrivenChannelAdapter(
+			AbstractMessageListenerContainer<K, V> listenerContainer) {
+		return messageDrivenChannelAdapter(listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode.record);
+	}
+
+	/**
+	 * Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}.
+	 * @param listenerContainer the {@link AbstractMessageListenerContainer}.
+	 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @param <A> the {@link KafkaMessageDrivenChannelAdapterSpec} extension type.
+	 * @return the Kafka09MessageDrivenChannelAdapterSpec.
+	 */
+	public static <K, V, A extends KafkaMessageDrivenChannelAdapterSpec<K, V, A>>
+	KafkaMessageDrivenChannelAdapterSpec<K, V, A> messageDrivenChannelAdapter(
+			AbstractMessageListenerContainer<K, V> listenerContainer,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
+		return new KafkaMessageDrivenChannelAdapterSpec<>(listenerContainer, listenerMode);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param containerProperties the {@link ContainerProperties} to use.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties) {
+		return messageDrivenChannelAdapter(consumerFactory, containerProperties,
+				KafkaMessageDrivenChannelAdapter.ListenerMode.record);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param containerProperties the {@link ContainerProperties} to use.
+	 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
+		return messageDrivenChannelAdapter(
+				new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
+						containerProperties), listenerMode);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param topicPartitions the {@link TopicPartition} vararg.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
+			TopicPartitionInitialOffset... topicPartitions) {
+		return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
+				topicPartitions);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
+	 * @param topicPartitions the {@link TopicPartition} vararg.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
+			TopicPartitionInitialOffset... topicPartitions) {
+		return messageDrivenChannelAdapter(
+				new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
+						topicPartitions), listenerMode);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param topics the topics vararg.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, String... topics) {
+		return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
+				topics);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
+	 * @param topics the topics vararg.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, String... topics) {
+		return messageDrivenChannelAdapter(
+				new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
+						topics), listenerMode);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param topicPattern the topicPattern vararg.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
+		return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
+				topicPattern);
+	}
+
+	/**
+	 * Create an initial
+	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
+	 * @param consumerFactory the {@link ConsumerFactory}.
+	 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
+	 * @param topicPattern the topicPattern vararg.
+	 * @param <K> the Kafka message key type.
+	 * @param <V> the Kafka message value type.
+	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
+	 */
+	public static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, Pattern topicPattern) {
+		return messageDrivenChannelAdapter(
+				new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
+						topicPattern),
+				listenerMode);
+	}
+
+	private static <K, V>
+	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
+	messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K, V> spec,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
+		return new KafkaMessageDrivenChannelAdapterSpec
+				.KafkaMessageDrivenChannelAdapterListenerContainerSpec<>(spec, listenerMode);
+	}
+
+	private Kafka() {
+		super();
+	}
+
+}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java

@@ -0,0 +1,450 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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.dsl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+
+import org.springframework.core.task.AsyncListenableTaskExecutor;
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.integration.dsl.MessageProducerSpec;
+import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.AcknowledgingMessageListener;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.listener.ErrorHandler;
+import org.springframework.kafka.listener.adapter.FilteringAcknowledgingMessageListenerAdapter;
+import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
+import org.springframework.kafka.listener.adapter.RetryingAcknowledgingMessageListenerAdapter;
+import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.converter.BatchMessageConverter;
+import org.springframework.kafka.support.converter.MessageConverter;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageProducerSpec} implementation for the {@link KafkaMessageDrivenChannelAdapter}.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ * @param <S> the target {@link KafkaMessageDrivenChannelAdapterSpec} implementation type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDrivenChannelAdapterSpec<K, V, S>>
+		extends MessageProducerSpec<S, KafkaMessageDrivenChannelAdapter<K, V>> {
+
+	KafkaMessageDrivenChannelAdapterSpec(AbstractMessageListenerContainer<K, V> messageListenerContainer,
+			KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
+		super(new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer, listenerMode));
+	}
+
+	/**
+	 * Set the message converter; must be a {@link RecordMessageConverter} or
+	 * {@link BatchMessageConverter} depending on mode.
+	 * @param messageConverter the converter.
+	 * @return the spec
+	 */
+	public S messageConverter(MessageConverter messageConverter) {
+		this.target.setMessageConverter(messageConverter);
+		return _this();
+	}
+
+	/**
+	 * Set the message converter to use with a record-based consumer.
+	 * @param messageConverter the converter.
+	 * @return the spec
+	 */
+	public S recordMessageConverter(RecordMessageConverter messageConverter) {
+		this.target.setRecordMessageConverter(messageConverter);
+		return _this();
+	}
+
+	/**
+	 * Set the message converter to use with a batch-based consumer.
+	 * @param messageConverter the converter.
+	 * @return the spec
+	 */
+	public S batchMessageConverter(BatchMessageConverter messageConverter) {
+		this.target.setBatchMessageConverter(messageConverter);
+		return _this();
+	}
+
+	/**
+	 * Specify a {@link RecordFilterStrategy} to wrap
+	 * {@code KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into
+	 * {@link FilteringAcknowledgingMessageListenerAdapter}.
+	 * @param recordFilterStrategy the {@link RecordFilterStrategy} to use.
+	 * @return the spec
+	 */
+	public S recordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrategy) {
+		this.target.setRecordFilterStrategy(recordFilterStrategy);
+		return _this();
+	}
+
+	/**
+	 * A {@code boolean} flag to indicate if {@link FilteringAcknowledgingMessageListenerAdapter}
+	 * should acknowledge discarded records or not.
+	 * Does not make sense if {@link #recordFilterStrategy(RecordFilterStrategy)} isn't specified.
+	 * @param ackDiscarded true to ack (commit offset for) discarded messages.
+	 * @return the spec
+	 */
+	public S ackDiscarded(boolean ackDiscarded) {
+		this.target.setAckDiscarded(ackDiscarded);
+		return _this();
+	}
+
+	/**
+	 * Specify a {@link RetryTemplate} instance to wrap
+	 * {@code KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into
+	 * {@link RetryingAcknowledgingMessageListenerAdapter}.
+	 * @param retryTemplate the {@link RetryTemplate} to use.
+	 * @return the spec
+	 */
+	public S retryTemplate(RetryTemplate retryTemplate) {
+		this.target.setRetryTemplate(retryTemplate);
+		return _this();
+	}
+
+	/**
+	 * A {@link RecoveryCallback} instance for retry operation;
+	 * if null, the exception will be thrown to the container after retries are exhausted.
+	 * Does not make sense if {@link #retryTemplate(RetryTemplate)} isn't specified.
+	 * @param recoveryCallback the recovery callback.
+	 * @return the spec
+	 */
+	public S recoveryCallback(RecoveryCallback<Void> recoveryCallback) {
+		this.target.setRecoveryCallback(recoveryCallback);
+		return _this();
+	}
+
+	/**
+	 /**
+	 * The {@code boolean} flag to specify the order how
+	 * {@link RetryingAcknowledgingMessageListenerAdapter} and
+	 * {@link FilteringAcknowledgingMessageListenerAdapter} are wrapped to each other,
+	 * if both of them are present.
+	 * Does not make sense if only one of {@link RetryTemplate} or
+	 * {@link RecordFilterStrategy} is present, or any.
+	 * @param filterInRetry the order for {@link RetryingAcknowledgingMessageListenerAdapter} and
+	 * {@link FilteringAcknowledgingMessageListenerAdapter} wrapping. Defaults to {@code false}.
+	 * @return the spec
+	 */
+	public S filterInRetry(boolean filterInRetry) {
+		this.target.setFilterInRetry(filterInRetry);
+		return _this();
+	}
+
+	/**
+	 * A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaMessageDrivenChannelAdapterSpec}
+	 * extension.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 */
+	public static class KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> extends
+			KafkaMessageDrivenChannelAdapterSpec<K, V, KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>>
+			implements ComponentsRegistration {
+
+		private KafkaMessageListenerContainerSpec<K, V> spec;
+
+		KafkaMessageDrivenChannelAdapterListenerContainerSpec(KafkaMessageListenerContainerSpec<K, V> spec,
+				KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
+			super(spec.container, listenerMode);
+			this.spec = spec;
+		}
+
+		/**
+		 * Configure a listener container by invoking the {@link Consumer} callback, with a
+		 * {@link KafkaMessageListenerContainerSpec} argument.
+		 * @param configurer the configurer Java 8 Lambda.
+		 * @return the spec.
+		 */
+		public KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> configureListenerContainer(
+				Consumer<KafkaMessageListenerContainerSpec<K, V>> configurer) {
+			Assert.notNull(configurer);
+			configurer.accept(this.spec);
+			return _this();
+		}
+
+		@Override
+		public Collection<Object> getComponentsToRegister() {
+			return Collections.singleton(this.spec.container);
+		}
+
+	}
+
+	/**
+	 * A helper class in the Builder pattern style to delegate options to the
+	 * {@link ConcurrentMessageListenerContainer}.
+	 *
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 */
+	public static class KafkaMessageListenerContainerSpec<K, V> {
+
+		private final ConcurrentMessageListenerContainer<K, V> container;
+
+
+		KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
+				ContainerProperties containerProperties) {
+			this.container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties);
+		}
+
+		KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
+				TopicPartitionInitialOffset... topicPartitions) {
+			this(consumerFactory, new ContainerProperties(topicPartitions));
+		}
+
+		KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory, String... topics) {
+			this(consumerFactory, new ContainerProperties(topics));
+		}
+
+		KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
+			this(consumerFactory, new ContainerProperties(topicPattern));
+		}
+
+		/**
+		 * Specify a concurrency maximum number for the {@link AbstractMessageListenerContainer}.
+		 * @param concurrency the concurrency maximum number.
+		 * @return the spec.
+		 * @see ConcurrentMessageListenerContainer#setConcurrency(int)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> concurrency(int concurrency) {
+			this.container.setConcurrency(concurrency);
+			return this;
+		}
+
+		/**
+		 * Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}.
+		 * @param errorHandler the {@link ErrorHandler}.
+		 * @return the spec.
+		 * @see ErrorHandler
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> errorHandler(ErrorHandler errorHandler) {
+			this.container.getContainerProperties().setErrorHandler(errorHandler);
+			return this;
+		}
+
+		/**
+		 * Set the ack mode to use when auto ack (in the configuration properties) is false.
+		 * <ul>
+		 * <li>RECORD: Ack after each record has been passed to the listener.</li>
+		 * <li>BATCH: Ack after each batch of records received from the consumer has been
+		 * passed to the listener</li>
+		 * <li>TIME: Ack after this number of milliseconds; (should be greater than
+		 * {@code #setPollTimeout(long) pollTimeout}.</li>
+		 * <li>COUNT: Ack after at least this number of records have been received</li>
+		 * <li>MANUAL: Listener is responsible for acking - use a
+		 * {@link AcknowledgingMessageListener}.
+		 * </ul>
+		 * @param ackMode the {@link AbstractMessageListenerContainer.AckMode}; default BATCH.
+		 * @return the spec.
+		 * @see AbstractMessageListenerContainer.AckMode
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> ackMode(AbstractMessageListenerContainer.AckMode ackMode) {
+			this.container.getContainerProperties().setAckMode(ackMode);
+			return this;
+		}
+
+		/**
+		 * Set the max time to block in the consumer waiting for records.
+		 * @param pollTimeout the timeout in ms; default 1000.
+		 * @return the spec.
+		 * @see ContainerProperties#setPollTimeout(long)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> pollTimeout(long pollTimeout) {
+			this.container.getContainerProperties().setPollTimeout(pollTimeout);
+			return this;
+		}
+
+		/**
+		 * Set the number of outstanding record count after which offsets should be
+		 * committed when {@link AbstractMessageListenerContainer.AckMode#COUNT}
+		 * or {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
+		 * @param count the count
+		 * @return the spec.
+		 * @see ContainerProperties#setAckCount(int)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> ackCount(int count) {
+			this.container.getContainerProperties().setAckCount(count);
+			return this;
+		}
+
+		/**
+		 * Set the time (ms) after which outstanding offsets should be committed when
+		 * {@link AbstractMessageListenerContainer.AckMode#TIME} or
+		 * {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
+		 * Should be larger than zero.
+		 * @param millis the time
+		 * @return the spec.
+		 * @see ContainerProperties#setAckTime(long)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> ackTime(long millis) {
+			this.container.getContainerProperties().setAckTime(millis);
+			return this;
+		}
+
+		/**
+		 * Set the executor for threads that poll the consumer.
+		 * @param consumerTaskExecutor the executor
+		 * @return the spec.
+		 * @see ContainerProperties#setConsumerTaskExecutor(AsyncListenableTaskExecutor)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> consumerTaskExecutor(
+				AsyncListenableTaskExecutor consumerTaskExecutor) {
+			this.container.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor);
+			return this;
+		}
+
+		/**
+		 * Set the executor for threads that invoke the listener.
+		 * @param listenerTaskExecutor the executor
+		 * @return the spec.
+		 * @see ContainerProperties#setListenerTaskExecutor(AsyncListenableTaskExecutor)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> listenerTaskExecutor(
+				AsyncListenableTaskExecutor listenerTaskExecutor) {
+			this.container.getContainerProperties().setListenerTaskExecutor(listenerTaskExecutor);
+			return this;
+		}
+
+		/**
+		 * When using Kafka group management and {@link #pauseEnabled(boolean)} is
+		 * true, set the delay after which the consumer should be paused. Default 10000.
+		 * @param pauseAfter the delay.
+		 * @return the spec.
+		 * @see ContainerProperties#setPauseAfter(long)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> pauseAfter(long pauseAfter) {
+			this.container.getContainerProperties().setPauseAfter(pauseAfter);
+			return this;
+		}
+
+		/**
+		 * Set to true to avoid rebalancing when this consumer is slow or throws a
+		 * qualifying exception - pause the consumer. Default: true.
+		 * @param pauseEnabled true to pause.
+		 * @return the spec.
+		 * @see #pauseAfter(long)
+		 * @see ContainerProperties#setPauseEnabled(boolean)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> pauseEnabled(boolean pauseEnabled) {
+			this.container.getContainerProperties().setPauseEnabled(pauseEnabled);
+			return this;
+		}
+
+		/**
+		 * Set the queue depth for handoffs from the consumer thread to the listener
+		 * thread. Default 1 (up to 2 in process).
+		 * @param queueDepth the queue depth.
+		 * @return the spec.
+		 * @see ContainerProperties#setQueueDepth(int)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> queueDepth(int queueDepth) {
+			this.container.getContainerProperties().setQueueDepth(queueDepth);
+			return this;
+		}
+
+		/**
+		 * Set the timeout for shutting down the container. This is the maximum amount of
+		 * time that the invocation to {@code #stop(Runnable)} will block for, before
+		 * returning.
+		 * @param shutdownTimeout the shutdown timeout.
+		 * @return the spec.
+		 * @see ContainerProperties#setShutdownTimeout(long)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> shutdownTimeout(long shutdownTimeout) {
+			this.container.getContainerProperties().setShutdownTimeout(shutdownTimeout);
+			return this;
+		}
+
+		/**
+		 * Set the user defined {@link ConsumerRebalanceListener} implementation.
+		 * @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance
+		 * @return the spec.
+		 * @see ContainerProperties#setConsumerRebalanceListener(ConsumerRebalanceListener)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> consumerRebalanceListener(
+				ConsumerRebalanceListener consumerRebalanceListener) {
+			this.container.getContainerProperties().setConsumerRebalanceListener(consumerRebalanceListener);
+			return this;
+		}
+
+		/**
+		 * Set the commit callback; by default a simple logging callback is used to log
+		 * success at DEBUG level and failures at ERROR level.
+		 * @param commitCallback the callback.
+		 * @return the spec.
+		 * @see ContainerProperties#setCommitCallback(OffsetCommitCallback)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> commitCallback(OffsetCommitCallback commitCallback) {
+			this.container.getContainerProperties().setCommitCallback(commitCallback);
+			return this;
+		}
+
+		/**
+		 * Set whether or not to call consumer.commitSync() or commitAsync() when the
+		 * container is responsible for commits. Default true. See
+		 * https://github.com/spring-projects/spring-kafka/issues/62 At the time of
+		 * writing, async commits are not entirely reliable.
+		 * @param syncCommits true to use commitSync().
+		 * @return the spec.
+		 * @see ContainerProperties#setSyncCommits(boolean)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> syncCommits(boolean syncCommits) {
+			this.container.getContainerProperties().setSyncCommits(syncCommits);
+			return this;
+		}
+
+		/**
+		 * Set the idle event interval; when set, an event is emitted if a poll returns
+		 * no records and this interval has elapsed since a record was returned.
+		 * @param idleEventInterval the interval.
+		 * @return the spec.
+		 * @see ContainerProperties#setIdleEventInterval(Long)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> idleEventInterval(Long idleEventInterval) {
+			this.container.getContainerProperties().setIdleEventInterval(idleEventInterval);
+			return this;
+		}
+
+		/**
+		 * Set whether the container should ack messages that throw exceptions or not.
+		 * @param ackOnError whether the container should acknowledge messages that throw
+		 * exceptions.
+		 * @return the spec.
+		 * @see ContainerProperties#setAckOnError(boolean)
+		 */
+		public KafkaMessageListenerContainerSpec<K, V> ackOnError(boolean ackOnError) {
+			this.container.getContainerProperties().setAckOnError(ackOnError);
+			return this;
+		}
+
+	}
+
+}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java

@@ -0,0 +1,251 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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.dsl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.function.Function;
+
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.integration.dsl.MessageHandlerSpec;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.ProducerListener;
+import org.springframework.kafka.support.converter.RecordMessageConverter;
+import org.springframework.messaging.Message;
+
+/**
+ * A {@link MessageHandlerSpec} implementation for the {@link KafkaProducerMessageHandler}.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+public class KafkaProducerMessageHandlerSpec<K, V>
+		extends MessageHandlerSpec<KafkaProducerMessageHandlerSpec<K, V>, KafkaProducerMessageHandler<K, V>> {
+
+	protected final KafkaTemplate<K, V> kafkaTemplate;
+
+	KafkaProducerMessageHandlerSpec(KafkaTemplate<K, V> kafkaTemplate) {
+		this.target = new KafkaProducerMessageHandler<K, V>(kafkaTemplate);
+		this.kafkaTemplate = kafkaTemplate;
+	}
+
+	/**
+	 * Configure the Kafka topic to send messages.
+	 * @param topic the Kafka topic name.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> topic(String topic) {
+		return topicExpression(new LiteralExpression(topic));
+	}
+
+
+	/**
+	 * Configure a SpEL expression to determine the Kafka topic at runtime against
+	 * request Message as a root object of evaluation context.
+	 * @param topicExpression the topic SpEL expression.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> topicExpression(String topicExpression) {
+		return topicExpression(PARSER.parseExpression(topicExpression));
+	}
+
+	/**
+	 * Configure an {@link Expression} to determine the Kafka topic at runtime against
+	 * request Message as a root object of evaluation context.
+	 * @param topicExpression the topic expression.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> topicExpression(Expression topicExpression) {
+		this.target.setTopicExpression(topicExpression);
+		return _this();
+	}
+
+	/**
+	 * Configure a {@link Function} that will be invoked at run time to determine the topic to
+	 * which a message will be sent. Typically used with a Java 8 Lambda expression:
+	 * <pre class="code">
+	 * {@code
+	 * .<Foo>topic(m -> m.getPayload().getTopic())
+	 * }
+	 * </pre>
+	 * @param topicFunction the topic function.
+	 * @param <P> the expected payload type.
+	 * @return the current {@link KafkaProducerMessageHandlerSpec}.
+	 * @see FunctionExpression
+	 */
+	public <P> KafkaProducerMessageHandlerSpec<K, V> topic(Function<Message<P>, String> topicFunction) {
+		return topicExpression(new FunctionExpression<>(topicFunction));
+	}
+
+	/**
+	 * Configure a SpEL expression to determine the Kafka message key to store at runtime against
+	 * request Message as a root object of evaluation context.
+	 * @param messageKeyExpression the message key SpEL expression.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> messageKeyExpression(String messageKeyExpression) {
+		return messageKeyExpression(PARSER.parseExpression(messageKeyExpression));
+	}
+
+	/**
+	 * Configure the message key to store message in Kafka topic.
+	 * @param messageKey the message key to use.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> messageKey(String messageKey) {
+		return messageKeyExpression(new LiteralExpression(messageKey));
+	}
+
+	/**
+	 * Configure an {@link Expression} to determine the Kafka message key to store at runtime against
+	 * request Message as a root object of evaluation context.
+	 * @param messageKeyExpression the message key expression.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> messageKeyExpression(Expression messageKeyExpression) {
+		this.target.setMessageKeyExpression(messageKeyExpression);
+		return _this();
+	}
+
+	/**
+	 * Configure a {@link Function} that will be invoked at run time to determine the message key under
+	 * which a message will be stored in the topic. Typically used with a Java 8 Lambda expression:
+	 * <pre class="code">
+	 * {@code
+	 * .<Foo>messageKey(m -> m.getPayload().getKey())
+	 * }
+	 * </pre>
+	 * @param messageKeyFunction the message key function.
+	 * @param <P> the expected payload type.
+	 * @return the current {@link KafkaProducerMessageHandlerSpec}.
+	 * @see FunctionExpression
+	 */
+	public <P> KafkaProducerMessageHandlerSpec<K, V> messageKey(Function<Message<P>, ?> messageKeyFunction) {
+		return messageKeyExpression(new FunctionExpression<>(messageKeyFunction));
+	}
+
+	/**
+	 * Configure a partitionId of Kafka topic.
+	 * @param partitionId the partitionId to use.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> partitionId(Integer partitionId) {
+		return partitionIdExpression(new ValueExpression<Integer>(partitionId));
+	}
+
+	/**
+	 * Configure a SpEL expression to determine the topic partitionId at runtime against
+	 * request Message as a root object of evaluation context.
+	 * @param partitionIdExpression the partitionId expression to use.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> partitionIdExpression(String partitionIdExpression) {
+		return partitionIdExpression(PARSER.parseExpression(partitionIdExpression));
+	}
+
+	/**
+	 * Configure a {@link Function} that will be invoked at run time to determine the partition id under
+	 * which a message will be stored in the topic. Typically used with a Java 8 Lambda expression:
+	 * <pre class="code">
+	 * {@code
+	 * .partitionId(m -> m.getHeaders().get("partitionId", Integer.class))
+	 * }
+	 * </pre>
+	 * @param partitionIdFunction the partitionId function.
+	 * @param <P> the expected payload type.
+	 * @return the spec.
+	 */
+	public <P> KafkaProducerMessageHandlerSpec<K, V> partitionId(Function<Message<P>, Integer> partitionIdFunction) {
+		return partitionIdExpression(new FunctionExpression<>(partitionIdFunction));
+	}
+
+	/**
+	 * Configure an {@link Expression} to determine the topic partitionId at runtime against
+	 * request Message as a root object of evaluation context.
+	 * @param partitionIdExpression the partitionId expression to use.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> partitionIdExpression(Expression partitionIdExpression) {
+		this.target.setPartitionIdExpression(partitionIdExpression);
+		return _this();
+	}
+
+	/**
+	 * A {@code boolean} indicating if the {@link KafkaProducerMessageHandler}
+	 * should wait for the send operation results or not. Defaults to {@code false}.
+	 * In {@code sync} mode a downstream send operation exception will be re-thrown.
+	 * @param sync the send mode; async by default.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> sync(boolean sync) {
+		this.target.setSync(sync);
+		return this;
+	}
+
+	/**
+	 * Specify a timeout in milliseconds how long {@link KafkaProducerMessageHandler}
+	 * should wait wait for send operation results. Defaults to 10 seconds.
+	 * @param sendTimeout the timeout to wait for result fo send operation.
+	 * @return the spec.
+	 */
+	public KafkaProducerMessageHandlerSpec<K, V> sendTimeout(long sendTimeout) {
+		this.target.setSendTimeout(sendTimeout);
+		return this;
+	}
+
+	/**
+	 * A {@link KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension.
+	 *
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 */
+	public static class KafkaProducerMessageHandlerTemplateSpec<K, V> extends KafkaProducerMessageHandlerSpec<K, V>
+			implements ComponentsRegistration {
+
+		KafkaProducerMessageHandlerTemplateSpec(ProducerFactory<K, V> producerFactory) {
+			super(new KafkaTemplate<>(producerFactory));
+		}
+
+		public KafkaProducerMessageHandlerTemplateSpec<K, V> producerListener(ProducerListener<K, V> producerListener) {
+			this.kafkaTemplate.setProducerListener(producerListener);
+			return this;
+		}
+
+		public KafkaProducerMessageHandlerTemplateSpec<K, V> messageConverter(RecordMessageConverter messageConverter) {
+			this.kafkaTemplate.setMessageConverter(messageConverter);
+			return this;
+		}
+
+		@Override
+		public Collection<Object> getComponentsToRegister() {
+			return Collections.singleton(this.kafkaTemplate);
+		}
+
+	}
+
+}
+

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides Kafka Components support for Spring Integration Java DSL.
+ */
+package org.springframework.integration.kafka.dsl;

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java

@@ -0,0 +1,201 @@
+/*
+ * Copyright 2015-2016 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
+ *
+ *      http://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.dsl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.MessageRejectedException;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
+import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Artem Bilan
+ * @author Nasko Vasilev
+ *
+ * @since 3.0
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class KafkaTests {
+
+	private static final String TEST_TOPIC = "test-topic";
+
+	private static final String TEST_TOPIC2 = "test-topic2";
+
+	private static final String TEST_TOPIC3 = "test-topic3";
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC, TEST_TOPIC2, TEST_TOPIC3);
+
+	@Autowired
+	@Qualifier("sendToKafkaFlow.input")
+	private MessageChannel sendToKafkaFlowInput;
+
+	@Autowired
+	private PollableChannel listeningFromKafkaResults;
+
+	@Autowired
+	@Qualifier("kafkaProducer.handler")
+	private KafkaProducerMessageHandler<?, ?> kafkaProducer;
+
+	@Autowired
+	@Qualifier("kafkaProducer3.handler")
+	private KafkaProducerMessageHandler<?, ?> kafkaProducer3;
+
+	@Autowired
+	private PollableChannel errorChannel;
+
+	@Test
+	public void testKafkaAdapters() {
+
+		assertThatThrownBy(() -> this.sendToKafkaFlowInput.send(new GenericMessage<>("foo")))
+				.hasMessageContaining("10 is not in the range");
+
+		this.kafkaProducer.setPartitionIdExpression(new ValueExpression<>(0));
+		this.kafkaProducer3.setPartitionIdExpression(new ValueExpression<>(0));
+		this.sendToKafkaFlowInput.send(new GenericMessage<>("foo"));
+
+		for (int i = 0; i < 100; i++) {
+			Message<?> receive = this.listeningFromKafkaResults.receive(10000);
+			assertThat(receive).isNotNull();
+			assertThat(receive.getPayload()).isEqualTo("FOO");
+			MessageHeaders headers = receive.getHeaders();
+			assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue();
+			Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
+			acknowledgment.acknowledge();
+			assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC);
+			assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(i + 1);
+			assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
+			assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo((long) i);
+		}
+
+		Message<String> message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build();
+
+		this.sendToKafkaFlowInput.send(message);
+
+		assertThat(this.listeningFromKafkaResults.receive(10)).isNull();
+
+		Message<?> error = this.errorChannel.receive(10000);
+		assertThat(error).isNotNull();
+		assertThat(error).isInstanceOf(ErrorMessage.class);
+		assertThat(error.getPayload()).isInstanceOf(MessageRejectedException.class);
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+
+		@Bean
+		public ConsumerFactory<Integer, String> consumerFactory() {
+			Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka);
+			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+			return new DefaultKafkaConsumerFactory<>(props);
+		}
+
+		@Bean
+		public PollableChannel errorChannel() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		public IntegrationFlow listeningFromKafkaFlow() {
+			return IntegrationFlows
+					.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
+							KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC)
+							.configureListenerContainer(c ->
+									c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL))
+							.errorChannel("errorChannel")
+							.retryTemplate(new RetryTemplate())
+							.filterInRetry(true))
+					.filter(Message.class, m ->
+									m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
+							f -> f.throwExceptionOnRejection(true))
+					.<String, String>transform(String::toUpperCase)
+					.channel(c -> c.queue("listeningFromKafkaResults"))
+					.get();
+		}
+
+		@Bean
+		public ProducerFactory<Integer, String> producerFactory() {
+			return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka));
+		}
+
+		@Bean
+		public IntegrationFlow sendToKafkaFlow() {
+			return f -> f
+					.<String>split(p -> Stream.generate(() -> p).limit(101).iterator(), null)
+					.publishSubscribeChannel(c -> c
+							.subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC),
+									e -> e.id("kafkaProducer")))
+							.subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC3),
+									e -> e.id("kafkaProducer3")))
+					);
+		}
+
+		private KafkaProducerMessageHandlerSpec<Integer, String>
+		kafkaMessageHandler(ProducerFactory<Integer, String> producerFactory, String topic) {
+			return Kafka
+					.outboundChannelAdapter(producerFactory)
+					.messageKey(m -> m
+							.getHeaders()
+							.get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
+					.partitionId(m -> 10)
+					.topicExpression("headers[kafka_topic] ?: '" + topic + "'");
+		}
+
+	}
+
+}

spring-integration-kafka/src/test/resources/log4j2-test.xml

@@ -8,8 +8,8 @@
 	<Loggers>
 		<Logger name="kafka" level="warn"/>
 		<Logger name="org.apache.kafka" level="warn"/>
-		<Logger name="org.springframework.kafka" level="debug"/>
-		<Logger name="org.springframework.integration" level="debug"/>
+		<Logger name="org.springframework.kafka" level="warn"/>
+		<Logger name="org.springframework.integration" level="warn"/>
 		<Root level="warn">
 			<AppenderRef ref="STDOUT" />
 		</Root>