SpringCloud/spring-cloud-stream-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Cleaning up samples

Browse Source 
Kotlin e2e sample with Boot 2.2.0.RELEASE
Kafka Streams word count functional updates
This commit is contained in:
Soby Chacko
2019-10-24 18:12:54 -04:00
parent 4ec74168d6
commit 594779ee77
10 changed files with 161 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
kafka-e2e-kotlin-sample/customer-service/pom.xml
View File

@@ -11,7 +11,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<version>2.2.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

2
kafka-e2e-kotlin-sample/order-service/pom.xml
View File

@@ -11,7 +11,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<version>2.2.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

2
kafka-e2e-kotlin-sample/shipping-service/pom.xml
View File

@@ -11,7 +11,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<version>2.2.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>

9
kafka-streams-samples/kafka-streams-word-count/README.adoc
View File

@@ -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`

93
kafka-streams-samples/kafka-streams-word-count/pom.xml
View File

@@ -6,26 +6,37 @@
<artifactId>kafka-streams-word-count</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>kafka-streams-word-count</name>
<description>Demo project for Spring Boot</description>
<parent>
<groupId>io.spring.cloud.stream.sample</groupId>
<artifactId>spring-cloud-stream-samples-parent</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../..</relativePath>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
@@ -42,6 +53,19 @@
<version>${kafka.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
<build>
@@ -53,4 +77,55 @@
</plugins>
</build>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>

27
kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/KafkaStreamsWordCountApplication.java
View File

@@ -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<Bytes, WordCount> process(KStream<Bytes, String> input) {
@Bean
public Function<KStream<Bytes, String>, KStream<Bytes, WordCount>> 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()))));

94
kafka-streams-samples/kafka-streams-word-count/src/main/java/kafka/streams/word/count/SampleRunner.java
View File

@@ -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<String> 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();
}
}

46
kafka-streams-samples/kafka-streams-word-count/src/main/resources/application.yml
View File

@@ -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

33
kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/KafkaStreamsWordCountApplicationTests.java
View File

@@ -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<String, String> consumer;
@Autowired
StreamsBuilderFactoryBean streamsBuilderFactoryBean;
@Before
public void before() {
streamsBuilderFactoryBean.setCloseTimeout(0);
}
@BeforeClass
public static void setUp() throws Exception {
public static void setUp() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {

9
kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/WordCountProcessorApplicationTests.java
View File

@@ -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<Bytes, String> input = builder.stream(INPUT_TOPIC, Consumed.with(nullSerde, stringSerde));
KafkaStreamsWordCountApplication.WordCountProcessorApplication app = new KafkaStreamsWordCountApplication.WordCountProcessorApplication();
KStream<Bytes, KafkaStreamsWordCountApplication.WordCount> output = app.process(input);
output.to(OUTPUT_TOPIC, Produced.with(nullSerde, countSerde));
final Function<KStream<Bytes, String>, KStream<Bytes, KafkaStreamsWordCountApplication.WordCount>> process = app.process();
final KStream<Bytes, KafkaStreamsWordCountApplication.WordCount> output = process.apply(input);
output.to(OUTPUT_TOPIC, Produced.with(nullSerde, countSerde));
testDriver = new TopologyTestDriver(builder.build(), getStreamsConfiguration());
}