GH-847: Support placeholders in @SendTo

Resolves https://github.com/spring-projects/spring-kafka/issues/847

**cherry-pick to 2.1.x**

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

@@ -22,6 +22,7 @@ import java.util.Arrays;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.MessageListenerContainer;
@@ -122,7 +123,14 @@ public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndp
 					throw new IllegalStateException("Invalid @" + SendTo.class.getSimpleName() + " annotation on '"
 							+ method + "' one destination must be set (got " + Arrays.toString(destinations) + ")");
 				}
-				return destinations.length == 1 ? resolve(destinations[0]) : "";
+				String topic = destinations.length == 1 ? destinations[0] : "";
+				if (getBeanFactory() instanceof ConfigurableListableBeanFactory) {
+					topic = ((ConfigurableListableBeanFactory) getBeanFactory()).resolveEmbeddedValue(topic);
+					if (topic != null) {
+						topic = resolve(topic);
+					}
+				}
+				return topic;
 			}
 		}
 		return null;

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

@@ -79,7 +79,7 @@ public class MultiMethodKafkaListenerEndpoint<K, V> extends MethodKafkaListenerE
 			}
 		}
 		DelegatingInvocableHandler delegatingHandler = new DelegatingInvocableHandler(invocableHandlerMethods,
-				defaultHandler, getBean(), getResolver(), getBeanExpressionContext());
+				defaultHandler, getBean(), getResolver(), getBeanExpressionContext(), getBeanFactory());
 		return new HandlerAdapter(delegatingHandler);
 	}
 

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

@@ -26,8 +26,10 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.BeanExpressionContext;
 import org.springframework.beans.factory.config.BeanExpressionResolver;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.expression.Expression;
@@ -75,6 +77,8 @@ public class DelegatingInvocableHandler {
 
 	private final BeanExpressionContext beanExpressionContext;
 
+	private final ConfigurableListableBeanFactory beanFactory;
+
 	/**
 	 * Construct an instance with the supplied handlers for the bean.
 	 * @param handlers the handlers.
@@ -84,6 +88,7 @@ public class DelegatingInvocableHandler {
 	 */
 	public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers, Object bean,
 			BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext) {
+
 		this(handlers, null, bean, beanExpressionResolver, beanExpressionContext);
 	}
 
@@ -99,11 +104,33 @@ public class DelegatingInvocableHandler {
 	public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
 			@Nullable InvocableHandlerMethod defaultHandler,
 			Object bean, BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext) {
+
+		this(handlers, defaultHandler, bean, beanExpressionResolver, beanExpressionContext, null);
+	}
+
+	/**
+	 * Construct an instance with the supplied handlers for the bean.
+	 * @param handlers the handlers.
+	 * @param defaultHandler the default handler.
+	 * @param bean the bean.
+	 * @param beanExpressionResolver the resolver.
+	 * @param beanExpressionContext the context.
+	 * @param beanFactory the bean factory.
+	 * @since 2.1.11
+	 */
+	public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
+			@Nullable InvocableHandlerMethod defaultHandler,
+			Object bean, BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext,
+			@Nullable BeanFactory beanFactory) {
+
 		this.handlers = new ArrayList<>(handlers);
 		this.defaultHandler = defaultHandler;
 		this.bean = bean;
 		this.resolver = beanExpressionResolver;
 		this.beanExpressionContext = beanExpressionContext;
+		this.beanFactory = beanFactory instanceof ConfigurableListableBeanFactory
+				? (ConfigurableListableBeanFactory) beanFactory
+				: null;
 	}
 
 	/**
@@ -173,7 +200,13 @@ public class DelegatingInvocableHandler {
 				throw new IllegalStateException("Invalid @" + SendTo.class.getSimpleName() + " annotation on '"
 						+ element + "' one destination must be set (got " + Arrays.toString(destinations) + ")");
 			}
-			replyTo = destinations.length == 1 ? resolve(destinations[0]) : null;
+			replyTo = destinations.length == 1 ? destinations[0] : null;
+			if (replyTo != null && this.beanFactory != null) {
+				replyTo = this.beanFactory.resolveEmbeddedValue(replyTo);
+				if (replyTo != null) {
+					replyTo = resolve(replyTo);
+				}
+			}
 		}
 		return replyTo;
 	}

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -141,7 +141,8 @@ public class EnableKafkaIntegrationTests {
 	@ClassRule
 	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true,
 			"annotated1", "annotated2", "annotated3",
-			"annotated4", "annotated5", "annotated6", "annotated7", "annotated8", "annotated9", "annotated10",
+			"annotated4", "annotated5", "annotated6", "annotated7", "annotated8", "annotated8reply",
+			"annotated9", "annotated10",
 			"annotated11", "annotated12", "annotated13", "annotated14", "annotated15", "annotated16", "annotated17",
 			"annotated18", "annotated19", "annotated20", "annotated21", "annotated21reply", "annotated22",
 			"annotated22reply", "annotated23", "annotated23reply", "annotated24", "annotated24reply",
@@ -335,11 +336,19 @@ public class EnableKafkaIntegrationTests {
 
 	@Test
 	public void testMulti() throws Exception {
+		Map<String, Object> consumerProps = new HashMap<>(this.consumerFactory.getConfigurationProperties());
+		consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "testReplying");
+		ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		Consumer<Integer, String> consumer = cf.createConsumer();
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "annotated8reply");
 		template.send("annotated8", 0, 1, "foo");
 		template.send("annotated8", 0, 1, null);
 		template.flush();
 		assertThat(this.multiListener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.multiListener.latch2.await(60, TimeUnit.SECONDS)).isTrue();
+		ConsumerRecord<Integer, String> reply = KafkaTestUtils.getSingleRecord(consumer, "annotated8reply");
+		assertThat(reply.value()).isEqualTo("OK");
+		consumer.close();
 	}
 
 	@Test
@@ -1538,7 +1547,7 @@ public class EnableKafkaIntegrationTests {
 
 		@KafkaListener(id = "replyingListenerWithErrorHandler", topics = "annotated23",
 				errorHandler = "replyErrorHandler")
-		@SendTo("annotated23reply")
+		@SendTo("${foo:annotated23reply}")
 		public String replyingListenerWithErrorHandler(String in) {
 			throw new RuntimeException("return this");
 		}
@@ -1671,8 +1680,10 @@ public class EnableKafkaIntegrationTests {
 		}
 
 		@KafkaHandler
-		public void bar(@Payload(required = false) KafkaNull nul, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key) {
+		@SendTo("#{'${foo:annotated8reply}'}")
+		public String bar(@Payload(required = false) KafkaNull nul, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key) {
 			this.latch2.countDown();
+			return "OK";
 		}
 
 		public void foo(String bar) {

src/reference/asciidoc/kafka.adoc

@@ -1300,7 +1300,9 @@ The `#root` object for the evaluation has 3 properties:
   - request - the inbound `ConsumerRecord` (or `ConsumerRecords` object for a batch listener))
   - source - the `org.springframework.messaging.Message<?>` converted from the `request`.
   - result - the method return result.
-- `@SendTo` (no properties) - this is treated as `!{source.headers['kafka_replyTopic']}` (since version _2.1.3_).
+- `@SendTo` (no properties) - this is treated as `!{source.headers['kafka_replyTopic']}` (since version 2.1.3).
+
+Starting with versions 2.1.11, 2.2.1, property placeholders are resolved within `@SendTo` values.
 
 The result of the expression evaluation must be a `String` representing the topic name.
 
@@ -1312,7 +1314,7 @@ public String replyingListener(String in) {
     ...
 }
 
-@KafkaListener(topics = "annotated22")
+@KafkaListener(topics = "${some.property:annotated22}")
 @SendTo("#{myBean.replyTopic}") // config time SpEL
 public Collection<String> replyingBatchListener(List<String> in) {
     ...