diff --git a/kafka-streams-samples/kafka-streams-word-count/pom.xml b/kafka-streams-samples/kafka-streams-word-count/pom.xml
index 4c7e5f6..e249bd1 100644
--- a/kafka-streams-samples/kafka-streams-word-count/pom.xml
+++ b/kafka-streams-samples/kafka-streams-word-count/pom.xml
@@ -31,6 +31,11 @@
spring-boot-starter-test
test
+
+ org.springframework.kafka
+ spring-kafka-test
+ test
+
diff --git a/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/KafkaStreamsWordCountApplicationTests.java b/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/KafkaStreamsWordCountApplicationTests.java
index 7b88f15..be7d4c8 100644
--- a/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/KafkaStreamsWordCountApplicationTests.java
+++ b/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/KafkaStreamsWordCountApplicationTests.java
@@ -1,18 +1,94 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package kafka.streams.word.count;
-import org.junit.Ignore;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.junit4.SpringRunner;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
@RunWith(SpringRunner.class)
-@SpringBootTest
+@SpringBootTest(
+ webEnvironment = SpringBootTest.WebEnvironment.NONE,
+ properties = {"server.port=0",
+ "spring.jmx.enabled=false",
+ "spring.cloud.stream.bindings.input.destination=words",
+ "spring.cloud.stream.bindings.output.destination=counts",
+ "spring.cloud.stream.bindings.output.contentType=application/json",
+ "spring.cloud.stream.kafka.streams.default.consumer.application-id=basic-word-count",
+ "spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+ "spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
+ "spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde"})
public class KafkaStreamsWordCountApplicationTests {
+ @ClassRule
+ public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, "words", "counts");
+
+ private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
+
+ private static Consumer consumer;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka);
+ consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+ consumer = cf.createConsumer();
+ embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts");
+
+ System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers", embeddedKafka.getBrokersAsString());
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ consumer.close();
+ System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers");
+ }
+
@Test
- @Ignore
- public void contextLoads() {
+ public void testKstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer() throws Exception {
+ Map senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+ DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps);
+ try {
+ KafkaTemplate template = new KafkaTemplate<>(pf, true);
+ template.setDefaultTopic("words");
+ template.sendDefault("foobar");
+ ConsumerRecords cr = KafkaTestUtils.getRecords(consumer);
+ assertThat(cr.count()).isGreaterThanOrEqualTo(1);
+ }
+ finally {
+ pf.destroy();
+ }
}
}