GH-723: Relax reply template generic types

Resolves #723

The replying template might have different generic types for the Key/Value
than the incoming container.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -97,7 +97,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private boolean batchListener;
 
-	private KafkaTemplate<K, V> replyTemplate;
+	private KafkaTemplate<?, ?> replyTemplate;
 
 	private String clientIdPrefix;
 
@@ -252,11 +252,11 @@ public void setBatchListener(boolean batchListener) {
 	 * @param replyTemplate the template.
 	 * @since 2.0
 	 */
-	public void setReplyTemplate(KafkaTemplate<K, V> replyTemplate) {
+	public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 		this.replyTemplate = replyTemplate;
 	}
 
-	protected KafkaTemplate<K, V> getReplyTemplate() {
+	protected KafkaTemplate<?, ?> getReplyTemplate() {
 		return this.replyTemplate;
 	}
 

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

@@ -101,7 +101,8 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 
 	private Expression replyTopicExpression;
 
-	private KafkaTemplate<K, V> replyTemplate;
+	@SuppressWarnings("rawtypes")
+	private KafkaTemplate replyTemplate;
 
 	private boolean hasAckParameter;
 
@@ -193,7 +194,7 @@ public void setReplyTopic(String replyTopicParam) {
 	 * @param replyTemplate the template.
 	 * @since 2.0
 	 */
-	public void setReplyTemplate(KafkaTemplate<K, V> replyTemplate) {
+	public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 		this.replyTemplate = replyTemplate;
 	}
 
@@ -390,7 +391,7 @@ else if (result instanceof Message) {
 					this.replyTemplate.send(builder.build());
 				}
 				else {
-					this.replyTemplate.send(topic, (V) result);
+					this.replyTemplate.send(topic, result);
 				}
 			}
 		}