diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index a8ecc8b10..5828247c7 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/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. diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java index 174967d64..491ac8f98 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java +++ b/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. */ diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 7e137edf1..1659064dd 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 bindingNameHolder = new ThreadLocal<>(); + private static final SpelExpressionParser PARSER = new SpelExpressionParser(); + private final KafkaBinderConfigurationProperties configurationProperties; private final Map topicsInUse = new ConcurrentHashMap<>(); @@ -1127,7 +1129,7 @@ public class KafkaMessageChannelBinder extends } private final class ProducerConfigurationMessageHandler - extends KafkaProducerMessageHandler implements Lifecycle { + extends KafkaProducerMessageHandler { private boolean running = true; @@ -1137,14 +1139,18 @@ public class KafkaMessageChannelBinder extends String topic, ExtendedProducerProperties producerProperties, ProducerFactory 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()) { diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index f69976e20..475692b5e 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/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 producer1Props = createProducerProperties(); + producer1Props.getExtension().setUseTopicHeader(true); + Binding producerBinding1 = binder.bindProducer("foo.x", - moduleOutputChannel1, createProducerProperties()); + moduleOutputChannel1, producer1Props); Binding producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, createProducerProperties()); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.getExtension().setAutoRebalanceEnabled(false); - Binding consumerBinding1 = binder.bindConsumer("foo.x", "test", + Binding consumerBinding1 = binder.bindConsumer("foo.x", "test1", moduleInputChannel, consumerProperties); - Binding consumerBinding2 = binder.bindConsumer("foo.y", "test", + Binding 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();