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.

Author: garyrussell

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 @@ protected boolean isConsumerRecordList() {
 	 * @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 {

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

@@ -477,10 +477,10 @@ public void testReplyingListener() throws Exception {
 		ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
 		Consumer<Integer, String> consumer = cf.createConsumer();
 		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "annotated21reply");
-		template.send("annotated21", 0, "annotated21reply");
+		template.send("annotated21", 0, "nnotated21reply"); // drop the leading 'a'.
 		template.flush();
 		ConsumerRecord<Integer, String> reply = KafkaTestUtils.getSingleRecord(consumer, "annotated21reply");
-		assertThat(reply.value()).isEqualTo("ANNOTATED21REPLY");
+		assertThat(reply.value()).isEqualTo("NNOTATED21REPLY");
 		consumer.close();
 	}
 
@@ -1019,7 +1019,7 @@ public KafkaListenerErrorHandler voidSendToErrorHandler() {
 
 		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 String errorHandler(String data) throws Exception {
 		}
 
 		@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<String> replyingBatchListener(List<String> in) {
 			return in.stream().map(String::toUpperCase).collect(Collectors.toList());
 		}