GH-423 Add option to use KafkaHeaders.TOPIC Header

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/423

Add an option to override the default binding destination with the value
of the header, if present.

docs/src/main/asciidoc/overview.adoc

@@ -327,7 +327,11 @@ The replication factor to use when provisioning topics. Overrides the binder-wid
 Ignored if `replicas-assignments` is present.
 +
 Default: none (the binder-wide default of 1 is used).
-
+useTopicHeader::
+Set to `true` to override the default binding destination (topic name) with the value of the `KafkaHeaders.TOPIC` message header in the outbound message.
+If the header is not present, the default binding destination is used.
+Default: `false`.
++
 
 NOTE: The Kafka binder uses the `partitionCount` setting of the producer as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used).
 Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value is used.

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java

@@ -50,6 +50,8 @@ public class KafkaProducerProperties {
 
 	private KafkaTopicProperties topic = new KafkaTopicProperties();
 
+	private boolean useTopicHeader;
+
 	public int getBufferSize() {
 		return this.bufferSize;
 	}
@@ -138,6 +140,14 @@ public class KafkaProducerProperties {
 		this.topic = topic;
 	}
 
+	public boolean isUseTopicHeader() {
+		return this.useTopicHeader;
+	}
+
+	public void setUseTopicHeader(boolean useTopicHeader) {
+		this.useTopicHeader = useTopicHeader;
+	}
+
 	/**
 	 * Enumeration for compression types.
 	 */

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -181,6 +181,8 @@ public class KafkaMessageChannelBinder extends
 
 	private static final ThreadLocal<String> bindingNameHolder = new ThreadLocal<>();
 
+	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
+
 	private final KafkaBinderConfigurationProperties configurationProperties;
 
 	private final Map<String, TopicInformation> topicsInUse = new ConcurrentHashMap<>();
@@ -1127,7 +1129,7 @@ public class KafkaMessageChannelBinder extends
 	}
 
 	private final class ProducerConfigurationMessageHandler
-			extends KafkaProducerMessageHandler<byte[], byte[]> implements Lifecycle {
+			extends KafkaProducerMessageHandler<byte[], byte[]> {
 
 		private boolean running = true;
 
@@ -1137,14 +1139,18 @@ public class KafkaMessageChannelBinder extends
 				String topic,
 				ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
 				ProducerFactory<byte[], byte[]> producerFactory) {
+
 			super(kafkaTemplate);
-			setTopicExpression(new LiteralExpression(topic));
-			setMessageKeyExpression(
-					producerProperties.getExtension().getMessageKeyExpression());
+			if (producerProperties.getExtension().isUseTopicHeader()) {
+				setTopicExpression(PARSER.parseExpression("headers['" + KafkaHeaders.TOPIC + "'] ?: '" + topic + "'"));
+			}
+			else {
+				setTopicExpression(new LiteralExpression(topic));
+			}
+			setMessageKeyExpression(producerProperties.getExtension().getMessageKeyExpression());
 			setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
 			if (producerProperties.isPartitioned()) {
-				SpelExpressionParser parser = new SpelExpressionParser();
-				setPartitionIdExpression(parser.parseExpression(
+				setPartitionIdExpression(PARSER.parseExpression(
 						"headers['" + BinderHeaders.PARTITION_HEADER + "']"));
 			}
 			if (producerProperties.getExtension().isSync()) {

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -1183,38 +1183,57 @@ public class KafkaBinderTests extends
 
 		QueueChannel moduleInputChannel = new QueueChannel();
 
+		ExtendedProducerProperties<KafkaProducerProperties> producer1Props = createProducerProperties();
+		producer1Props.getExtension().setUseTopicHeader(true);
+
 		Binding<MessageChannel> producerBinding1 = binder.bindProducer("foo.x",
-				moduleOutputChannel1, createProducerProperties());
+				moduleOutputChannel1, producer1Props);
 		Binding<MessageChannel> producerBinding2 = binder.bindProducer("foo.y",
 				moduleOutputChannel2, createProducerProperties());
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
-		Binding<MessageChannel> consumerBinding1 = binder.bindConsumer("foo.x", "test",
+		Binding<MessageChannel> consumerBinding1 = binder.bindConsumer("foo.x", "test1",
 				moduleInputChannel, consumerProperties);
-		Binding<MessageChannel> consumerBinding2 = binder.bindConsumer("foo.y", "test",
+		Binding<MessageChannel> consumerBinding2 = binder.bindConsumer("foo.y", "test2",
 				moduleInputChannel, consumerProperties);
 
-		String testPayload1 = "foo" + UUID.randomUUID().toString();
+		String testPayload1 = "foo1";
 		Message<?> message1 = org.springframework.integration.support.MessageBuilder
 				.withPayload(testPayload1.getBytes()).build();
-		String testPayload2 = "foo" + UUID.randomUUID().toString();
+		String testPayload2 = "foo2";
 		Message<?> message2 = org.springframework.integration.support.MessageBuilder
 				.withPayload(testPayload2.getBytes()).build();
+		String testPayload3 = "foo3";
+		Message<?> message3 = org.springframework.integration.support.MessageBuilder
+				.withPayload(testPayload3.getBytes())
+				.setHeader(KafkaHeaders.TOPIC, "foo.y")
+				.build();
 
 		// Let the consumer actually bind to the producer before sending a msg
 		binderBindUnbindLatency();
 		moduleOutputChannel1.send(message1);
 		moduleOutputChannel2.send(message2);
+		moduleOutputChannel1.send(message3);
 
-		Message<?>[] messages = new Message[2];
+		Message<?>[] messages = new Message[3];
 		messages[0] = receive(moduleInputChannel);
 		messages[1] = receive(moduleInputChannel);
+		messages[2] = receive(moduleInputChannel);
 
 		assertThat(messages[0]).isNotNull();
 		assertThat(messages[1]).isNotNull();
+		assertThat(messages[1]).isNotNull();
 		assertThat(messages).extracting("payload").containsExactlyInAnyOrder(
-				testPayload1.getBytes(), testPayload2.getBytes());
+				testPayload1.getBytes(), testPayload2.getBytes(), testPayload3.getBytes());
+		Arrays.asList(messages).forEach(message -> {
+			if (new String((byte[]) message.getPayload()).equals("foo1")) {
+				assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("foo.x");
+			}
+			else {
+				assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("foo.y");
+			}
+		});
 
 		producerBinding1.unbind();
 		producerBinding2.unbind();