From 3771e2a5e615dc2fc34f200b5d7074b5fe7367f3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 30 Jan 2018 13:40:58 -0500 Subject: [PATCH] Fix @SendTo Runtime SpEL Detection Previously, `setReplyTopic` detected runtime SpEL by checking if the expression starts with `!{`. Since we use a `TemplateParserContext` the topic can be evaluated even if the `!{` is not at the beginning. This is not a breaking change since `!` and `{` are not valid topic characters. --- .../adapter/MessagingMessageListenerAdapter.java | 4 ++-- .../kafka/annotation/EnableKafkaIntegrationTests.java | 10 +++++----- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index a07804ef..776aff1a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.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. @@ -158,7 +158,7 @@ public abstract class MessagingMessageListenerAdapter implements ConsumerS * @since 2.0 */ public void setReplyTopic(String replyTopic) { - if (replyTopic.startsWith(PARSER_CONTEXT.getExpressionPrefix())) { + if (replyTopic.contains(PARSER_CONTEXT.getExpressionPrefix())) { this.replyTopicExpression = PARSER.parseExpression(replyTopic, PARSER_CONTEXT); } else { 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 ca1764c4..5eb4c37f 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 @@ -477,10 +477,10 @@ public class EnableKafkaIntegrationTests { ConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); Consumer consumer = cf.createConsumer(); embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "annotated21reply"); - template.send("annotated21", 0, "annotated21reply"); + template.send("annotated21", 0, "nnotated21reply"); // drop the leading 'a'. template.flush(); ConsumerRecord reply = KafkaTestUtils.getSingleRecord(consumer, "annotated21reply"); - assertThat(reply.value()).isEqualTo("ANNOTATED21REPLY"); + assertThat(reply.value()).isEqualTo("NNOTATED21REPLY"); consumer.close(); } @@ -1019,7 +1019,7 @@ public class EnableKafkaIntegrationTests { private Object listen16Message; - private CountDownLatch listen16ErrorLatch = new CountDownLatch(1); + private final CountDownLatch listen16ErrorLatch = new CountDownLatch(1); @Bean public ConsumerAwareErrorHandler listen16ErrorHandler() { @@ -1290,13 +1290,13 @@ public class EnableKafkaIntegrationTests { } @KafkaListener(id = "replyingListener", topics = "annotated21") - @SendTo("!{request.value()}") // runtime SpEL - test payload is the reply queue + @SendTo("a!{request.value()}") // runtime SpEL - test payload is the reply queue minus leading 'a' public String replyingListener(String in) { return in.toUpperCase(); } @KafkaListener(id = "replyingBatchListener", topics = "annotated22", containerFactory = "batchFactory") - @SendTo("#{'annotated22reply'}") // config time SpEL + @SendTo("a#{'nnotated22reply'}") // config time SpEL public Collection replyingBatchListener(List in) { return in.stream().map(String::toUpperCase).collect(Collectors.toList()); }