From 46cbeb1804543ff27bcac6c4c384e8462d116daf Mon Sep 17 00:00:00 2001 From: Walliee Date: Sat, 17 Aug 2019 03:44:56 -0400 Subject: [PATCH] Add hook to specify sendTimeoutExpression resolves #724 --- docs/src/main/asciidoc/overview.adoc | 7 ++++++ .../properties/KafkaProducerProperties.java | 10 ++++++++ .../kafka/KafkaMessageChannelBinder.java | 3 +++ .../stream/binder/kafka/KafkaBinderTests.java | 24 +++++++++++++++++++ 4 files changed, 44 insertions(+) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 20219f6b3..eba2fd4a2 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -290,6 +290,13 @@ sync:: Whether the producer is synchronous. + Default: `false`. +sendTimeoutExpression:: +A SpEL expression evaluated against the outgoing message used to evaluate the time to wait for ack when synchronous publish is enabled -- for example, `headers['mySendTimeout']`. +The value of the timeout is in milliseconds. +With versions before 3.0, the payload could not be used unless native encoding was being used because, by the time this expression was evaluated, the payload was already in the form of a `byte[]`. +Now, the expression is evaluated before the payload is converted. ++ +Default: `none`. batchTimeout:: How long the producer waits to allow more messages to accumulate in the same batch before sending the messages. (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. 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 0e1db7204..068c0c8a7 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 @@ -40,6 +40,8 @@ public class KafkaProducerProperties { private boolean sync; + private Expression sendTimeoutExpression; + private int batchTimeout; private Expression messageKeyExpression; @@ -79,6 +81,14 @@ public class KafkaProducerProperties { this.sync = sync; } + public Expression getSendTimeoutExpression() { + return this.sendTimeoutExpression; + } + + public void setSendTimeoutExpression(Expression sendTimeoutExpression) { + this.sendTimeoutExpression = sendTimeoutExpression; + } + public int getBatchTimeout() { return this.batchTimeout; } 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 35d30f366..02f4c500d 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 @@ -1214,6 +1214,9 @@ public class KafkaMessageChannelBinder extends if (producerProperties.getExtension().isSync()) { setSync(true); } + if (producerProperties.getExtension().getSendTimeoutExpression() != null) { + setSendTimeoutExpression(producerProperties.getExtension().getSendTimeoutExpression()); + } this.producerFactory = producerFactory; } 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 143c08b48..d7d3d5b56 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 @@ -93,7 +93,9 @@ import org.springframework.cloud.stream.provisioning.ProvisioningException; import org.springframework.context.ApplicationContext; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -2287,6 +2289,28 @@ public class KafkaBinderTests extends producerBinding.unbind(); } + @Test + @SuppressWarnings("unchecked") + public void testSendTimeoutExpressionProducerMetadata() throws Exception { + Binder binder = getBinder(createConfigurationProperties()); + DirectChannel output = new DirectChannel(); + String testTopicName = UUID.randomUUID().toString(); + ExtendedProducerProperties properties = createProducerProperties(); + properties.getExtension().setSync(true); + SpelExpressionParser parser = new SpelExpressionParser(); + Expression sendTimeoutExpression = parser.parseExpression("5000"); + properties.getExtension().setSendTimeoutExpression(sendTimeoutExpression); + Binding producerBinding = binder.bindProducer(testTopicName, + output, properties); + DirectFieldAccessor accessor = new DirectFieldAccessor( + extractEndpoint(producerBinding)); + KafkaProducerMessageHandler wrappedInstance = (KafkaProducerMessageHandler) accessor + .getWrappedInstance(); + assertThat(new DirectFieldAccessor(wrappedInstance).getPropertyValue("sendTimeoutExpression") + .equals(sendTimeoutExpression)); + producerBinding.unbind(); + } + @Test @SuppressWarnings("unchecked") public void testAutoCreateTopicsDisabledOnBinderStillWorksAsLongAsBrokerCreatesTopic()