GH-296: Add Kafka-backed MessageChannels

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

* * Add DSL support

* * Polishing and XML support

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2020 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.channel;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Abstract MessageChannel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public abstract class AbstractKafkaChannel extends AbstractMessageChannel {
+
+	protected final LogAccessor logger = new LogAccessor(super.logger); // NOSONAR final
+
+	private final KafkaOperations<?, ?> template;
+
+	protected final String topic; // NOSONAR final
+
+	private String groupId;
+
+	/**
+	 * Construct an instance with the provided paramters.
+	 * @param template the template.
+	 * @param topic the topic.
+	 */
+	public AbstractKafkaChannel(KafkaOperations<?, ?> template, String topic) {
+		Assert.notNull(template, "'template' cannot be null");
+		Assert.notNull(topic, "'topic' cannot be null");
+		this.template = template;
+		this.topic = topic;
+	}
+
+	/**
+	 * Set the group id for the consumer; if not set, the bean name will be used.
+	 * @param groupId the group id.
+	 */
+	public void setGroupId(String groupId) {
+		this.groupId = groupId;
+	}
+
+	protected String getGroupId() {
+		return this.groupId;
+	}
+
+	@Override
+	protected boolean doSend(Message<?> message, long timeout) {
+		try {
+			this.template.send(MessageBuilder.fromMessage(message)
+						.setHeader(KafkaHeaders.TOPIC, this.topic)
+						.build())
+					.get(timeout, TimeUnit.MILLISECONDS);
+		}
+		catch (@SuppressWarnings("unused") InterruptedException e) {
+			Thread.currentThread().interrupt();
+			this.logger.debug(() -> "Interrupted while waiting for send result for: " + message);
+			return false;
+		}
+		catch (ExecutionException e) {
+			this.logger.error(e.getCause(), () -> "Interrupted while waiting for send result for: " + message);
+			return false;
+		}
+		catch (TimeoutException e) {
+			this.logger.debug(e, () -> "Timed out while waiting for send result for: " + message);
+			return false;
+		}
+		return true;
+	}
+
+}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java

@@ -0,0 +1,220 @@
+/*
+ * Copyright 2020 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.channel;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.integration.support.management.PollableChannelManagement;
+import org.springframework.integration.support.management.metrics.CounterFacade;
+import org.springframework.integration.support.management.metrics.MetricsCaptor;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
+import org.springframework.util.Assert;
+
+/**
+ * Pollable channel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class PollableKafkaChannel extends AbstractKafkaChannel
+		implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware {
+
+	private final KafkaMessageSource<?, ?> source;
+
+	private CounterFacade receiveCounter;
+
+	private volatile int executorInterceptorsSize;
+
+	/**
+	 * Construct an instance with the provided parameters.
+	 * @param template the template for sending.
+	 * @param source the source for receiving.
+	 */
+	public PollableKafkaChannel(KafkaOperations<?, ?> template, KafkaMessageSource<?, ?> source) {
+		super(template, topic(source));
+		this.source = source;
+		if (source.getConsumerProperties().getGroupId() == null) {
+			String groupId = getGroupId();
+			source.getConsumerProperties().setGroupId(groupId != null ? groupId : getBeanName());
+		}
+	}
+
+	private static String topic(KafkaMessageSource<?, ?> source) {
+		Assert.notNull(source, "'source' cannot be null");
+		Assert.isTrue(source.getConsumerProperties().getTopics().length == 1, "Only one topic is allowed");
+		return source.getConsumerProperties().getTopics()[0];
+	}
+
+	@Override
+	public int getReceiveCount() {
+		return getMetrics().getReceiveCount();
+	}
+
+	@Override
+	public long getReceiveCountLong() {
+		return getMetrics().getReceiveCountLong();
+	}
+
+	@Override
+	public int getReceiveErrorCount() {
+		return getMetrics().getReceiveErrorCount();
+	}
+
+	@Override
+	public long getReceiveErrorCountLong() {
+		return getMetrics().getReceiveErrorCountLong();
+	}
+
+	@Override
+	@Nullable
+	public Message<?> receive() {
+		return doReceive();
+	}
+
+	@Override
+	@Nullable
+	public Message<?> receive(long timeout) {
+		return doReceive();
+	}
+
+	@Nullable
+	protected Message<?> doReceive() {
+		ChannelInterceptorList interceptorList = getIChannelInterceptorList();
+		Deque<ChannelInterceptor> interceptorStack = null;
+		AtomicBoolean counted = new AtomicBoolean();
+		boolean countsEnabled = isCountsEnabled();
+		boolean traceEnabled = isLoggingEnabled() && logger.isTraceEnabled();
+		try {
+			if (traceEnabled) {
+				logger.trace("preReceive on channel '" + this + "'");
+			}
+			if (interceptorList.getInterceptors().size() > 0) {
+				interceptorStack = new ArrayDeque<>();
+				if (!interceptorList.preReceive(this, interceptorStack)) {
+					return null;
+				}
+			}
+			Message<?> message = this.source.receive();
+			if (message != null) {
+				incrementReceiveCounter();
+				message = interceptorList.postReceive(message, this);
+			}
+			interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
+			return message;
+		}
+		catch (RuntimeException ex) {
+			if (countsEnabled && !counted.get()) {
+				incrementReceiveErrorCounter(ex);
+			}
+			interceptorList.afterReceiveCompletion(null, this, ex, interceptorStack);
+			throw ex;
+		}
+	}
+
+	private void incrementReceiveCounter() {
+		MetricsCaptor metricsCaptor = getMetricsCaptor();
+		if (metricsCaptor != null) {
+			if (this.receiveCounter == null) {
+				this.receiveCounter = buildReceiveCounter(metricsCaptor, null);
+			}
+			this.receiveCounter.increment();
+		}
+	}
+
+	private void incrementReceiveErrorCounter(Exception ex) {
+		MetricsCaptor metricsCaptor = getMetricsCaptor();
+		if (metricsCaptor != null) {
+			buildReceiveCounter(metricsCaptor, ex).increment();
+		}
+		getMetrics().afterError();
+	}
+
+	private CounterFacade buildReceiveCounter(MetricsCaptor metricsCaptor, @Nullable Exception ex) {
+		CounterFacade counterFacade = metricsCaptor
+				.counterBuilder(RECEIVE_COUNTER_NAME)
+				.tag("name", getComponentName() == null ? "unknown" : getComponentName())
+				.tag("type", "channel")
+				.tag("result", ex == null ? "success" : "failure")
+				.tag("exception", ex == null ? "none" : ex.getClass().getSimpleName())
+				.description("Messages received")
+				.build();
+		this.meters.add(counterFacade);
+		return counterFacade;
+	}
+
+	@Override
+	public void setInterceptors(List<ChannelInterceptor> interceptors) {
+		super.setInterceptors(interceptors);
+		for (ChannelInterceptor interceptor : interceptors) {
+			if (interceptor instanceof ExecutorChannelInterceptor) {
+				this.executorInterceptorsSize++;
+			}
+		}
+	}
+
+	@Override
+	public void addInterceptor(ChannelInterceptor interceptor) {
+		super.addInterceptor(interceptor);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize++;
+		}
+	}
+
+	@Override
+	public void addInterceptor(int index, ChannelInterceptor interceptor) {
+		super.addInterceptor(index, interceptor);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize++;
+		}
+	}
+
+	@Override
+	public boolean removeInterceptor(ChannelInterceptor interceptor) {
+		boolean removed = super.removeInterceptor(interceptor);
+		if (removed && interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize--;
+		}
+		return removed;
+	}
+
+	@Override
+	@Nullable
+	public ChannelInterceptor removeInterceptor(int index) {
+		ChannelInterceptor interceptor = super.removeInterceptor(index);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize--;
+		}
+		return interceptor;
+	}
+
+	@Override
+	public boolean hasExecutorInterceptors() {
+		return this.executorInterceptorsSize > 0;
+	}
+
+}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2020 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.channel;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.context.Phased;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.dispatcher.BroadcastingDispatcher;
+import org.springframework.integration.dispatcher.MessageDispatcher;
+import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
+import org.springframework.integration.dispatcher.UnicastingDispatcher;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.util.Assert;
+
+/**
+ * Subscribable channel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class SubscribableKafkaChannel extends AbstractKafkaChannel implements SubscribableChannel, SmartLifecycle {
+
+	private static final int DEFAULT_PHASE = Integer.MAX_VALUE / 2; // same as MessageProducerSupport
+
+	private final KafkaListenerContainerFactory<?> factory;
+
+	private final boolean pubSub;
+
+	private MessageDispatcher dispatcher;
+
+	private MessageListenerContainer container;
+
+	private boolean autoStartup = true;
+
+	private int phase = DEFAULT_PHASE;
+
+	private volatile boolean running;
+
+	/**
+	 * Construct an instance with the provided parameters.
+	 * @param template template for sending.
+	 * @param factory factory for creating a container for receiving.
+	 * @param channelTopic the topic.
+	 */
+	public SubscribableKafkaChannel(KafkaOperations<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String channelTopic) {
+
+		this(template, factory, channelTopic, false);
+	}
+
+	/**
+	 * Construct an instance with the provided parameters.
+	 * @param template template for sending.
+	 * @param factory factory for creating a container for receiving.
+	 * @param channelTopic the topic.
+	 * @param pubSub true for a publish/subscribe channel.
+	 */
+	public SubscribableKafkaChannel(KafkaOperations<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String channelTopic, boolean pubSub) {
+
+		super(template, channelTopic);
+		Assert.notNull(factory, "'factory' cannot be null");
+		this.factory = factory;
+		this.pubSub = pubSub;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	/**
+	 * Set the phase.
+	 * @param phase the phase.
+	 * @see Phased
+	 */
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	/**
+	 * Set the auto startup.
+	 * @param autoStartup true to automatically start.
+	 * @see SmartLifecycle
+	 */
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	@Override
+	protected void onInit() {
+		if (this.pubSub) {
+			BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true);
+			broadcastingDispatcher.setBeanFactory(this.getBeanFactory());
+			this.dispatcher = broadcastingDispatcher;
+		}
+		else {
+			UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
+			unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
+			this.dispatcher = unicastingDispatcher;
+		}
+		this.container = this.factory.createContainer(this.topic);
+		String groupId = getGroupId();
+		this.container.getContainerProperties().setGroupId(groupId != null ? groupId : getBeanName());
+		this.container.getContainerProperties().setMessageListener(
+				new RecordMessagingMessageListenerAdapter<Object, Object>(null, null) {
+
+					@Override
+					public void onMessage(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment,
+							Consumer<?, ?> consumer) {
+
+						SubscribableKafkaChannel.this.dispatcher
+								.dispatch(toMessagingMessage(record, acknowledgment, consumer));
+					}
+
+		});
+	}
+
+	@Override
+	public void start() {
+		this.container.start();
+		this.running = true;
+	}
+
+	@Override
+	public void stop() {
+		this.container.stop();
+		this.running = false;
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		this.container.stop(() -> {
+			callback.run();
+			this.running = false;
+		});
+	}
+
+	@Override
+	public boolean subscribe(MessageHandler handler) {
+		return this.dispatcher.addHandler(handler);
+	}
+
+	@Override
+	public boolean unsubscribe(MessageHandler handler) {
+		return this.dispatcher.removeHandler(handler);
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes related to message channels.
+ */
+package org.springframework.integration.kafka.channel;

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

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2020 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.AbstractChannelParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.kafka.channel.PollableKafkaChannel;
+import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for a channel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class KafkaChannelParser extends AbstractChannelParser {
+
+	@Override
+	protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder;
+		String factory = element.getAttribute("container-factory");
+		boolean hasFactory = StringUtils.hasText(factory);
+		String source = element.getAttribute("message-source");
+		boolean hasSource = StringUtils.hasText(source);
+		String template = element.getAttribute("kafka-template");
+		String topic = element.getAttribute("topic");
+		boolean pubSub = "publish-subscribe-channel".equals(element.getLocalName());
+		if (hasFactory) {
+			builder = BeanDefinitionBuilder.genericBeanDefinition(SubscribableKafkaChannel.class);
+			builder.addConstructorArgReference(template);
+			builder.addConstructorArgReference(factory);
+			builder.addConstructorArgValue(topic);
+			builder.addConstructorArgValue(pubSub);
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
+			IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "role");
+		}
+		else if (hasSource) {
+			builder = BeanDefinitionBuilder.genericBeanDefinition(PollableKafkaChannel.class);
+			builder.addConstructorArgReference(template);
+			builder.addConstructorArgReference(source);
+		}
+		else {
+			if (pubSub) {
+				parserContext.getReaderContext().error("A 'container-factory' is required", element);
+			}
+			else {
+				parserContext.getReaderContext().error("Either a 'container-factory' or 'message-source' is required",
+						element);
+			}
+			return null;
+		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "group-id");
+		return builder;
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -35,6 +35,10 @@ public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 		registerBeanDefinitionParser("outbound-gateway", new KafkaOutboundGatewayParser());
 		registerBeanDefinitionParser("inbound-gateway", new KafkaInboundGatewayParser());
 		registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser());
+		KafkaChannelParser channelParser = new KafkaChannelParser();
+		registerBeanDefinitionParser("channel", channelParser);
+		registerBeanDefinitionParser("pollable-channel", channelParser);
+		registerBeanDefinitionParser("publish-subscribe-channel", channelParser);
 	}
 
 }

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2020 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.dsl;
+
+import org.springframework.integration.dsl.MessageChannelSpec;
+import org.springframework.integration.kafka.channel.AbstractKafkaChannel;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ *
+ * Spec for a message channel backed by a Kafka topic.
+ *
+ * @param <S> the spec type.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public abstract class AbstractKafkaChannelSpec<S extends AbstractKafkaChannelSpec<S>>
+		extends MessageChannelSpec<S, AbstractKafkaChannel> {
+
+	protected final KafkaTemplate<?, ?> template; // NOSONAR final
+
+	protected final String topic; // NOSONAR final
+
+	protected String groupId; // NOSONAR
+
+	protected AbstractKafkaChannelSpec(KafkaTemplate<?, ?> template, String topic) {
+		this.template = template;
+		this.topic = topic;
+	}
+
+	@Override
+	public S id(String idToSet) { // NOSONAR - increase visibility
+		return super.id(idToSet);
+	}
+
+	/**
+	 * Set the group id to use on the consumer side.
+	 * @param group the group id.
+	 * @return the spec.
+	 */
+	public S groupId(String group) {
+		this.groupId = group;
+		return _this();
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -22,7 +22,9 @@ import java.util.regex.Pattern;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
@@ -535,6 +537,47 @@ public final class Kafka {
 		return new KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<>(containerSpec, templateSpec);
 	}
 
+	/**
+	 * Create a spec for a subscribable channel with the provided parameters.
+	 * @param template the template.
+	 * @param containerFactory the container factory.
+	 * @param topic the topic.
+	 * @return the spec.
+	 * @since 3.3
+	 */
+	public static KafkaSubscribableChannelSpec channel(KafkaTemplate<?, ?> template,
+			KafkaListenerContainerFactory<?> containerFactory, String topic) {
+
+		return new KafkaSubscribableChannelSpec(template, containerFactory, topic, false);
+	}
+
+	/**
+	 * Create a spec for a publish/subscribe channel with the provided parameters.
+	 * @param template the template.
+	 * @param containerFactory the container factory.
+	 * @param topic the topic.
+	 * @return the spec.
+	 * @since 3.3
+	 */
+	public static KafkaSubscribableChannelSpec publishSubscribeChannel(KafkaTemplate<?, ?> template,
+			KafkaListenerContainerFactory<?> containerFactory, String topic) {
+
+		return new KafkaSubscribableChannelSpec(template, containerFactory, topic, true);
+	}
+
+	/**
+	 * Create a spec for a pollable channel with the provided parameters.
+	 * @param template the template.
+	 * @param source the source.
+	 * @return the spec.
+	 * @since 3.3
+	 */
+	public static KafkaPollableChannelSpec pollableChannel(KafkaTemplate<?, ?> template,
+			KafkaMessageSource<?, ?> source) {
+
+		return new KafkaPollableChannelSpec(template, source);
+	}
+
 	private static <K, V>
 	KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
 			KafkaMessageListenerContainerSpec<K, V> spec, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2020 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.dsl;
+
+import org.springframework.integration.kafka.channel.PollableKafkaChannel;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * Spec for a pollable channel.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec<KafkaPollableChannelSpec> {
+
+	protected KafkaPollableChannelSpec(KafkaTemplate<?, ?> template, KafkaMessageSource<?, ?> source) {
+		super(null, null);
+		this.channel = new PollableKafkaChannel(template, source);
+	}
+
+}

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2020 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.dsl;
+
+import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * Spec for a subscribable channel.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class KafkaSubscribableChannelSpec extends AbstractKafkaChannelSpec<KafkaSubscribableChannelSpec> {
+
+	protected KafkaSubscribableChannelSpec(KafkaTemplate<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String topic, boolean pubSub) {
+
+		super(template, topic);
+		this.channel = new SubscribableKafkaChannel(template, factory, topic, pubSub);
+	}
+
+}

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

@@ -18,6 +18,63 @@
 		]]></xsd:documentation>
 	</xsd:annotation>
 
+	<xsd:element name="channel">
+		<xsd:annotation>
+			<xsd:documentation>
+				Creates a subscribable channel that is backed by a Kafka topic.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="subscribableChannelType">
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="pollable-channel">
+		<xsd:annotation>
+			<xsd:documentation>
+				Creates a pollable channel that is backed by a Kafka topic.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="channelType">
+					<xsd:attribute name="message-source" type="xsd:string" use="required">
+						<xsd:annotation>
+							<xsd:documentation>
+								An 'org.springframework.integration.kafka.inbound.KafkaMessageSource' bean reference for
+								a pollable channel.
+								Mutually exclusive with 'listener-container'.
+							</xsd:documentation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type
+										type="org.springframework.integration.kafka.inbound.KafkaMessageSource"/>
+								</tool:annotation>
+							</xsd:appinfo>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="publish-subscribe-channel">
+		<xsd:annotation>
+			<xsd:documentation>
+				Creates a subscribable pub/sub channel that is backed by a Kafka topic.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="subscribableChannelType">
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:element name="outbound-channel-adapter">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -396,6 +453,56 @@
 		</xsd:restriction>
 	</xsd:simpleType>
 
+	<xsd:complexType name="channelType">
+		<xsd:sequence>
+			<xsd:element name="interceptors" type="integration:channelInterceptorsType" minOccurs="0" maxOccurs="1">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						A list of ChannelInterceptor instances to be applied to this channel.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+		</xsd:sequence>
+		<xsd:attribute name="id" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation>
+					Unique ID for this Message Channel.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attributeGroup ref="kafkaTemplate"/>
+	</xsd:complexType>
+
+	<xsd:complexType name="subscribableChannelType">
+		<xsd:complexContent>
+			<xsd:extension base="channelType">
+				<xsd:attribute name="topic" type="xsd:string" use="required">
+					<xsd:annotation>
+						<xsd:documentation>
+							The topic name.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="container-factory" type="xsd:string" use="required">
+					<xsd:annotation>
+						<xsd:documentation>
+							An 'org.springframework.kafka.config.KafkaListenerContainerFactory' bean reference for a
+							message-driven (subscribable) channel.
+							Mutually exclusive with 'message-source'.
+						</xsd:documentation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type
+									type="org.springframework.kafka.config.KafkaListenerContainerFactory"/>
+							</tool:annotation>
+						</xsd:appinfo>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
 	<xsd:complexType name="outboundType">
 		<xsd:all>
 			<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
@@ -641,6 +748,13 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="group-id">
+			<xsd:annotation>
+				<xsd:documentation>
+					Set the 'group.id' Kafka consumer property.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 	</xsd:attributeGroup>
 
 	<xsd:attributeGroup name="headerMapper">

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2020 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.channnel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.kafka.channel.PollableKafkaChannel;
+import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.ConsumerProperties;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+@SpringJUnitConfig
+@EmbeddedKafka(topics = { "channel.1", "channel.2", "channel.3" }, partitions = 1)
+public class ChannelTests {
+
+	@Test
+	void subscribablePtp(@Autowired SubscribableChannel ptp) throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		AtomicReference<Message<?>> message = new AtomicReference<>();
+		ptp.subscribe(msg -> {
+			message.set(msg);
+			latch.countDown();
+		});
+		Message<?> msg = new GenericMessage<>("foo");
+		ptp.send(msg, 10_000L);
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(message.get().getPayload()).isEqualTo("foo");
+		assertThat(message.get().getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.1");
+	}
+
+	@Test
+	void pubSub(@Autowired SubscribableChannel pubSub) throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(2);
+		AtomicReference<Message<?>> message = new AtomicReference<>();
+		pubSub.subscribe(msg -> {
+			message.set(msg);
+			latch.countDown();
+		});
+		pubSub.subscribe(msg -> {
+			message.set(msg);
+			latch.countDown();
+		});
+		Message<?> msg = new GenericMessage<>("foo");
+		pubSub.send(msg, 10_000L);
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(message.get().getPayload()).isEqualTo("foo");
+		assertThat(message.get().getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.2");
+	}
+
+	@Test
+	void pollable(@Autowired PollableChannel pollable) {
+		Message<?> msg = new GenericMessage<>("foo");
+		pollable.send(msg, 10_000L);
+		Message<?> message = pollable.receive(10_000L);
+		assertThat(message.getPayload()).isEqualTo("foo");
+		assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("channel.3");
+	}
+
+	@Configuration
+	public static class Config {
+
+		@Autowired
+		private EmbeddedKafkaBroker broker;
+
+		@Bean
+		public ProducerFactory<Integer, String> pf() {
+			return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(this.broker));
+		}
+
+		@Bean
+		public ConsumerFactory<Integer, String> cf() {
+			Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("channelTests", "false", this.broker);
+			consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+			return new DefaultKafkaConsumerFactory<>(consumerProps);
+		}
+
+		@Bean
+		public KafkaTemplate<Integer, String> template(ProducerFactory<Integer, String> pf) {
+			return new KafkaTemplate<>(pf);
+		}
+
+		@Bean
+		public ConcurrentKafkaListenerContainerFactory<Integer, String> factory(ConsumerFactory<Integer, String> cf) {
+			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(cf);
+			return factory;
+		}
+
+		@Bean
+		public SubscribableKafkaChannel ptp(KafkaTemplate<Integer, String> template,
+				KafkaListenerContainerFactory<?> factory) {
+
+			SubscribableKafkaChannel channel = new SubscribableKafkaChannel(template, factory, "channel.1");
+			channel.setGroupId("channel.1");
+			return channel;
+		}
+
+		@Bean
+		public SubscribableKafkaChannel pubSub(KafkaTemplate<Integer, String> template,
+				KafkaListenerContainerFactory<?> factory) {
+
+			SubscribableKafkaChannel channel = new SubscribableKafkaChannel(template, factory, "channel.2", true);
+			channel.setGroupId("channel.2");
+			return channel;
+		}
+
+		@Bean
+		public KafkaMessageSource<Integer, String> source(ConsumerFactory<Integer, String> cf) {
+			return new KafkaMessageSource<>(cf, new ConsumerProperties("channel.3"));
+		}
+
+		@Bean
+		public PollableKafkaChannel pollable(KafkaTemplate<Integer, String> template, KafkaMessageSource<?, ?> source) {
+			return new PollableKafkaChannel(template, source);
+		}
+
+	}
+
+}

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

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2020 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 static org.mockito.Mockito.mock;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ImportResource;
+import org.springframework.integration.kafka.channel.PollableKafkaChannel;
+import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.KafkaOperations;
+import org.springframework.kafka.listener.ConsumerProperties;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+@SpringJUnitConfig
+public class ChannelParserTests {
+
+	@Autowired
+	KafkaListenerContainerFactory<?> containerFactory;
+
+	@Autowired
+	KafkaOperations<?, ?> template;
+
+	@Autowired
+	KafkaMessageSource<?, ?> source;
+
+	@Autowired
+	SubscribableKafkaChannel ptp;
+
+	@Autowired
+	PollableKafkaChannel pollable;
+
+	@Autowired
+	SubscribableKafkaChannel pubSub;
+
+	@Test
+	void testParser() {
+		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "topic")).isEqualTo("ptpTopic");
+		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "topic")).isEqualTo("pubSubTopic");
+		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "container")).isNotNull();
+		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "container")).isNotNull();
+		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "template")).isSameAs(this.template);
+		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "template")).isSameAs(this.template);
+		assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "template")).isSameAs(this.template);
+		assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "source")).isSameAs(this.source);
+		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "groupId")).isEqualTo("ptpGroup");
+		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "groupId")).isEqualTo("pubSubGroup");
+		assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "groupId")).isEqualTo("pollableGroup");
+	}
+
+	@Configuration
+	@ImportResource("org/springframework/integration/kafka/config/xml/channels-context.xml")
+	public static class Config {
+
+		@SuppressWarnings("unchecked")
+		@Bean
+		public KafkaListenerContainerFactory<?> containerFactory() {
+			ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(mock(ConsumerFactory.class));
+			return factory;
+		}
+
+		@Bean
+		public KafkaOperations<?, ?> template() {
+			return mock(KafkaOperations.class);
+		}
+
+		@SuppressWarnings("unchecked")
+		@Bean
+		public KafkaMessageSource<?, ?> source() {
+			ConsumerProperties properties = new ConsumerProperties("test");
+			return new KafkaMessageSource<Object, Object>(mock(ConsumerFactory.class), properties, true);
+		}
+
+	}
+
+}

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

@@ -0,0 +1,19 @@
+<?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:channel kafka-template="template" id="ptp" topic="ptpTopic" group-id="ptpGroup"
+		auto-startup="false" container-factory="containerFactory" />
+
+	<int-kafka:pollable-channel kafka-template="template" id="pollable" message-source="source"
+		group-id = "pollableGroup"/>
+
+	<int-kafka:publish-subscribe-channel kafka-template="template" id="pubSub" topic="pubSubTopic"
+		group-id="pubSubGroup" auto-startup="false" container-factory="containerFactory" />
+
+</beans>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -44,7 +44,9 @@ import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Pollers;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
+import org.springframework.integration.kafka.channel.PollableKafkaChannel;
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
+import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
 import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
 import org.springframework.integration.support.MessageBuilder;
@@ -91,7 +93,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 @SpringJUnitConfig
 @DirtiesContext
 @EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3,
-		KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5 })
+		KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5, KafkaDslTests.TEST_TOPIC6, KafkaDslTests.TEST_TOPIC7,
+		KafkaDslTests.TEST_TOPIC8 })
 public class KafkaDslTests {
 
 	static final String TEST_TOPIC1 = "test-topic1";
@@ -104,6 +107,12 @@ public class KafkaDslTests {
 
 	static final String TEST_TOPIC5 = "test-topic5";
 
+	static final String TEST_TOPIC6 = "test-topic6";
+
+	static final String TEST_TOPIC7 = "test-topic7";
+
+	static final String TEST_TOPIC8 = "test-topic8";
+
 	@Autowired
 	@Qualifier("sendToKafkaFlow.input")
 	private MessageChannel sendToKafkaFlowInput;
@@ -213,6 +222,16 @@ public class KafkaDslTests {
 		assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO");
 	}
 
+	@Test
+	void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel) {
+		topic6Channel.send(new GenericMessage<>("foo"));
+		Message<?> received = topic8Channel.receive();
+		assertThat(received)
+				.isNotNull()
+				.extracting("payload")
+				.isEqualTo("foo");
+	}
+
 	@Configuration
 	@EnableIntegration
 	@EnableKafka
@@ -359,6 +378,33 @@ public class KafkaDslTests {
 					.get();
 		}
 
+		@Bean
+		public KafkaSubscribableChannelSpec topic6Channel(KafkaTemplate<Integer, String> template,
+				ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
+			return Kafka.channel(template, containerFactory, TEST_TOPIC6);
+		}
+
+		@Bean
+		public KafkaTemplate<Integer, String> template(ProducerFactory<Integer, String> pf) {
+			return new KafkaTemplate<>(pf);
+		}
+
+		@Bean
+		public KafkaMessageSource<Integer, String> channelSource(ConsumerFactory<Integer, String> cf) {
+			return new KafkaMessageSource<>(cf, new ConsumerProperties(TEST_TOPIC8));
+		}
+
+		@Bean
+		public IntegrationFlow channels(KafkaTemplate<Integer, String> template,
+				ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory,
+				KafkaMessageSource<?, ?> channelSource) {
+
+			return IntegrationFlows.from(topic6Channel(template, containerFactory))
+					.channel(Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7))
+					.channel(Kafka.pollableChannel(template, channelSource).id("topic8Channel"))
+					.get();
+		}
+
 		private GenericMessageListenerContainer<Integer, String> replyContainer() {
 			ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC5);
 			containerProperties.setGroupId("outGate");