GH-555: Resolve PPs in EmbeddedKafka topics

Fixes #555

* Update `EmbeddedKafkaContextCustomizer` to resolve property placeholders
* Update `KafkaStreamsTests` to use topic from property
* Add property resolving details to `EmbeddedKafka` javadoc and testing.adoc

* Simple polishing

Author: zacholauson

spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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.
@@ -51,6 +51,7 @@
  *
  * @author Artem Bilan
  * @author Elliot Metsger
+ * @author Zach Olauson
  *
  * @since 1.3
  *
@@ -85,6 +86,8 @@
 	int partitions() default 2;
 
 	/**
+	 * Topics that should be created
+	 * Topics may contain property placeholders, e.g. {@code topics = "${kafka.topic.one:topicOne}"}
 	 * @return the topics to create
 	 */
 	String[] topics() default { };

spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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.
@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.StringReader;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Properties;
 
@@ -40,6 +41,7 @@
  *
  * @author Artem Bilan
  * @author Elliot Metsger
+ * @author Zach Olauson
  *
  * @since 1.3
  */
@@ -57,13 +59,19 @@ public void customizeContext(ConfigurableApplicationContext context, MergedConte
 		ConfigurableListableBeanFactory beanFactory = context.getBeanFactory();
 		Assert.isInstanceOf(DefaultSingletonBeanRegistry.class, beanFactory);
 
+		ConfigurableEnvironment environment = context.getEnvironment();
+
+		String[] topics =
+				Arrays.stream(this.embeddedKafka.topics())
+						.map(environment::resolvePlaceholders)
+						.toArray(String[]::new);
+
 		KafkaEmbedded kafkaEmbedded = new KafkaEmbedded(this.embeddedKafka.count(),
 				this.embeddedKafka.controlledShutdown(),
 				this.embeddedKafka.partitions(),
-				this.embeddedKafka.topics());
+				topics);
 
 		Properties properties = new Properties();
-		ConfigurableEnvironment environment = context.getEnvironment();
 
 		for (String pair : this.embeddedKafka.brokerProperties()) {
 			if (!StringUtils.hasText(pair)) {

spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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.
@@ -26,6 +26,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KeyValue;
@@ -63,6 +64,7 @@
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.util.concurrent.SettableListenableFuture;
 
@@ -71,15 +73,17 @@
  * @author Marius Bogoevici
  * @author Gary Russell
  * @author Elliot Metsger
+ * @author Zach Olauson
  *
  * @since 1.1.4
  */
 @RunWith(SpringRunner.class)
 @DirtiesContext
+@TestPropertySource(properties = "streaming.topic.two=streamingTopic2")
 @EmbeddedKafka(partitions = 1,
 		topics = {
 				KafkaStreamsTests.STREAMING_TOPIC1,
-				KafkaStreamsTests.STREAMING_TOPIC2,
+				"${streaming.topic.two}",
 				KafkaStreamsTests.FOOS },
 		brokerProperties = {
 				"auto.create.topics.enable=${topics.autoCreate:false}",
@@ -89,22 +93,22 @@ public class KafkaStreamsTests {
 
 	static final String STREAMING_TOPIC1 = "streamingTopic1";
 
-	static final String STREAMING_TOPIC2 = "streamingTopic2";
-
 	static final String FOOS = "foos";
 
 	@Autowired
 	private KafkaTemplate<Integer, String> kafkaTemplate;
 
 	@Autowired
-	private SettableListenableFuture<String> resultFuture;
+	private SettableListenableFuture<ConsumerRecord<?, String>> resultFuture;
 
 	@Autowired
 	private StreamsBuilderFactoryBean streamsBuilderFactoryBean;
 
 	@Autowired
 	private KafkaEmbedded kafkaEmbedded;
 
+	@Value("${streaming.topic.two}")
+	private String streamingTopic2;
 
 	@Test
 	public void testKStreams() throws Exception {
@@ -129,11 +133,12 @@ public void testKStreams() throws Exception {
 		this.kafkaTemplate.sendDefault(0, payload2);
 		this.kafkaTemplate.flush();
 
-		String result = resultFuture.get(600, TimeUnit.SECONDS);
+		ConsumerRecord<?, String> result = resultFuture.get(600, TimeUnit.SECONDS);
 
 		assertThat(result).isNotNull();
 
-		assertThat(result).isEqualTo(payload.toUpperCase() + payload2.toUpperCase());
+		assertThat(result.topic()).isEqualTo(streamingTopic2);
+		assertThat(result.value()).isEqualTo(payload.toUpperCase() + payload2.toUpperCase());
 
 		assertThat(stateLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
@@ -152,6 +157,9 @@ public static class KafkaStreamsConfiguration {
 		@Value("${" + KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
 		private String brokerAddresses;
 
+		@Value("${streaming.topic.two}")
+		private String streamingTopic2;
+
 		@Bean
 		public ProducerFactory<Integer, String> producerFactory() {
 			return new DefaultKafkaProducerFactory<>(producerConfigs());
@@ -196,7 +204,7 @@ public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
 					.reduce((value1, value2) -> value1 + value2, Materialized.as("windowStore"))
 					.toStream()
 					.map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
-					.filter((i, s) -> s.length() > 40).to(STREAMING_TOPIC2);
+					.filter((i, s) -> s.length() > 40).to(streamingTopic2);
 
 			stream.print(Printed.toSysOut());
 
@@ -224,12 +232,12 @@ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer,
 		}
 
 		@Bean
-		public SettableListenableFuture<String> resultFuture() {
+		public SettableListenableFuture<ConsumerRecord<?, String>> resultFuture() {
 			return new SettableListenableFuture<>();
 		}
 
-		@KafkaListener(topics = STREAMING_TOPIC2)
-		public void listener(String payload) {
+		@KafkaListener(topics = "${streaming.topic.two}")
+		public void listener(ConsumerRecord<?, String> payload) {
 			resultFuture().set(payload);
 		}
 

src/reference/asciidoc/testing.adoc

@@ -150,17 +150,17 @@ public class KafkaStreamsTests {
 }
 ----
 
-The `brokerProperties` and `brokerPropertiesLocation` attributes of `@EmbeddedKafka` support property placeholder resolutions:
+The `topics`, `brokerProperties`, and `brokerPropertiesLocation` attributes of `@EmbeddedKafka` support property placeholder resolutions:
 [source, java]
 ----
 @TestPropertySource(locations = "classpath:/test.properties")
-@EmbeddedKafka(topics = "any-topic",
+@EmbeddedKafka(topics = { "any-topic", "${kafka.topics.another-topic}" },
         brokerProperties = { "log.dir=${kafka.broker.logs-dir}",
                             "listeners=PLAINTEXT://localhost:${kafka.broker.port}",
                             "auto.create.topics.enable=${kafka.broker.topics-enable:true}" }
         brokerPropertiesLocation = "classpath:/broker.properties")
 ----
-In the example above, the property placeholders `${kafka.broker.logs-dir}` and `${kafka.broker.port}` are resolved from the Spring `Environment`.
+In the example above, the property placeholders `${kafka.topics.another-topic}`, `${kafka.broker.logs-dir}`, and `${kafka.broker.port}` are resolved from the Spring `Environment`.
 In addition the broker properties are loaded from the `broker.properties` classpath resource specified by the `brokerPropertiesLocation`.
 Property placeholders are resolved for the `brokerPropertiesLocation` URL and for any property placeholders found in the resource.
 Properties defined by `brokerProperties` override properties found in `brokerPropertiesLocation`.