Naros · Oct 23, 2026 · Oct 16, 2026 · Oct 19, 2026
diff --git a/.github/workflows/kstreams-workflow.yml b/.github/workflows/kstreams-workflow.yml
@@ -5,23 +5,37 @@ on:
     paths:
       - 'kstreams/**'
       - '.github/workflows/kstreams-workflow.yml'
+      - 'scripts/run-example-test.py'
   pull_request:
     paths:
       - 'kstreams/**'
       - '.github/workflows/kstreams-workflow.yml'
+      - 'scripts/run-example-test.py'
       - '.env'
 
 jobs:
   build:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v6
+      - uses: actions/setup-java@v4
+        with:
+          distribution: 'temurin'
+          java-version: '21'
       - name: Cache local Maven repository
         uses: actions/cache@v4
         with:
           path: ~/.m2/repository
-          key: ${{ runner.os }}-maven-${{ hashFiles('kstreams/**/pom.xml') }}
+          key: ${{ runner.os }}-maven-${{ hashFiles('kstreams/**/pom.xml', 'pom.xml') }}
           restore-keys: |
             ${{ runner.os }}-maven-
       - name: Check changes in [kstreams | poc-ddd-aggregates] example
         run: cd kstreams && mvn clean install -f poc-ddd-aggregates/pom.xml -B -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn -Dmaven.wagon.http.pool=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+      - name: Install Python dependencies
+        run: pip install pyyaml requests
+      - name: Run kstreams test
+        run: python scripts/run-example-test.py kstreams
diff --git a/kstreams/docker-compose.yaml b/kstreams/docker-compose.yaml
@@ -28,8 +28,6 @@ services:
 
   connect_source:
     image: quay.io/debezium/connect:${DEBEZIUM_VERSION}
-    build:
-      context: debezium-mongodb
     ports:
      - 8083:8083
     depends_on:
@@ -61,7 +59,8 @@ services:
   aggregator:
     image: debezium/kstreams-aggregator:${DEBEZIUM_VERSION}
     build:
-      context: poc-ddd-aggregates
+      context: ..
+      dockerfile: kstreams/poc-ddd-aggregates/Dockerfile
     environment:
      - JAVA_MAIN_CLASS=io.debezium.examples.aggregation.StreamingAggregatesDDD
      - PARENT_TOPIC=dbserver1.inventory.customers

diff --git a/kstreams/poc-ddd-aggregates/Dockerfile b/kstreams/poc-ddd-aggregates/Dockerfile
@@ -1,41 +1,40 @@
 # 1. Prepare JDK
-FROM fedora:37 as jdk
-
-RUN mkdir /tmp/jdk \
-     && cd /tmp/jdk \
-     && curl -O https://download.java.net/java/GA/jdk11/13/GPL/openjdk-11.0.1_linux-x64_bin.tar.gz \
-     && tar -xvf openjdk-11.0.1_linux-x64_bin.tar.gz
+FROM eclipse-temurin:21-jdk AS jdk
 
 # Run this to get list of needed modules;
 # Not running it as part of the Dockerfile as it needs the app JAR as input, so it'd have to be executed each time
-# jdeps --print-module-deps --class-path poc-ddd-aggregates/target/dependencies/* poc-ddd-aggregates/target/poc-ddd-aggregates-0.1-SNAPSHOT.jar
+# jdeps --ignore-missing-deps --multi-release 21 --print-module-deps poc-ddd-aggregates/target/poc-ddd-aggregates-0.1-SNAPSHOT.jar poc-ddd-aggregates/target/dependencies/*.jar
 
-RUN /tmp/jdk/jdk-11.0.1/bin/jlink \
-    --add-modules java.desktop,java.management,java.naming,java.security.jgss,java.security.sasl,java.sql,jdk.unsupported \
-    --verbose --strip-debug --compress 2 --no-header-files --no-man-pages \
+RUN jlink \
+    --add-modules java.base,java.desktop,java.management,java.security.jgss,java.security.sasl,java.sql,jdk.unsupported \
+    --verbose --strip-debug --compress zip-6 --no-header-files --no-man-pages \
     --output /opt/jre-minimal
 
 # 2. Build
-FROM maven:3.8.6-openjdk-11 as build
+# The build context is the repository root so that the shared parent POM is available;
+# Dockerfile.dockerignore limits the context to the files this build needs.
+FROM maven:3.9-eclipse-temurin-21 AS build
 
-COPY pom.xml /tmp/poc-ddd-aggregates/
+WORKDIR /tmp/build
+COPY pom.xml .
+COPY kstreams/poc-ddd-aggregates/pom.xml kstreams/poc-ddd-aggregates/
 
 # Download all Maven dependencies; unless the POM changes, this step will be cached for future builds
-RUN mvn dependency:go-offline -f /tmp/poc-ddd-aggregates/pom.xml
+RUN mvn dependency:go-offline -f kstreams/poc-ddd-aggregates/pom.xml
 
-RUN mvn dependency:copy-dependencies -f /tmp/poc-ddd-aggregates/pom.xml -DoutputDirectory=target/dependencies -DincludeScope=compile
-COPY src /tmp/poc-ddd-aggregates/src
-RUN mvn package -DskipTests -o -f /tmp/poc-ddd-aggregates/pom.xml
+RUN mvn dependency:copy-dependencies -f kstreams/poc-ddd-aggregates/pom.xml -DoutputDirectory=target/dependencies -DincludeScope=runtime
+COPY kstreams/poc-ddd-aggregates/src kstreams/poc-ddd-aggregates/src
+RUN mvn package -DskipTests -o -f kstreams/poc-ddd-aggregates/pom.xml
 
 # 3. Create actual image (jlink-ed JDK, dependencies, JAR and launcher)
-FROM registry.fedoraproject.org/fedora-minimal:37
+FROM registry.fedoraproject.org/fedora-minimal:44
 
 COPY --from=jdk /opt/jre-minimal /opt/poc-ddd-aggregates/jdk
 RUN cd /opt/poc-ddd-aggregates \
     && curl -sO https://raw.githubusercontent.com/fabric8io-images/run-java-sh/master/fish-pepper/run-java-sh/fp-files/run-java.sh \
     && chmod u+x run-java.sh
-COPY --from=build /tmp/poc-ddd-aggregates/target/dependencies/* /opt/poc-ddd-aggregates/lib/
-COPY run-aggregator.sh /opt/poc-ddd-aggregates
-COPY --from=build /tmp/poc-ddd-aggregates/target/poc-ddd-aggregates-0.1-SNAPSHOT.jar /opt/poc-ddd-aggregates/lib/poc-ddd-aggregates-0.1-SNAPSHOT.jar
+COPY --from=build /tmp/build/kstreams/poc-ddd-aggregates/target/dependencies/* /opt/poc-ddd-aggregates/lib/
+COPY kstreams/poc-ddd-aggregates/run-aggregator.sh /opt/poc-ddd-aggregates
+COPY --from=build /tmp/build/kstreams/poc-ddd-aggregates/target/poc-ddd-aggregates-0.1-SNAPSHOT.jar /opt/poc-ddd-aggregates/lib/poc-ddd-aggregates-0.1-SNAPSHOT.jar
 
 # ENTRYPOINT /opt/poc-ddd-aggregates/run-aggregator.sh
diff --git a/kstreams/poc-ddd-aggregates/Dockerfile.dockerignore b/kstreams/poc-ddd-aggregates/Dockerfile.dockerignore
@@ -0,0 +1,6 @@
+# The build context is the repository root (for the parent POM); send only what the build uses
+*
+!pom.xml
+!kstreams/poc-ddd-aggregates/pom.xml
+!kstreams/poc-ddd-aggregates/src
+!kstreams/poc-ddd-aggregates/run-aggregator.sh
diff --git a/kstreams/poc-ddd-aggregates/pom.xml b/kstreams/poc-ddd-aggregates/pom.xml
@@ -3,37 +3,33 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
+    <parent>
+        <groupId>io.debezium.examples</groupId>
+        <artifactId>debezium-examples-parent</artifactId>
+        <version>1.0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
     <groupId>io.debezium.examples</groupId>
     <artifactId>poc-ddd-aggregates</artifactId>
     <version>0.1-SNAPSHOT</version>
 
-    <properties>
-        <mvn.compiler.version>3.8.1</mvn.compiler.version>
-        <mvn.dependency.version>3.1.1</mvn.dependency.version>
-        <apache.kafka.version>1.0.0</apache.kafka.version>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <java.version>11</java.version>
-    </properties>
-
     <dependencies>
 
         <dependency>
             <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka_2.11</artifactId>
-            <version>${apache.kafka.version}</version>
+            <artifactId>kafka-streams</artifactId>
         </dependency>
 
         <dependency>
             <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-streams</artifactId>
-            <version>${apache.kafka.version}</version>
+            <artifactId>kafka-clients</artifactId>
         </dependency>
 
+        <!-- Used directly by the JSON serdes; it used to come in only through the kafka_2.11 broker jar -->
         <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-            <version>${apache.kafka.version}</version>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
         </dependency>
 
     </dependencies>
@@ -43,31 +39,18 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>${mvn.compiler.version}</version>
-                <configuration>
-                    <source>${java.version}</source>
-                    <target>${java.version}</target>
-                </configuration>
+                <version>${version.maven.compiler}</version>
             </plugin>
         </plugins>
 	<pluginManagement>
             <plugins>
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-dependency-plugin</artifactId>
-	            <version>${mvn.dependency.version}</version>
+	            <version>${version.maven.dependency}</version>
                 </plugin>
             </plugins>
         </pluginManagement>
     </build>
 
-    <repositories>
-
-        <repository>
-            <id>confluent</id>
-            <url>https://packages.confluent.io/maven/</url>
-        </repository>
-
-    </repositories>
-
 </project>
diff --git a/...ddd-aggregates/src/main/java/io/debezium/examples/aggregation/StreamingAggregatesDDD.java b/...ddd-aggregates/src/main/java/io/debezium/examples/aggregation/StreamingAggregatesDDD.java
@@ -6,17 +6,17 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.utils.Bytes;
-import org.apache.kafka.streams.Consumed;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.Grouped;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Printed;
 import org.apache.kafka.streams.kstream.Produced;
-import org.apache.kafka.streams.kstream.Serialized;
 import org.apache.kafka.streams.state.KeyValueStore;
 
 import io.debezium.examples.aggregation.model.Address;
@@ -45,7 +45,7 @@ public static void main(String[] args) {
         Properties props = new Properties();
         props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streaming-aggregates-ddd");
         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10*1024);
+        props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 10*1024);
         props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
         props.put(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, 500);
         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@@ -70,7 +70,7 @@ public static void main(String[] args) {
 
         //2a) pseudo-aggreate addresses to keep latest relationship info
         KTable<DefaultId,LatestAddress> tempTable = addressStream
-                .groupByKey(Serialized.with(defaultIdSerde, addressSerde))
+                .groupByKey(Grouped.with(defaultIdSerde, addressSerde))
                 .aggregate(
                         () -> new LatestAddress(),
                         (DefaultId addressId, Address address, LatestAddress latest) -> {
@@ -86,7 +86,7 @@ public static void main(String[] args) {
         //2b) aggregate addresses per customer id
         KTable<DefaultId, Addresses> addressTable = tempTable.toStream()
                 .map((addressId, latestAddress) -> new KeyValue<>(latestAddress.getCustomerId(),latestAddress))
-                .groupByKey(Serialized.with(defaultIdSerde,latestAddressSerde))
+                .groupByKey(Grouped.with(defaultIdSerde,latestAddressSerde))
                 .aggregate(
                         () -> new Addresses(),
                         (customerId, latestAddress, addresses) -> {

diff --git a/kstreams/test.yaml b/kstreams/test.yaml
@@ -0,0 +1,100 @@
+name: kstreams
+description: >
+  Tests the MySQL -> Debezium -> Kafka Streams pipeline that joins customers and
+  their addresses into DDD aggregates on the final_ddd_aggregates topic.
+env_file: ../.env
+compose_file: docker-compose.yaml
+
+cleanup:
+  type: docker_compose_down
+  volumes: true
+
+steps:
+  - name: Ensure clean state before starting
+    type: docker_compose_down
+    volumes: true
+
+  - name: Build the aggregator image
+    type: docker_compose_build
+    services:
+      - aggregator
+
+  - name: Start Kafka, MySQL and Kafka Connect
+    type: docker_compose_up
+    detach: true
+    services:
+      - kafka
+      - mysql
+      - connect_source
+
+  - name: Wait for Kafka Connect REST API to be ready
+    type: http_wait
+    url: http://localhost:8083/connectors
+    timeout_seconds: 240
+    interval_seconds: 5
+
+  - name: Register MySQL source connector
+    type: http_post
+    url: http://localhost:8083/connectors
+    body_file: mysql-source.json
+    expected_status: [200, 201]
+
+  - name: Wait for MySQL source connector to be RUNNING
+    type: http_wait
+    url: http://localhost:8083/connectors/mysql-source/status
+    timeout_seconds: 90
+    interval_seconds: 5
+    expected_json_path: connector.state
+    expected_value: RUNNING
+
+  - name: Verify snapshot of the customers table
+    type: kafka_consume
+    topic: dbserver1.inventory.customers
+    timeout_seconds: 60
+    expected_content: Sally
+
+  - name: Verify snapshot of the addresses table
+    type: kafka_consume
+    topic: dbserver1.inventory.addresses
+    timeout_seconds: 60
+    expected_content: '"customer_id":1001'
+
+  # Start the aggregator only once both source topics exist
+  - name: Start the Kafka Streams aggregator
+    type: docker_compose_up
+    detach: true
+    services:
+      - aggregator
+
+  - name: Wait for the aggregator to build the first aggregate
+    type: wait_for_log
+    service: aggregator
+    expected_content: Sally
+    timeout_seconds: 180
+
+  - name: Verify the customer 1001 aggregate on final_ddd_aggregates
+    type: kafka_consume
+    topic: final_ddd_aggregates
+    timeout_seconds: 30
+    expected_content:
+      - '"first_name":"Sally"'
+      - '"customer_id":1001'
+
+  - name: Update customer 1001 in MySQL
+    type: docker_compose_exec
+    service: mysql
+    command: mysql -u mysqluser -pmysqlpw inventory -e "UPDATE customers SET first_name='Sarah' WHERE id=1001;"
+
+  - name: Wait for the aggregator to process the update
+    type: wait_for_log
+    service: aggregator
+    expected_content: Sarah
+    timeout_seconds: 60
+
+  - name: Verify the updated aggregate on final_ddd_aggregates
+    type: kafka_consume
+    topic: final_ddd_aggregates
+    timeout_seconds: 30
+    expected_content:
+      - '"first_name":"Sarah"'
+      - '"customer_id":1001'
diff --git a/pom.xml b/pom.xml
@@ -43,6 +43,7 @@
         <version.confluent>7.9.0</version.confluent>
         <version.surefire-plugin>3.5.5</version.surefire-plugin>
         <version.maven.compiler>3.15.0</version.maven.compiler>
+        <version.maven.dependency>3.11.0</version.maven.dependency>
 
         <!-- Common encoding defaults -->
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>