diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java index d0977a86..969c705c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java +++ b/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 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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java index 8f42f873..940d4523 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/MultiMethodKafkaListenerEndpoint.java @@ -79,7 +79,7 @@ public class MultiMethodKafkaListenerEndpoint extends MethodKafkaListenerE } } DelegatingInvocableHandler delegatingHandler = new DelegatingInvocableHandler(invocableHandlerMethods, - defaultHandler, getBean(), getResolver(), getBeanExpressionContext()); + defaultHandler, getBean(), getResolver(), getBeanExpressionContext(), getBeanFactory()); return new HandlerAdapter(delegatingHandler); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java index 106a4b42..5edeb9f9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java +++ b/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 handlers, Object bean, BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext) { + this(handlers, null, bean, beanExpressionResolver, beanExpressionContext); } @@ -99,11 +104,33 @@ public class DelegatingInvocableHandler { public DelegatingInvocableHandler(List 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 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; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 74918979..aec1a316 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/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 consumerProps = new HashMap<>(this.consumerFactory.getConfigurationProperties()); + consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "testReplying"); + ConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + Consumer 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 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) { diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 0987b996..b4da8f5f 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/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 replyingBatchListener(List in) { ...