This example needs an explicit serialization-config

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java
index 7efd7516b8c772560c34ec793fbd90d20fbee2dd..5c2ff6b39d1eba71a7c64e93cd9bf495c53aaea6 100644 (file)
--- a/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java
@@ -2,10 +2,12 @@ package de.juplo.kafka.wordcount.counter;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.*;
 import org.apache.kafka.streams.kstream.*;
 import org.apache.kafka.streams.kstream.internals.TimeWindow;
 import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.WindowStore;
 import org.apache.kafka.streams.test.TestRecord;
 import org.junit.jupiter.api.Test;
 
@@ -39,7 +41,9 @@ public class AggregationTopologyTest
         .windowedBy(WINDOWS)
         .reduce(
             (aggregate, value) -> aggregate + "-" + value,
-            Materialized.as(STORE_NAME))
+            Materialized.<String, String, WindowStore<Bytes, byte[]>>as(STORE_NAME)
+                .withKeySerde(Serdes.String())
+                .withValueSerde(Serdes.String()))
         .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
         .toStream((k,v) -> k.toString())
         .to(OUTPUT);