From 61e7936978180e194d0ce67c97e4dc2d81cdbdfb Mon Sep 17 00:00:00 2001 From: Henryk Konsek Date: Thu, 11 May 2017 16:27:53 +0200 Subject: [PATCH] Add key expression support to Kafka producer Fix #134 - Add `messageKeyExpression` producer property. Added key expression unit test. Added messageKeyExpression documentation. --- .../properties/KafkaProducerProperties.java | 15 ++++++++- .../src/main/asciidoc/overview.adoc | 4 +++ .../kafka/KafkaMessageChannelBinder.java | 2 ++ .../stream/binder/kafka/KafkaBinderTests.java | 31 +++++++++++++++++++ 4 files changed, 51 insertions(+), 1 deletion(-) 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 df0c78ebb..8cc961309 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder.kafka.properties; +import org.springframework.expression.Expression; + import java.util.HashMap; import java.util.Map; @@ -23,6 +25,7 @@ import javax.validation.constraints.NotNull; /** * @author Marius Bogoevici + * @author Henryk Konsek */ public class KafkaProducerProperties { @@ -34,6 +37,8 @@ public class KafkaProducerProperties { private int batchTimeout; + private Expression messageKeyExpression; + private Map configuration = new HashMap<>(); public int getBufferSize() { @@ -69,6 +74,14 @@ public class KafkaProducerProperties { this.batchTimeout = batchTimeout; } + public Expression getMessageKeyExpression() { + return messageKeyExpression; + } + + public void setMessageKeyExpression(Expression messageKeyExpression) { + this.messageKeyExpression = messageKeyExpression; + } + public Map getConfiguration() { return this.configuration; } diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 970b8d581..3414631b5 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -190,6 +190,10 @@ batchTimeout:: (Normally the producer does not wait at all, and simply sends all the messages that accumulated while the previous send was in progress.) A non-zero value may increase throughput at the expense of latency. + Default: `0`. +messageKeyExpression:: +Expression (executed against incoming message) used to resolve the key of the produced Kafka message. For example `headers.key` or `payload.myKey`. ++ +Default: `none`. configuration:: Map with a key/value pair containing generic Kafka producer properties. + 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 a25fc040d..aa3800820 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 @@ -84,6 +84,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Gary Russell * @author Mark Fisher * @author Soby Chacko + * @author Henryk Konsek */ public class KafkaMessageChannelBinder extends AbstractMessageChannelBinder, @@ -377,6 +378,7 @@ public class KafkaMessageChannelBinder extends DefaultKafkaProducerFactory producerFactory) { super(kafkaTemplate); setTopicExpression(new LiteralExpression(topic)); + setMessageKeyExpression(producerProperties.getExtension().getMessageKeyExpression()); setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory()); if (producerProperties.isPartitioned()) { SpelExpressionParser parser = new SpelExpressionParser(); 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 c9c40cdcf..834c7c259 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 @@ -32,6 +32,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.LongDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; import org.assertj.core.api.Condition; import org.junit.Ignore; import org.junit.Test; @@ -81,6 +82,7 @@ import static org.junit.Assert.assertTrue; /** * @author Soby Chacko * @author Ilayaperumal Gopinathan + * @author Henryk Konsek */ public abstract class KafkaBinderTests extends PartitionCapableBinderTests, ExtendedProducerProperties> { @@ -464,6 +466,35 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests producerProperties = createProducerProperties(); + producerProperties.getExtension().getConfiguration().put("key.serializer", StringSerializer.class.getName()); + producerProperties.getExtension().setMessageKeyExpression(spelExpressionParser.parseExpression("headers.key")); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + String uniqueBindingId = UUID.randomUUID().toString(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + Binding producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", + moduleOutputChannel, producerProperties); + Binding consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, + moduleInputChannel, consumerProperties); + Thread.sleep(1000); + Message message = MessageBuilder.withPayload("somePayload").setHeader("key", "myDynamicKey").build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + Message inbound = receive(moduleInputChannel); + assertThat(inbound).isNotNull(); + String receivedKey = new String(inbound.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, byte[].class)); + assertThat(receivedKey).isEqualTo("myDynamicKey"); + producerBinding.unbind(); + consumerBinding.unbind(); + } + @Test @SuppressWarnings("unchecked") public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception {