GH-946: Detect Unnecessary Replying Template

Resolves #946

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

@@ -21,6 +21,9 @@
 import java.util.Collection;
 import java.util.regex.Pattern;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationEventPublisher;
@@ -35,6 +38,7 @@
 import org.springframework.kafka.listener.GenericErrorHandler;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.adapter.ReplyHeadersConfigurer;
+import org.springframework.kafka.requestreply.ReplyingKafkaOperations;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.retry.RecoveryCallback;
@@ -57,6 +61,8 @@
 public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K, V>, K, V>
 		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, InitializingBean {
 
+	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR protected
+
 	private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null);
 
 	private GenericErrorHandler<?> errorHandler;
@@ -203,6 +209,13 @@ public void setApplicationEventPublisher(ApplicationEventPublisher applicationEv
 	 * @since 2.0
 	 */
 	public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
+		if (replyTemplate instanceof ReplyingKafkaOperations) {
+			this.logger.warn(
+				"The 'replyTemplate' should not be an implementation of 'ReplyingKafkaOperations'; "
+				+ "such implementations are for client-side request/reply operations; here we "
+				+ "are simply sending a reply to an incoming request so the reply container will "
+				+ "never be used and will consume unnecessary resources.");
+		}
 		this.replyTemplate = replyTemplate;
 	}
 

src/reference/asciidoc/kafka.adoc

@@ -735,7 +735,7 @@ Aside from the logs, there was no indication that there was a problem.
 To restore the previous behavior, you canset the property to `false`.
 
 [[using-ConcurrentMessageListenerContainer]]
-====== Using`ConcurrentMessageListenerContainer`
+====== Using `ConcurrentMessageListenerContainer`
 
 The single constructor is similar to the first `KafkaListenerContainer` constructor.
 The following listing shows the constructor's signature: