diff --git a/kafka-e2e-kotlin-sample/customer-service/pom.xml b/kafka-e2e-kotlin-sample/customer-service/pom.xml index 7da8680..0878b55 100644 --- a/kafka-e2e-kotlin-sample/customer-service/pom.xml +++ b/kafka-e2e-kotlin-sample/customer-service/pom.xml @@ -11,7 +11,7 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.RELEASE diff --git a/kafka-e2e-kotlin-sample/order-service/pom.xml b/kafka-e2e-kotlin-sample/order-service/pom.xml index 2cf1eaa..529705e 100644 --- a/kafka-e2e-kotlin-sample/order-service/pom.xml +++ b/kafka-e2e-kotlin-sample/order-service/pom.xml @@ -11,7 +11,7 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.RELEASE diff --git a/kafka-e2e-kotlin-sample/shipping-service/pom.xml b/kafka-e2e-kotlin-sample/shipping-service/pom.xml index c41a6ab..8582ec9 100644 --- a/kafka-e2e-kotlin-sample/shipping-service/pom.xml +++ b/kafka-e2e-kotlin-sample/shipping-service/pom.xml @@ -11,7 +11,7 @@ org.springframework.boot spring-boot-starter-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.RELEASE diff --git a/kafka-streams-samples/kafka-streams-word-count/README.adoc b/kafka-streams-samples/kafka-streams-word-count/README.adoc index e6e1e4e..61e5ffc 100644 --- a/kafka-streams-samples/kafka-streams-word-count/README.adoc +++ b/kafka-streams-samples/kafka-streams-word-count/README.adoc @@ -5,7 +5,6 @@ This is an example of a Spring Cloud Stream processor using Kafka Streams suppor The example is based on the word count application from the https://github.com/confluentinc/examples/blob/3.2.x/kafka-streams/src/main/java/io/confluent/examples/streams/WordCountLambdaExample.java[reference documentation]. It uses a single input and a single output. In essence, the application receives text messages from an input topic and computes word occurrence counts in a configurable time window and report that in an output topic. -This sample uses lambda expressions and thus requires Java 8+. The sample uses a default timewindow of 30 seconds. === Running the app: @@ -24,10 +23,18 @@ Issue the following commands: `docker exec -it kafka-wordcount /opt/kafka/bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic words` +Or if you prefer `kafkacat`: + +`kafkacat -b localhost:9092 -t words -P` + On another terminal: `docker exec -it kafka-wordcount /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic counts` +Or if you prefer `kafkacat`: + +`kafkacat -b localhost:9092 -t counts` + Enter some text in the console producer and watch the output in the console consumer. Once you are done, stop the Kafka cluster: `docker-compose down` \ No newline at end of file diff --git a/kafka-streams-samples/kafka-streams-word-count/pom.xml b/kafka-streams-samples/kafka-streams-word-count/pom.xml index 8899666..8bd0bc8 100644 --- a/kafka-streams-samples/kafka-streams-word-count/pom.xml +++ b/kafka-streams-samples/kafka-streams-word-count/pom.xml @@ -6,26 +6,37 @@ kafka-streams-word-count 0.0.1-SNAPSHOT jar - kafka-streams-word-count Demo project for Spring Boot - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. + org.springframework.boot + spring-boot-starter-parent + 2.2.0.RELEASE + + + Hoxton.BUILD-SNAPSHOT + + + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + org.springframework.cloud spring-cloud-stream-binder-kafka-streams - - org.springframework.cloud - spring-cloud-stream-binder-kafka - org.springframework.boot spring-boot-starter-test @@ -42,6 +53,19 @@ ${kafka.version} test + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + @@ -53,4 +77,55 @@ + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release-local + + false + + + + diff --git a/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java b/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java index 9c96c31..1369b79 100644 --- a/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java +++ b/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java @@ -16,22 +16,23 @@ package kafka.streams.word.count; +import java.time.Duration; +import java.util.Arrays; +import java.util.Date; +import java.util.function.Function; + import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.Serialized; import org.apache.kafka.streams.kstream.TimeWindows; + import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; -import org.springframework.messaging.handler.annotation.SendTo; - -import java.util.Arrays; -import java.util.Date; +import org.springframework.context.annotation.Bean; @SpringBootApplication public class KafkaStreamsWordCountApplication { @@ -40,22 +41,20 @@ public class KafkaStreamsWordCountApplication { SpringApplication.run(KafkaStreamsWordCountApplication.class, args); } - @EnableBinding(KafkaStreamsProcessor.class) public static class WordCountProcessorApplication { public static final String INPUT_TOPIC = "input"; public static final String OUTPUT_TOPIC = "output"; public static final int WINDOW_SIZE_MS = 30000; - @StreamListener(INPUT_TOPIC) - @SendTo(OUTPUT_TOPIC) - public KStream process(KStream input) { + @Bean + public Function, KStream> process() { - return input + return input -> input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) - .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) - .windowedBy(TimeWindows.of(WINDOW_SIZE_MS)) + .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) + .windowedBy(TimeWindows.of(Duration.ofMillis(WINDOW_SIZE_MS))) .count(Materialized.as("WordCounts-1")) .toStream() .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); diff --git a/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/SampleRunner.java b/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/SampleRunner.java deleted file mode 100644 index a33bd91..0000000 --- a/kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/SampleRunner.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * Copyright 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. - * You may obtain a copy of the License at - * - * https://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.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.context.annotation.Bean; -import org.springframework.integration.annotation.InboundChannelAdapter; -import org.springframework.integration.annotation.Poller; -import org.springframework.integration.core.MessageSource; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.support.GenericMessage; - -import java.util.Random; -import java.util.concurrent.atomic.AtomicBoolean; - -/** - * Provides a test source and sink to trigger the kafka streams processor - * and test the output respectively. - * - * @author Soby Chacko - */ -public class SampleRunner { - - //Following code is only used as a test harness. - - //Following source is used as test producer. - @EnableBinding(TestSource.class) - static class TestProducer { - - private AtomicBoolean semaphore = new AtomicBoolean(true); - - private String[] randomWords = new String[]{"foo", "bar", "foobar", "baz", "fox"}; - private Random random = new Random(); - - @Bean - @InboundChannelAdapter(channel = TestSource.OUTPUT, poller = @Poller(fixedDelay = "1000")) - public MessageSource sendTestData() { - return () -> { - int idx = random.nextInt(5); - return new GenericMessage<>(randomWords[idx]); - }; - } - } - - //Following sink is used as test consumer for the above processor. It logs the data received through the processor. - @EnableBinding(TestSink.class) - static class TestConsumer { - - private final Log logger = LogFactory.getLog(getClass()); - - @StreamListener(TestSink.INPUT) - public void receive(String data) { - logger.info("Data received..." + data); - } - } - - interface TestSink { - - String INPUT = "input1"; - - @Input(INPUT) - SubscribableChannel input1(); - - } - - interface TestSource { - - String OUTPUT = "output1"; - - @Output(TestSource.OUTPUT) - MessageChannel output(); - - } -} diff --git a/kafka-streams-samples/kafka-streams-word-count/src/main/resources/application.yml b/kafka-streams-samples/kafka-streams-word-count/src/main/resources/application.yml index 0f5f0a0..848296b 100644 --- a/kafka-streams-samples/kafka-streams-word-count/src/main/resources/application.yml +++ b/kafka-streams-samples/kafka-streams-word-count/src/main/resources/application.yml @@ -1,18 +1,28 @@ -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 - default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde - bindings.input.consumer.application-id: basic-word-count -spring.cloud.stream.bindings.output: - destination: counts -spring.cloud.stream.bindings.input: - destination: words -#For testing -spring.cloud.stream.bindings.input1.destination: counts -spring.cloud.stream.bindings.output1.destination: words -spring.cloud.stream.bindings.input1.binder: kafka -spring.cloud.stream.bindings.output1.binder: kafka - - - +spring.cloud.stream: + bindings: + process-in-0: + destination: words + process-out-0: + destination: counts + kafka: + streams: + binder: + applicationId: hello-word-count-sample + configuration: + commit.interval.ms: 100 + default: + key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde + value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde +#Enable metrics +management: + endpoint: + health: + show-details: ALWAYS + endpoints: + web: + exposure: + include: metrics,health +#Enable logging to debug for spring kafka config +logging: + level: + org.springframework.kafka.config: debug 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 75e6d6b..d4d4b4d 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 @@ -16,15 +16,20 @@ package kafka.streams.word.count; +import java.util.Map; + 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.Before; import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; @@ -33,21 +38,12 @@ 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( 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.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"}) + properties = {"server.port=0"}) public class KafkaStreamsWordCountApplicationTests { @ClassRule @@ -57,25 +53,32 @@ public class KafkaStreamsWordCountApplicationTests { private static Consumer consumer; + @Autowired + StreamsBuilderFactoryBean streamsBuilderFactoryBean; + + @Before + public void before() { + streamsBuilderFactoryBean.setCloseTimeout(0); + } + @BeforeClass - public static void setUp() throws Exception { + public static void setUp() { 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"); - //Since there are both binders present in this app, we resort to the spring kafka broker property. - System.setProperty("spring.kafka.bootstrap-servers", embeddedKafka.getBrokersAsString()); + System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers", embeddedKafka.getBrokersAsString()); } @AfterClass public static void tearDown() { consumer.close(); - System.clearProperty("spring.kafka.bootstrap-servers"); + System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers"); } @Test - public void testKafkaStreamsWordCountProcessor() throws Exception { + public void testKafkaStreamsWordCountProcessor() { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); try { diff --git a/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/WordCountProcessorApplicationTests.java b/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/WordCountProcessorApplicationTests.java index 7bbbd28..86d5faa 100644 --- a/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/WordCountProcessorApplicationTests.java +++ b/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/WordCountProcessorApplicationTests.java @@ -34,6 +34,7 @@ import org.junit.Test; import org.springframework.kafka.support.serializer.JsonSerde; import java.util.*; +import java.util.function.Function; import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; @@ -75,8 +76,12 @@ public class WordCountProcessorApplicationTests { final StreamsBuilder builder = new StreamsBuilder(); KStream input = builder.stream(INPUT_TOPIC, Consumed.with(nullSerde, stringSerde)); KafkaStreamsWordCountApplication.WordCountProcessorApplication app = new KafkaStreamsWordCountApplication.WordCountProcessorApplication(); - KStream output = app.process(input); - output.to(OUTPUT_TOPIC, Produced.with(nullSerde, countSerde)); + final Function, KStream> process = app.process(); + + final KStream output = process.apply(input); + + output.to(OUTPUT_TOPIC, Produced.with(nullSerde, countSerde)); + testDriver = new TopologyTestDriver(builder.build(), getStreamsConfiguration()); }