Skip to content

Commit e547d0a

Browse files
nachomdosobychacko
nachomdo
authored and
sobychacko
committed
Kafka Streams sample demo for TopologyTestDriver
Fixes: #1853 * Demonstrating Kafka Streams `TopologyTestDriver` * README updates in the sample * Update Spring Boot parent version of the samples to 3.2.5
1 parent 4439bf3 commit e547d0a

File tree

14 files changed

+385
-10
lines changed

14 files changed

+385
-10
lines changed

‎samples/README.adoc

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
== Samples
22

3-
* sample-01 - simple producer/consumer with dead-letter topic
4-
* sample-02 - multi-method listener
5-
* sample-03 - transactions
6-
* sample-04 - topic based (non-blocking) retry
7-
* sample-05 - global embedded Kafka testing
3+
* sample-01 - Simple producer/consumer with dead-letter topic
4+
* sample-02 - Multi-method listener
5+
* sample-03 - Transactions
6+
* sample-04 - Topic based (non-blocking) retry
7+
* sample-05 - Global embedded Kafka testing
8+
* sample-06 - Kafka Streams tests with TopologyTestDriver

‎samples/sample-01/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
<parent>
1515
<groupId>org.springframework.boot</groupId>
1616
<artifactId>spring-boot-starter-parent</artifactId>
17-
<version>3.2.1</version>
17+
<version>3.2.5</version>
1818
<relativePath/> <!-- lookup parent from repository -->
1919
</parent>
2020

‎samples/sample-02/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
<parent>
1515
<groupId>org.springframework.boot</groupId>
1616
<artifactId>spring-boot-starter-parent</artifactId>
17-
<version>3.2.1</version>
17+
<version>3.2.5</version>
1818
<relativePath/> <!-- lookup parent from repository -->
1919
</parent>
2020

‎samples/sample-03/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
<parent>
1515
<groupId>org.springframework.boot</groupId>
1616
<artifactId>spring-boot-starter-parent</artifactId>
17-
<version>3.2.1</version>
17+
<version>3.2.5</version>
1818
<relativePath/> <!-- lookup parent from repository -->
1919
</parent>
2020

‎samples/sample-04/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
<parent>
1515
<groupId>org.springframework.boot</groupId>
1616
<artifactId>spring-boot-starter-parent</artifactId>
17-
<version>3.2.1</version>
17+
<version>3.2.5</version>
1818
<relativePath/> <!-- lookup parent from repository -->
1919
</parent>
2020

‎samples/sample-05/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
<parent>
66
<groupId>org.springframework.boot</groupId>
77
<artifactId>spring-boot-starter-parent</artifactId>
8-
<version>3.2.1</version>
8+
<version>3.2.5</version>
99
<relativePath/> <!-- lookup parent from repository -->
1010
</parent>
1111

‎samples/sample-06/README.adoc

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
== Sample 6
2+
3+
This sample demonstrates a simple Kafka Streams topology tested with TopologyTestDriver.
4+
5+
The application contains a simple Kafka Streams topology that counts the keys seen so far in a stateful manner.
6+
The corresponding `TopologyTestDriver` based JUnit test verifies the behavior of the business logic in the Kafka Streams topology.
7+
8+
9+
Console output describe the topology as shown below:
10+
11+
. ____ _ __ _ _
12+
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
13+
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
14+
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
15+
' |____| .__|_| |_|_| |_\__, | / / / /
16+
=========|_|==============|___/=/_/_/_/
17+
:: Spring Boot :: (v2.5.2)
18+
19+
2021-06-30 17:38:33.637 INFO 92063 --- [ main] com.example.ApplicationTests : Starting ApplicationTests using Java 11.0.10 on C02FL1KSMD6T with PID 92063 (started by igomez in /Users/igomez/Projects/spring-kafka/samples/sample-05)
20+
2021-06-30 17:38:33.638 INFO 92063 --- [ main] com.example.ApplicationTests : The following profiles are active: test
21+
2021-06-30 17:38:35.027 INFO 92063 --- [ main] com.example.ApplicationTests : Started ApplicationTests in 1.73 seconds (JVM running for 2.833)
22+
2021-06-30 17:38:35.695 INFO 92063 --- [ main] com.example.ApplicationTests : Topologies:
23+
Sub-topology: 0
24+
Source: KSTREAM-SOURCE-0000000000 (topics: [input])
25+
--> KSTREAM-AGGREGATE-0000000002
26+
Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])
27+
--> KTABLE-SUPPRESS-0000000003
28+
<-- KSTREAM-SOURCE-0000000000
29+
Processor: KTABLE-SUPPRESS-0000000003 (stores: [KTABLE-SUPPRESS-STATE-STORE-0000000004])
30+
--> KTABLE-TOSTREAM-0000000005
31+
<-- KSTREAM-AGGREGATE-0000000002
32+
Processor: KTABLE-TOSTREAM-0000000005 (stores: [])
33+
--> KSTREAM-SINK-0000000006
34+
<-- KTABLE-SUPPRESS-0000000003
35+
Sink: KSTREAM-SINK-0000000006 (topic: output)
36+
<-- KTABLE-TOSTREAM-0000000005

‎samples/sample-06/pom.xml

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
4+
<modelVersion>4.0.0</modelVersion>
5+
6+
<groupId>com.example</groupId>
7+
<artifactId>kafka-sample-06</artifactId>
8+
<version>3.2.0-SNAPSHOT</version>
9+
<packaging>jar</packaging>
10+
11+
<name>kafka-sample-06</name>
12+
<description>Kafka Sample 6</description>
13+
<parent>
14+
<groupId>org.springframework.boot</groupId>
15+
<artifactId>spring-boot-starter-parent</artifactId>
16+
<version>3.2.5</version>
17+
<relativePath/> <!-- lookup parent from repository -->
18+
</parent>
19+
20+
<properties>
21+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
22+
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
23+
<java.version>17</java.version>
24+
</properties>
25+
26+
<dependencies>
27+
28+
<dependency>
29+
<groupId>org.springframework.kafka</groupId>
30+
<artifactId>spring-kafka</artifactId>
31+
</dependency>
32+
33+
<dependency>
34+
<groupId>org.apache.kafka</groupId>
35+
<artifactId>kafka-streams</artifactId>
36+
</dependency>
37+
38+
<dependency>
39+
<groupId>org.springframework.boot</groupId>
40+
<artifactId>spring-boot-starter-test</artifactId>
41+
<scope>test</scope>
42+
</dependency>
43+
44+
<dependency>
45+
<groupId>org.springframework.boot</groupId>
46+
<artifactId>spring-boot-starter-web</artifactId>
47+
</dependency>
48+
49+
<dependency>
50+
<groupId>org.apache.kafka</groupId>
51+
<artifactId>kafka-streams-test-utils</artifactId>
52+
<scope>test</scope>
53+
</dependency>
54+
55+
<dependency>
56+
<groupId>org.apache.kafka</groupId>
57+
<artifactId>kafka-clients</artifactId>
58+
<scope>test</scope>
59+
<classifier>test</classifier>
60+
</dependency>
61+
62+
<dependency>
63+
<groupId>org.awaitility</groupId>
64+
<artifactId>awaitility</artifactId>
65+
<scope>test</scope>
66+
</dependency>
67+
</dependencies>
68+
69+
<build>
70+
<plugins>
71+
<plugin>
72+
<groupId>org.springframework.boot</groupId>
73+
<artifactId>spring-boot-maven-plugin</artifactId>
74+
</plugin>
75+
</plugins>
76+
</build>
77+
78+
<repositories>
79+
<repository>
80+
<id>spring-snapshots</id>
81+
<name>Spring Snapshots</name>
82+
<url>https://repo.spring.io/libs-snapshot-local</url>
83+
</repository>
84+
<repository>
85+
<id>spring-milestones</id>
86+
<name>Spring milestones</name>
87+
<url>https://repo.spring.io/libs-milestone-local</url>
88+
</repository>
89+
<repository>
90+
<id>rsocket-snapshots</id>
91+
<name>RSocket Snapshots</name>
92+
<url>https://oss.jfrog.org/oss-snapshot-local</url>
93+
<snapshots>
94+
<enabled>true</enabled>
95+
</snapshots>
96+
</repository>
97+
<repository>
98+
<id>spring-releases</id>
99+
<name>Spring Releases</name>
100+
<url>https://repo.spring.io/release</url>
101+
</repository>
102+
</repositories>
103+
<pluginRepositories>
104+
<pluginRepository>
105+
<id>spring-snapshots</id>
106+
<name>Spring Snapshots</name>
107+
<url>https://repo.spring.io/snapshot</url>
108+
<snapshots>
109+
<enabled>true</enabled>
110+
</snapshots>
111+
</pluginRepository>
112+
<pluginRepository>
113+
<id>spring-milestones</id>
114+
<name>Spring Milestones</name>
115+
<url>https://repo.spring.io/milestone</url>
116+
<snapshots>
117+
<enabled>false</enabled>
118+
</snapshots>
119+
</pluginRepository>
120+
<pluginRepository>
121+
<id>spring-releases</id>
122+
<name>Spring Releases</name>
123+
<url>https://repo.spring.io/release</url>
124+
</pluginRepository>
125+
</pluginRepositories>
126+
127+
128+
</project>

‎samples/sample-06/src/main/java/com/example/Application.java

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
/*
2+
* Copyright 2024 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.example;
18+
19+
import org.slf4j.Logger;
20+
import org.slf4j.LoggerFactory;
21+
22+
import org.springframework.boot.SpringApplication;
23+
import org.springframework.boot.autoconfigure.SpringBootApplication;
24+
import org.springframework.kafka.annotation.EnableKafkaStreams;
25+
26+
/**
27+
*
28+
* @author Nacho Munoz
29+
* @since 3.2.0
30+
*/
31+
@EnableKafkaStreams
32+
@SpringBootApplication
33+
public class Application {
34+
35+
private final Logger logger = LoggerFactory.getLogger(Application.class);
36+
37+
public static void main(String[] args) {
38+
SpringApplication.run(Application.class, args);
39+
}
40+
41+
}

‎samples/sample-06/src/main/java/com/example/Topology.java

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
/*
2+
* Copyright 2024 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.example;
18+
19+
import java.time.Duration;
20+
21+
import org.apache.kafka.common.serialization.Serdes;
22+
import org.apache.kafka.streams.StreamsBuilder;
23+
import org.apache.kafka.streams.kstream.Consumed;
24+
import org.apache.kafka.streams.kstream.Suppressed;
25+
26+
import org.springframework.beans.factory.annotation.Autowired;
27+
import org.springframework.beans.factory.annotation.Value;
28+
import org.springframework.context.annotation.Configuration;
29+
import org.springframework.stereotype.Component;
30+
31+
32+
/**
33+
* A basic topology that counts records by key and materialises the output into a new topic
34+
*
35+
* @author Nacho Munoz
36+
* @author Soby Chacko
37+
* @since 3.2.0
38+
*/
39+
@Configuration
40+
@Component
41+
public class Topology {
42+
private final String inputTopic;
43+
44+
private final String outputTopic;
45+
46+
@Autowired
47+
public Topology(@Value("${input-topic.name}") final String inputTopic,
48+
@Value("${output-topic.name}") final String outputTopic) {
49+
this.inputTopic = inputTopic;
50+
this.outputTopic = outputTopic;
51+
}
52+
53+
@Autowired
54+
public void defaultTopology(final StreamsBuilder builder) {
55+
builder.stream(inputTopic, Consumed.with(Serdes.Integer(), Serdes.String()))
56+
.groupByKey()
57+
.count()
58+
.suppress(Suppressed.untilTimeLimit(Duration.ofMillis(5), Suppressed.BufferConfig.unbounded()))
59+
.toStream()
60+
.to(outputTopic);
61+
62+
}
63+
64+
}

0 commit comments

Comments
 (0)