GH-296: DSL: Support inline pub/sub subscriptions

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

* * Add generic type to specsto avoid cast to `BroadcastCapableChannel`.

* * Remove unneeded fields in the abstract spec

* Restore log4j config

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

@@ -75,7 +75,7 @@ public abstract class AbstractKafkaChannel extends AbstractMessageChannel {
 			this.template.send(MessageBuilder.fromMessage(message)
 						.setHeader(KafkaHeaders.TOPIC, this.topic)
 						.build())
-					.get(timeout, TimeUnit.MILLISECONDS);
+					.get(timeout < 0 ? Long.MAX_VALUE : timeout, TimeUnit.MILLISECONDS);
 		}
 		catch (@SuppressWarnings("unused") InterruptedException e) {
 			Thread.currentThread().interrupt();

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

@@ -0,0 +1,52 @@
+/*
+ * 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.springframework.integration.channel.BroadcastCapableChannel;
+import org.springframework.integration.dispatcher.BroadcastingDispatcher;
+import org.springframework.integration.dispatcher.MessageDispatcher;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.KafkaOperations;
+
+/**
+ * Publish/subscribe channel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class PublishSubscribeKafkaChannel extends SubscribableKafkaChannel implements BroadcastCapableChannel {
+
+	/**
+	 * 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 PublishSubscribeKafkaChannel(KafkaOperations<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String channelTopic) {
+		super(template, factory, channelTopic);
+	}
+
+	@Override
+	protected MessageDispatcher createDispatcher() {
+		BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true);
+		broadcastingDispatcher.setBeanFactory(getBeanFactory());
+		return broadcastingDispatcher;
+	}
+
+}

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

@@ -21,7 +21,6 @@ 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;
@@ -47,8 +46,6 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 
 	private final KafkaListenerContainerFactory<?> factory;
 
-	private final boolean pubSub;
-
 	private MessageDispatcher dispatcher;
 
 	private MessageListenerContainer container;
@@ -68,23 +65,9 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 	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
@@ -122,16 +105,7 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 
 	@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.dispatcher = createDispatcher();
 		this.container = this.factory.createContainer(this.topic);
 		String groupId = getGroupId();
 		this.container.getContainerProperties().setGroupId(groupId != null ? groupId : getBeanName());
@@ -149,6 +123,12 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 		});
 	}
 
+	protected MessageDispatcher createDispatcher() {
+		UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();
+		unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy());
+		return unicastingDispatcher;
+	}
+
 	@Override
 	public void start() {
 		this.container.start();

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

@@ -23,6 +23,7 @@ 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.PublishSubscribeKafkaChannel;
 import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
 import org.springframework.util.StringUtils;
 
@@ -46,11 +47,12 @@ public class KafkaChannelParser extends AbstractChannelParser {
 		String topic = element.getAttribute("topic");
 		boolean pubSub = "publish-subscribe-channel".equals(element.getLocalName());
 		if (hasFactory) {
-			builder = BeanDefinitionBuilder.genericBeanDefinition(SubscribableKafkaChannel.class);
+			builder = BeanDefinitionBuilder.genericBeanDefinition(pubSub
+					? PublishSubscribeKafkaChannel.class
+					: 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");

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

@@ -18,32 +18,23 @@ 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.
+ * @param <C> the channel 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
+public abstract class AbstractKafkaChannelSpec<S extends AbstractKafkaChannelSpec<S, C>, C extends AbstractKafkaChannel>
+		extends MessageChannelSpec<S, C> {
 
 	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);

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

@@ -545,10 +545,10 @@ public final class Kafka {
 	 * @return the spec.
 	 * @since 3.3
 	 */
-	public static KafkaSubscribableChannelSpec channel(KafkaTemplate<?, ?> template,
+	public static KafkaPointToPointChannelSpec channel(KafkaTemplate<?, ?> template,
 			KafkaListenerContainerFactory<?> containerFactory, String topic) {
 
-		return new KafkaSubscribableChannelSpec(template, containerFactory, topic, false);
+		return new KafkaPointToPointChannelSpec(template, containerFactory, topic);
 	}
 
 	/**
@@ -559,10 +559,10 @@ public final class Kafka {
 	 * @return the spec.
 	 * @since 3.3
 	 */
-	public static KafkaSubscribableChannelSpec publishSubscribeChannel(KafkaTemplate<?, ?> template,
+	public static KafkaPublishSubscribeChannelSpec publishSubscribeChannel(KafkaTemplate<?, ?> template,
 			KafkaListenerContainerFactory<?> containerFactory, String topic) {
 
-		return new KafkaSubscribableChannelSpec(template, containerFactory, topic, true);
+		return new KafkaPublishSubscribeChannelSpec(template, containerFactory, topic);
 	}
 
 	/**

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

@@ -0,0 +1,38 @@
+/*
+ * 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 point to point channel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class KafkaPointToPointChannelSpec extends KafkaSubscribableChannelSpec<SubscribableKafkaChannel> {
+
+	protected KafkaPointToPointChannelSpec(KafkaTemplate<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String topic) {
+
+		this.channel = new SubscribableKafkaChannel(template, factory, topic);
+	}
+
+}

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

@@ -27,10 +27,9 @@ import org.springframework.kafka.core.KafkaTemplate;
  * @since 3.3
  *
  */
-public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec<KafkaPollableChannelSpec> {
+public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec<KafkaPollableChannelSpec, PollableKafkaChannel> {
 
 	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/KafkaPublishSubscribeChannelSpec.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.PublishSubscribeKafkaChannel;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+
+/**
+ * Spec for a publish/subscribe channel backed by a Kafka topic.
+ *
+ * @author Gary Russell
+ * @since 3.3
+ *
+ */
+public class KafkaPublishSubscribeChannelSpec
+		extends KafkaSubscribableChannelSpec<PublishSubscribeKafkaChannel> {
+
+	protected KafkaPublishSubscribeChannelSpec(KafkaTemplate<?, ?> template, KafkaListenerContainerFactory<?> factory,
+			String topic) {
+
+		this.channel = new PublishSubscribeKafkaChannel(template, factory, topic);
+	}
+
+}

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

@@ -17,23 +17,17 @@
 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.
  *
+ * @param <C> the channel type.
+ *
  * @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);
-	}
+public abstract class KafkaSubscribableChannelSpec<C extends SubscribableKafkaChannel>
+		extends AbstractKafkaChannelSpec<KafkaSubscribableChannelSpec<C>, C> {
 
 }

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

@@ -30,6 +30,7 @@ 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.PublishSubscribeKafkaChannel;
 import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
@@ -146,7 +147,7 @@ public class ChannelTests {
 		public SubscribableKafkaChannel pubSub(KafkaTemplate<Integer, String> template,
 				KafkaListenerContainerFactory<?> factory) {
 
-			SubscribableKafkaChannel channel = new SubscribableKafkaChannel(template, factory, "channel.2", true);
+			SubscribableKafkaChannel channel = new PublishSubscribeKafkaChannel(template, factory, "channel.2");
 			channel.setGroupId("channel.2");
 			return channel;
 		}

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

@@ -38,6 +38,7 @@ 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.BroadcastCapableChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
@@ -94,7 +95,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 @DirtiesContext
 @EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3,
 		KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5, KafkaDslTests.TEST_TOPIC6, KafkaDslTests.TEST_TOPIC7,
-		KafkaDslTests.TEST_TOPIC8 })
+		KafkaDslTests.TEST_TOPIC8, KafkaDslTests.TEST_TOPIC9 })
 public class KafkaDslTests {
 
 	static final String TEST_TOPIC1 = "test-topic1";
@@ -113,6 +114,8 @@ public class KafkaDslTests {
 
 	static final String TEST_TOPIC8 = "test-topic8";
 
+	static final String TEST_TOPIC9 = "test-topic9";
+
 	@Autowired
 	@Qualifier("sendToKafkaFlow.input")
 	private MessageChannel sendToKafkaFlowInput;
@@ -156,7 +159,7 @@ public class KafkaDslTests {
 	private Gate gate;
 
 	@Test
-	public void testKafkaAdapters() throws Exception {
+	void testKafkaAdapters() throws Exception {
 		this.sendToKafkaFlowInput.send(new GenericMessage<>("foo", Collections.singletonMap("foo", "bar")));
 
 		assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper);
@@ -217,20 +220,25 @@ public class KafkaDslTests {
 	}
 
 	@Test
-	public void testGateways() throws Exception {
+	void testGateways() throws Exception {
 		assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO");
 	}
 
 	@Test
-	void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel) {
+	void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel,
+			@Autowired PollableKafkaChannel topic9Channel) {
 		topic6Channel.send(new GenericMessage<>("foo"));
 		Message<?> received = topic8Channel.receive();
 		assertThat(received)
 				.isNotNull()
 				.extracting("payload")
 				.isEqualTo("foo");
-	}
+		received = topic9Channel.receive();
+		assertThat(received)
+				.isNotNull()
+				.extracting("payload")
+				.isEqualTo("foo");	}
 
 	@Configuration
 	@EnableIntegration
@@ -379,8 +387,9 @@ public class KafkaDslTests {
 		}
 
 		@Bean
-		public KafkaSubscribableChannelSpec topic6Channel(KafkaTemplate<Integer, String> template,
+		public KafkaPointToPointChannelSpec topic6Channel(KafkaTemplate<Integer, String> template,
 				ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
+
 			return Kafka.channel(template, containerFactory, TEST_TOPIC6);
 		}
 
@@ -400,8 +409,19 @@ public class KafkaDslTests {
 				KafkaMessageSource<?, ?> channelSource) {
 
 			return IntegrationFlows.from(topic6Channel(template, containerFactory))
-					.channel(Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7))
-					.channel(Kafka.pollableChannel(template, channelSource).id("topic8Channel"))
+					.publishSubscribeChannel(pubSub(template, containerFactory), channel -> channel
+							.subscribe(f -> f.channel(
+									Kafka.pollableChannel(template, channelSource).id("topic8Channel")))
+							.subscribe(f -> f.channel(
+									Kafka.pollableChannel(template, channelSource).id("topic9Channel"))))
+					.get();
+		}
+
+		@Bean
+		public BroadcastCapableChannel pubSub(KafkaTemplate<Integer, String> template,
+				ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
+
+			return Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7)
 					.get();
 		}