Add KafkaTemplate Request/Reply support

- Add `AsyncKafkaTemplate` with `sendAndReceive()`
- In `@KafkaListener`, echo the `correlationId` header

* Polishing

- increase timeout on good test
- add exception to future on timeout

* Polishing - allow `@SendTo` expression to return `byte[]`.

* More Polishing - add standard REPLY_TOPIC header and optional REPLY_PARTITION header.

* Improve test - wait for reply container assignment.

When documenting this, we should recommend that users don't start sending
until the reply partitions are assigned.

* Polishing; docs; rename to `ReplyingKafkaTemplate` - Async was misleading since KT is also Async

* Polishing - PR Comments

* Polishing - PR Comments and add `SimpleKafkaHeaderMapper`

* Fix package tangle - move new template to its own package.

* Use binary for REPLY_PARTITION header.

* Polishing - PR comments.

* Rebase and final polishing

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -31,6 +31,7 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.SendResult;
@@ -194,6 +195,12 @@ public ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record) {
 	@Override
 	public ListenableFuture<SendResult<K, V>> send(Message<?> message) {
 		ProducerRecord<?, ?> producerRecord = this.messageConverter.fromMessage(message, this.defaultTopic);
+		if (!producerRecord.headers().iterator().hasNext()) { // possibly no Jackson
+			byte[] correlationId = message.getHeaders().get(KafkaHeaders.CORRELATION_ID, byte[].class);
+			if (correlationId != null) {
+				producerRecord.headers().add(KafkaHeaders.CORRELATION_ID, correlationId);
+			}
+		}
 		return doSend((ProducerRecord<K, V>) producerRecord);
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -44,7 +44,7 @@
  * @author Artem Bilan
  */
 public abstract class AbstractMessageListenerContainer<K, V>
-		implements MessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle {
+		implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware {
 
 	/**
 	 * The default {@link SmartLifecycle} phase for listener containers {@value #DEFAULT_PHASE}.

spring-kafka/src/main/java/org/springframework/kafka/listener/GenericMessageListenerContainer.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+/**
+ * Generic message listener container; adds parameters.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 2.1.3
+ *
+ */
+public interface GenericMessageListenerContainer<K, V> extends MessageListenerContainer {
+
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -20,6 +20,8 @@
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.lang.reflect.WildcardType;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -45,13 +47,17 @@
 import org.springframework.kafka.listener.ListenerExecutionFailedException;
 import org.springframework.kafka.listener.MessageListener;
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * An abstract {@link MessageListener} adapter providing the necessary infrastructure
@@ -154,16 +160,22 @@ protected boolean isConsumerRecordList() {
 	/**
 	 * Set the topic to which to send any result from the method invocation.
 	 * May be a SpEL expression {@code !{...}} evaluated at runtime.
-	 * @param replyTopic the topic or expression.
+	 * @param replyTopicParam the topic or expression.
 	 * @since 2.0
 	 */
-	public void setReplyTopic(String replyTopic) {
+	public void setReplyTopic(String replyTopicParam) {
+		String replyTopic = replyTopicParam;
+		if (!StringUtils.hasText(replyTopic)) {
+			replyTopic = PARSER_CONTEXT.getExpressionPrefix() + "source.headers['"
+					+ KafkaHeaders.REPLY_TOPIC + "']" + PARSER_CONTEXT.getExpressionSuffix();
+		}
 		if (replyTopic.contains(PARSER_CONTEXT.getExpressionPrefix())) {
 			this.replyTopicExpression = PARSER.parseExpression(replyTopic, PARSER_CONTEXT);
 		}
 		else {
 			this.replyTopicExpression = new LiteralExpression(replyTopic);
 		}
+
 	}
 
 	/**
@@ -273,7 +285,7 @@ protected void handleResult(Object resultArg, Object request, Object source) {
 		String replyTopic = evaluateReplyTopic(request, source, resultArg);
 		Assert.state(replyTopic == null || this.replyTemplate != null,
 				"a KafkaTemplate is required to support replies");
-		sendResponse(result, replyTopic);
+		sendResponse(result, replyTopic, source);
 	}
 
 	private String evaluateReplyTopic(Object request, Object source, Object result) {
@@ -293,30 +305,85 @@ private String evaluateTopic(Object request, Object source, Object result, Expre
 		}
 		else {
 			Object value = sendTo.getValue(this.evaluationContext, new ReplyExpressionRoot(request, source, result));
-			Assert.state(value instanceof String, "replyTopic expression must evaluate to a String or Address");
+			boolean isByteArray = value instanceof byte[];
+			if (!(value instanceof String || isByteArray)) {
+				throw new IllegalStateException(
+					"replyTopic expression must evaluate to a String or byte[], it is: "
+					+ (value == null ? null : value.getClass().getName()));
+			}
+			if (isByteArray) {
+				return new String((byte[]) value, StandardCharsets.UTF_8);
+			}
 			return (String) value;
 		}
 	}
 
-	@SuppressWarnings("unchecked")
+	/**
+	 * Send the result to the topic.
+	 *
+	 * @param result the result.
+	 * @param topic the topic.
+	 * @deprecated in favor of {@link #sendResponse(Object, String, Object)}.
+	 */
+	@Deprecated
 	protected void sendResponse(Object result, String topic) {
+		sendResponse(result, topic, null);
+	}
+
+	/**
+	 * Send the result to the topic.
+	 *
+	 * @param result the result.
+	 * @param topic the topic.
+	 * @param source the source (input).
+	 * @since 2.1.3
+	 */
+	@SuppressWarnings("unchecked")
+	protected void sendResponse(Object result, String topic, @Nullable Object source) {
 		if (topic == null) {
 			if (this.logger.isDebugEnabled()) {
 				this.logger.debug("No replyTopic to handle the reply: " + result);
 			}
 		}
+		else if (result instanceof Message) {
+			this.replyTemplate.send((Message<?>) result);
+		}
 		else {
 			if (result instanceof Collection) {
 				((Collection<V>) result).forEach(v -> {
 					this.replyTemplate.send(topic, v);
 				});
 			}
 			else {
-				this.replyTemplate.send(topic, (V) result);
+				byte[] correlationId = null;
+				boolean sourceIsMessage = source instanceof Message;
+				if (sourceIsMessage
+						&& ((Message<?>) source).getHeaders().get(KafkaHeaders.CORRELATION_ID) != null) {
+					correlationId = ((Message<?>) source).getHeaders().get(KafkaHeaders.CORRELATION_ID, byte[].class);
+				}
+				if (sourceIsMessage) {
+					MessageBuilder<Object> builder = MessageBuilder.withPayload(result)
+							.setHeader(KafkaHeaders.TOPIC, topic);
+					if (correlationId != null) {
+						builder.setHeader(KafkaHeaders.CORRELATION_ID, correlationId);
+					}
+					setPartition(builder, ((Message<?>) source));
+					this.replyTemplate.send(builder.build());
+				}
+				else {
+					this.replyTemplate.send(topic, (V) result);
+				}
 			}
 		}
 	}
 
+	private void setPartition(MessageBuilder<Object> builder, Message<?> source) {
+		byte[] partitionBytes = source.getHeaders().get(KafkaHeaders.REPLY_PARTITION, byte[].class);
+		if (partitionBytes != null) {
+			builder.setHeader(KafkaHeaders.PARTITION_ID, ByteBuffer.wrap(partitionBytes).getInt());
+		}
+	}
+
 	protected final String createMessagingErrorMessage(String description, Object payload) {
 		return description + "\n"
 				+ "Endpoint handler details:\n"

spring-kafka/src/main/java/org/springframework/kafka/requestreply/CorrelationKey.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.requestreply;
+
+import java.math.BigInteger;
+import java.util.Arrays;
+
+import org.springframework.util.Assert;
+
+/**
+ * Wrapper for byte[] that can be used as a hash key. We could have used BigInteger
+ * instead but this wrapper is much less expensive. We do use a BigInteger in
+ * {@link #toString()} though.
+ *
+ * @author Gary Russell
+ * @since 2.1.3
+ */
+public final class CorrelationKey {
+
+	final private byte[] correlationId;
+
+	private volatile Integer hashCode;
+
+	public CorrelationKey(byte[] correlationId) {
+		Assert.notNull(correlationId, "'correlationId' cannot be null");
+		this.correlationId = correlationId;
+	}
+
+	public byte[] getCorrelationId() {
+		return this.correlationId;
+	}
+
+	@Override
+	public int hashCode() {
+		if (this.hashCode != null) {
+			return this.hashCode;
+		}
+		final int prime = 31;
+		int result = 1;
+		result = prime * result + Arrays.hashCode(this.correlationId);
+		this.hashCode = result;
+		return result;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj) {
+			return true;
+		}
+		if (obj == null) {
+			return false;
+		}
+		if (getClass() != obj.getClass()) {
+			return false;
+		}
+		CorrelationKey other = (CorrelationKey) obj;
+		if (!Arrays.equals(this.correlationId, other.correlationId)) {
+			return false;
+		}
+		return true;
+	}
+
+	@Override
+	public String toString() {
+		return "[" + new BigInteger(this.correlationId) + "]";
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaOperations.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.requestreply;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+/**
+ * Request/reply operations.
+ *
+ * @param <K> the key type.
+ * @param <V> the outbound data type.
+ * @param <R> the reply data type.
+ *
+ * @author Gary Russell
+ * @since 2.1.3
+ *
+ */
+public interface ReplyingKafkaOperations<K, V, R> {
+
+	/**
+	 * Send a request and receive a reply.
+	 * @param record the record to send.
+	 * @return a RequestReplyFuture.
+	 */
+	RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record);
+
+}