From: Kai Moritz <kai@juplo.de>
Date: Wed, 17 Jul 2024 05:23:34 +0000 (+0200)
Subject: Renamed test into `AggregateTest` -- MOVE
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=c167aac99a37569a297cffe40b9a20049956f151;p=demos%2Fkafka%2Fwordcount

Renamed test into `AggregateTest` -- MOVE
---

diff --git a/src/test/java/de/juplo/kafka/streams/aggregate/AggregateTest.java b/src/test/java/de/juplo/kafka/streams/aggregate/AggregateTest.java
new file mode 100644
index 0000000..082b1f0
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/streams/aggregate/AggregateTest.java
@@ -0,0 +1,131 @@
+package de.juplo.kafka.streams.aggregate;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.*;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.test.TestRecord;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.Properties;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+@Slf4j
+public class AggregationTopologyTest
+{
+  @Test
+  public void test()
+  {
+    StreamsBuilder builder = new StreamsBuilder();
+
+    KStream<String, String> input = builder.stream(INPUT);
+
+    input
+        .peek((k,v) -> log.info("peek-0 -- {} = {}", k, v))
+        .groupByKey()
+        .reduce((aggregate, value) -> aggregate + "-" + value)
+        .toStream()
+        .peek((k,v) -> log.info("peek-1 -- {} = {}", k, v))
+        .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);
+    properties.put(StreamsConfig.InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX, 0l);
+
+    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("A");
+
+    sendAt("B", 64);
+    assertThatOutcomeIs("A-B");
+
+    sendAt("C", 65);
+    assertThatOutcomeIs("A-B-C");
+
+    sendAt("D", 66);
+    assertThatOutcomeIs("A-B-C-D");
+
+    sendAt("E", 69);
+    assertThatOutcomeIs("A-B-C-D-E");
+
+    sendAt("F", 70);
+    assertThatOutcomeIs("A-B-C-D-E-F");
+
+    sendAt("G", 74);
+    assertThatOutcomeIs("A-B-C-D-E-F-G");
+
+    sendAt("H", 75);
+    assertThatOutcomeIs("A-B-C-D-E-F-G-H");
+
+    sendAt("I", 100);
+    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I");
+
+    sendAt("J", 120);
+    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I-J");
+
+    sendAt("K", 140);
+    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I-J-K");
+
+    sendAt("L", 160);
+    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I-J-K-L");
+  }
+
+
+  static final String INPUT = "TEST-INPUT";
+  static final String OUTPUT = "TEST-OUTPUT";
+
+  static final String KEY = "foo";
+
+
+  TestInputTopic<String, String> in;
+  TestOutputTopic<String, String> out;
+
+
+  void sendAt(String value, int second)
+  {
+    TestRecord<String, String> record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second));
+    log.info(
+        "Sending  @ {}: {} = {}",
+        record.getRecordTime().toEpochMilli(),
+        record.key(),
+        record.value());
+    in.pipeInput(record);
+  }
+
+  void assertThatOutcomeIs(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 -> record.value());
+  }
+}
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 d21ad47..0000000
--- a/src/test/java/de/juplo/kafka/wordcount/counter/AggregationTopologyTest.java
+++ /dev/null
@@ -1,131 +0,0 @@
-package de.juplo.kafka.wordcount.counter;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.streams.*;
-import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.test.TestRecord;
-import org.junit.jupiter.api.Test;
-
-import java.time.Instant;
-import java.util.Properties;
-import java.util.stream.Stream;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-
-@Slf4j
-public class AggregationTopologyTest
-{
-  @Test
-  public void test()
-  {
-    StreamsBuilder builder = new StreamsBuilder();
-
-    KStream<String, String> input = builder.stream(INPUT);
-
-    input
-        .peek((k,v) -> log.info("peek-0 -- {} = {}", k, v))
-        .groupByKey()
-        .reduce((aggregate, value) -> aggregate + "-" + value)
-        .toStream()
-        .peek((k,v) -> log.info("peek-1 -- {} = {}", k, v))
-        .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);
-    properties.put(StreamsConfig.InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX, 0l);
-
-    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("A");
-
-    sendAt("B", 64);
-    assertThatOutcomeIs("A-B");
-
-    sendAt("C", 65);
-    assertThatOutcomeIs("A-B-C");
-
-    sendAt("D", 66);
-    assertThatOutcomeIs("A-B-C-D");
-
-    sendAt("E", 69);
-    assertThatOutcomeIs("A-B-C-D-E");
-
-    sendAt("F", 70);
-    assertThatOutcomeIs("A-B-C-D-E-F");
-
-    sendAt("G", 74);
-    assertThatOutcomeIs("A-B-C-D-E-F-G");
-
-    sendAt("H", 75);
-    assertThatOutcomeIs("A-B-C-D-E-F-G-H");
-
-    sendAt("I", 100);
-    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I");
-
-    sendAt("J", 120);
-    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I-J");
-
-    sendAt("K", 140);
-    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I-J-K");
-
-    sendAt("L", 160);
-    assertThatOutcomeIs("A-B-C-D-E-F-G-H-I-J-K-L");
-  }
-
-
-  static final String INPUT = "TEST-INPUT";
-  static final String OUTPUT = "TEST-OUTPUT";
-
-  static final String KEY = "foo";
-
-
-  TestInputTopic<String, String> in;
-  TestOutputTopic<String, String> out;
-
-
-  void sendAt(String value, int second)
-  {
-    TestRecord<String, String> record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second));
-    log.info(
-        "Sending  @ {}: {} = {}",
-        record.getRecordTime().toEpochMilli(),
-        record.key(),
-        record.value());
-    in.pipeInput(record);
-  }
-
-  void assertThatOutcomeIs(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 -> record.value());
-  }
-}