Add hook to specify sendTimeoutExpression

resolves #724

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.

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;
 	}

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;
 		}
 

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<KafkaProducerProperties> properties = createProducerProperties();
+		properties.getExtension().setSync(true);
+		SpelExpressionParser parser = new SpelExpressionParser();
+		Expression sendTimeoutExpression = parser.parseExpression("5000");
+		properties.getExtension().setSendTimeoutExpression(sendTimeoutExpression);
+		Binding<MessageChannel> 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()