From: Kai Moritz Date: Wed, 17 Jul 2024 10:17:03 +0000 (+0200) Subject: Renamed test into `AggregateWindowedHoppingSuppressedTest` -- MOVE X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=357e01aff317d306eb2ea9a65947a7f1054ea085;p=demos%2Fkafka%2Fwordcount Renamed test into `AggregateWindowedHoppingSuppressedTest` -- MOVE --- diff --git a/src/test/java/de/juplo/kafka/streams/aggregate/windowed/hopping/AggregateWindowedHoppingSuppressedTest.java b/src/test/java/de/juplo/kafka/streams/aggregate/windowed/hopping/AggregateWindowedHoppingSuppressedTest.java new file mode 100644 index 0000000..65eeeab --- /dev/null +++ b/src/test/java/de/juplo/kafka/streams/aggregate/windowed/hopping/AggregateWindowedHoppingSuppressedTest.java @@ -0,0 +1,195 @@ +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.WindowStore; +import org.apache.kafka.streams.test.TestRecord; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.util.Properties; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + + +@Slf4j +public class AggregationTopologyTest +{ + static final Duration WINDOW_SIZE = Duration.ofSeconds(10); + static final Duration ADVANCE = Duration.ofSeconds(3); + static final TimeWindows WINDOWS = TimeWindows + .ofSizeWithNoGrace(WINDOW_SIZE) + .advanceBy(ADVANCE); + + + @Test + public void test() + { + StreamsBuilder builder = new StreamsBuilder(); + + KStream input = builder.stream(INPUT); + + input + .groupByKey() + .windowedBy(WINDOWS) + .reduce( + (aggregate, value) -> aggregate + "-" + value, + Materialized.>as("aggregated-store") + .withKeySerde(Serdes.String()) + .withValueSerde(Serdes.String())) + .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())) + .toStream((k,v) -> k.toString()) + .to(OUTPUT); + + Topology topology = builder.build(); + log.info("Generated topology: {}", topology.describe()); + + Properties properties = new Properties(); + properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); + properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); + + TopologyTestDriver testDriver = new TopologyTestDriver(topology, properties); + + in = testDriver.createInputTopic( + INPUT, + Serdes.String().serializer(), + Serdes.String().serializer()); + out = testDriver.createOutputTopic( + OUTPUT, + Serdes.String().deserializer(), + Serdes.String().deserializer()); + + + sendAt("A", 63); + assertThatOutcomeIs(); + + sendAt("B", 64); + assertThatOutcomeIs(KeyValue.pair(windowFor(54), "A")); + + sendAt("C", 65); + assertThatOutcomeIs(); + + sendAt("D", 66); + assertThatOutcomeIs(); + + sendAt("E", 69); + assertThatOutcomeIs(KeyValue.pair(windowFor(57), "A-B-C-D")); + + sendAt("F", 70); + assertThatOutcomeIs(KeyValue.pair(windowFor(60), "A-B-C-D-E")); + + sendAt("G", 74); + assertThatOutcomeIs(KeyValue.pair(windowFor(63), "A-B-C-D-E-F")); + + sendAt("H", 75); + assertThatOutcomeIs(); + + sendAt("I", 100); + assertThatOutcomeIs( + KeyValue.pair(windowFor(66), "D-E-F-G-H"), + KeyValue.pair(windowFor(69), "E-F-G-H"), + KeyValue.pair(windowFor(72), "G-H"), + KeyValue.pair(windowFor(75), "H")); + + sendAt("J", 120); + assertThatOutcomeIs( + KeyValue.pair(windowFor(93), "I"), + KeyValue.pair(windowFor(96), "I"), + KeyValue.pair(windowFor(99), "I")); + + sendAt("K", 140); + assertThatOutcomeIs( + KeyValue.pair(windowFor(111), "J"), + KeyValue.pair(windowFor(114), "J"), + KeyValue.pair(windowFor(117), "J"), + KeyValue.pair(windowFor(120), "J")); + + sendAt("L", 160); + assertThatOutcomeIs( + KeyValue.pair(windowFor(132), "K"), + KeyValue.pair(windowFor(135), "K"), + KeyValue.pair(windowFor(138), "K")); + + // Never received, if no newer message is send + // KeyValue.pair(windowFor(153), "L") + // KeyValue.pair(windowFor(156), "L") + //KeyValue.pair(windowFor(159), "L") + } + + + static final String INPUT = "TEST-INPUT"; + static final String OUTPUT = "TEST-OUTPUT"; + + static final String KEY = "foo"; + + + TestInputTopic in; + TestOutputTopic out; + + + void sendAt(String value, int second) + { + TestRecord record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second)); + log.info( + "Sending @ {}: {} = {}", + record.getRecordTime().toEpochMilli(), + record.key(), + record.value()); + in.pipeInput(record); + } + + void assertThatOutcomeIs(KeyValue, String>... expected) + { + assertThat(outcome()).containsExactly(expected); + } + + Stream, String>> outcome() + { + return out + .readRecordsToList() + .stream() + .peek(record -> log.info( + "Received @ {}: {} = {}", + record.getRecordTime().toEpochMilli(), + record.key(), + record.value())) + .map(record -> KeyValue.pair(parse(record.key()), record.value())); + } + + + static final Pattern PATTERN = Pattern.compile("^\\[([^@]*)@(\\d+)/(\\d+)\\]$"); + + Windowed parse(String serialized) + { + Matcher matcher = PATTERN.matcher(serialized); + + if (!matcher.matches()) + { + throw new IllegalArgumentException(serialized + "does not match " + PATTERN.pattern()); + } + + String key = matcher.group(1); + String start = matcher.group(2); + String end = matcher.group(3); + + Window window = new TimeWindow(Long.parseLong(start), Long.parseLong(end)); + + return new Windowed<>(key, window); + } + + Windowed windowFor(int second) + { + Instant startTime = Instant.ofEpochSecond(second); + Instant endTime = startTime.plus(WINDOW_SIZE); + TimeWindow window = new TimeWindow(startTime.toEpochMilli(), endTime.toEpochMilli()); + return new Windowed<>(KEY, window); + } +} diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java deleted file mode 100644 index 65eeeab..0000000 --- a/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java +++ /dev/null @@ -1,195 +0,0 @@ -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.WindowStore; -import org.apache.kafka.streams.test.TestRecord; -import org.junit.jupiter.api.Test; - -import java.time.Duration; -import java.time.Instant; -import java.util.Properties; -import java.util.regex.Matcher; -import java.util.regex.Pattern; -import java.util.stream.Stream; - -import static org.assertj.core.api.Assertions.assertThat; - - -@Slf4j -public class AggregationTopologyTest -{ - static final Duration WINDOW_SIZE = Duration.ofSeconds(10); - static final Duration ADVANCE = Duration.ofSeconds(3); - static final TimeWindows WINDOWS = TimeWindows - .ofSizeWithNoGrace(WINDOW_SIZE) - .advanceBy(ADVANCE); - - - @Test - public void test() - { - StreamsBuilder builder = new StreamsBuilder(); - - KStream input = builder.stream(INPUT); - - input - .groupByKey() - .windowedBy(WINDOWS) - .reduce( - (aggregate, value) -> aggregate + "-" + value, - Materialized.>as("aggregated-store") - .withKeySerde(Serdes.String()) - .withValueSerde(Serdes.String())) - .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())) - .toStream((k,v) -> k.toString()) - .to(OUTPUT); - - Topology topology = builder.build(); - log.info("Generated topology: {}", topology.describe()); - - Properties properties = new Properties(); - properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); - properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); - - TopologyTestDriver testDriver = new TopologyTestDriver(topology, properties); - - in = testDriver.createInputTopic( - INPUT, - Serdes.String().serializer(), - Serdes.String().serializer()); - out = testDriver.createOutputTopic( - OUTPUT, - Serdes.String().deserializer(), - Serdes.String().deserializer()); - - - sendAt("A", 63); - assertThatOutcomeIs(); - - sendAt("B", 64); - assertThatOutcomeIs(KeyValue.pair(windowFor(54), "A")); - - sendAt("C", 65); - assertThatOutcomeIs(); - - sendAt("D", 66); - assertThatOutcomeIs(); - - sendAt("E", 69); - assertThatOutcomeIs(KeyValue.pair(windowFor(57), "A-B-C-D")); - - sendAt("F", 70); - assertThatOutcomeIs(KeyValue.pair(windowFor(60), "A-B-C-D-E")); - - sendAt("G", 74); - assertThatOutcomeIs(KeyValue.pair(windowFor(63), "A-B-C-D-E-F")); - - sendAt("H", 75); - assertThatOutcomeIs(); - - sendAt("I", 100); - assertThatOutcomeIs( - KeyValue.pair(windowFor(66), "D-E-F-G-H"), - KeyValue.pair(windowFor(69), "E-F-G-H"), - KeyValue.pair(windowFor(72), "G-H"), - KeyValue.pair(windowFor(75), "H")); - - sendAt("J", 120); - assertThatOutcomeIs( - KeyValue.pair(windowFor(93), "I"), - KeyValue.pair(windowFor(96), "I"), - KeyValue.pair(windowFor(99), "I")); - - sendAt("K", 140); - assertThatOutcomeIs( - KeyValue.pair(windowFor(111), "J"), - KeyValue.pair(windowFor(114), "J"), - KeyValue.pair(windowFor(117), "J"), - KeyValue.pair(windowFor(120), "J")); - - sendAt("L", 160); - assertThatOutcomeIs( - KeyValue.pair(windowFor(132), "K"), - KeyValue.pair(windowFor(135), "K"), - KeyValue.pair(windowFor(138), "K")); - - // Never received, if no newer message is send - // KeyValue.pair(windowFor(153), "L") - // KeyValue.pair(windowFor(156), "L") - //KeyValue.pair(windowFor(159), "L") - } - - - static final String INPUT = "TEST-INPUT"; - static final String OUTPUT = "TEST-OUTPUT"; - - static final String KEY = "foo"; - - - TestInputTopic in; - TestOutputTopic out; - - - void sendAt(String value, int second) - { - TestRecord record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second)); - log.info( - "Sending @ {}: {} = {}", - record.getRecordTime().toEpochMilli(), - record.key(), - record.value()); - in.pipeInput(record); - } - - void assertThatOutcomeIs(KeyValue, String>... expected) - { - assertThat(outcome()).containsExactly(expected); - } - - Stream, String>> outcome() - { - return out - .readRecordsToList() - .stream() - .peek(record -> log.info( - "Received @ {}: {} = {}", - record.getRecordTime().toEpochMilli(), - record.key(), - record.value())) - .map(record -> KeyValue.pair(parse(record.key()), record.value())); - } - - - static final Pattern PATTERN = Pattern.compile("^\\[([^@]*)@(\\d+)/(\\d+)\\]$"); - - Windowed parse(String serialized) - { - Matcher matcher = PATTERN.matcher(serialized); - - if (!matcher.matches()) - { - throw new IllegalArgumentException(serialized + "does not match " + PATTERN.pattern()); - } - - String key = matcher.group(1); - String start = matcher.group(2); - String end = matcher.group(3); - - Window window = new TimeWindow(Long.parseLong(start), Long.parseLong(end)); - - return new Windowed<>(key, window); - } - - Windowed windowFor(int second) - { - Instant startTime = Instant.ofEpochSecond(second); - Instant endTime = startTime.plus(WINDOW_SIZE); - TimeWindow window = new TimeWindow(startTime.toEpochMilli(), endTime.toEpochMilli()); - return new Windowed<>(KEY, window); - } -}