Renamed test into `StreamStreamLeftJoinTopologyTest` -- MOVE

diff --git a/src/test/java/de/juplo/kafka/streams/join/stream/stream/StreamStreamLeftJoinTest.java b/src/test/java/de/juplo/kafka/streams/join/stream/stream/StreamStreamLeftJoinTest.java
new file mode 100644 (file)
index 0000000..5323663
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/streams/join/stream/stream/StreamStreamLeftJoinTest.java
@@ -0,0 +1,154 @@
+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.JoinWindows;
+import org.apache.kafka.streams.kstream.KStream;
+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.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+@Slf4j
+public class JoinTopologyTest
+{
+  static final Duration WINDOW_SIZE = Duration.ofSeconds(10);
+  static final Duration GRACE_PERIOD = Duration.ofSeconds(5);
+  static final JoinWindows WINDOWS = JoinWindows.ofTimeDifferenceAndGrace(WINDOW_SIZE, GRACE_PERIOD);
+
+
+  @Test
+  public void test()
+  {
+    StreamsBuilder builder = new StreamsBuilder();
+
+    KStream<String, String> left = builder.stream(LEFT);
+    KStream<String, String> right = builder.stream(RIGHT);
+
+    left
+        .join(
+            right,
+            (valueLeft, valueRight) -> valueLeft + "-" + valueRight,
+            WINDOWS)
+        .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);
+
+    inLeft = testDriver.createInputTopic(
+        LEFT,
+        Serdes.String().serializer(),
+        Serdes.String().serializer());
+    inRight = testDriver.createInputTopic(
+        RIGHT,
+        Serdes.String().serializer(),
+        Serdes.String().serializer());
+    out = testDriver.createOutputTopic(
+        OUTPUT,
+        Serdes.String().deserializer(),
+        Serdes.String().deserializer());
+
+
+    sendLeftAt("A", 63);
+    assertThatOutcomeIs();
+
+    sendRightAt("a", 64);
+    assertThatOutcomeIs("A-a");
+
+    sendLeftAt("B", 65);
+    assertThatOutcomeIs("B-a");
+
+    sendRightAt("b", 66);
+    assertThatOutcomeIs("A-b", "B-b");
+
+    sendLeftAt("C", 69);
+    assertThatOutcomeIs("C-a", "C-b");
+
+    sendRightAt("c", 70);
+    assertThatOutcomeIs("A-c", "B-c", "C-c");
+
+    sendRightAt("d", 74);
+    assertThatOutcomeIs("B-d", "C-d"); // !
+
+    sendLeftAt("D", 75);
+    assertThatOutcomeIs("D-b", "D-c", "D-d");
+
+    sendLeftAt("E", 100);
+    assertThatOutcomeIs();
+
+    sendLeftAt("F", 120);
+    assertThatOutcomeIs();
+
+    sendRightAt("f", 140);
+    assertThatOutcomeIs();
+
+    sendLeftAt("G", 160);
+    assertThatOutcomeIs();
+  }
+
+
+  static final String LEFT = "TEST-LEFT";
+  static final String RIGHT = "TEST-RIGHT";
+  static final String OUTPUT = "TEST-OUTPUT";
+
+  static final String KEY = "foo";
+
+
+  TestInputTopic<String, String> inLeft;
+  TestInputTopic<String, String> inRight;
+  TestOutputTopic<String, String> out;
+
+
+  void sendLeftAt(String value, int second)
+  {
+    TestRecord<String, String> record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second));
+    log.info(
+        "Sending  LEFT  @ {}: {} = {}",
+        record.getRecordTime().toEpochMilli(),
+        record.key(),
+        record.value());
+    inLeft.pipeInput(record);
+  }
+
+  void sendRightAt(String value, int second)
+  {
+    TestRecord<String, String> record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second));
+    log.info(
+        "Sending  RIGHT @ {}: {} = {}",
+        record.getRecordTime().toEpochMilli(),
+        record.key(),
+        record.value());
+    inRight.pipeInput(record);
+  }
+
+  void assertThatOutcomeIs(String... expected)
+  {
+    assertThat(outcome()).containsExactly(expected);
+  }
+
+  Stream<String> outcome()
+  {
+    return out
+        .readRecordsToList()
+        .stream()
+        .peek(record -> log.info(
+            "Received join  @ {}: {} = {}",
+            record.getRecordTime().toEpochMilli(),
+            record.key(),
+            record.value()))
+        .map(record -> record.value());
+  }
+}

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/JoinTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/JoinTopologyTest.java
deleted file mode 100644 (file)
index 5323663..0000000
--- a/src/test/java/de/juplo/kafka/wordcount/counter/JoinTopologyTest.java
+++ /dev/null
@@ -1,154 +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.JoinWindows;
-import org.apache.kafka.streams.kstream.KStream;
-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.stream.Stream;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-
-@Slf4j
-public class JoinTopologyTest
-{
-  static final Duration WINDOW_SIZE = Duration.ofSeconds(10);
-  static final Duration GRACE_PERIOD = Duration.ofSeconds(5);
-  static final JoinWindows WINDOWS = JoinWindows.ofTimeDifferenceAndGrace(WINDOW_SIZE, GRACE_PERIOD);
-
-
-  @Test
-  public void test()
-  {
-    StreamsBuilder builder = new StreamsBuilder();
-
-    KStream<String, String> left = builder.stream(LEFT);
-    KStream<String, String> right = builder.stream(RIGHT);
-
-    left
-        .join(
-            right,
-            (valueLeft, valueRight) -> valueLeft + "-" + valueRight,
-            WINDOWS)
-        .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);
-
-    inLeft = testDriver.createInputTopic(
-        LEFT,
-        Serdes.String().serializer(),
-        Serdes.String().serializer());
-    inRight = testDriver.createInputTopic(
-        RIGHT,
-        Serdes.String().serializer(),
-        Serdes.String().serializer());
-    out = testDriver.createOutputTopic(
-        OUTPUT,
-        Serdes.String().deserializer(),
-        Serdes.String().deserializer());
-
-
-    sendLeftAt("A", 63);
-    assertThatOutcomeIs();
-
-    sendRightAt("a", 64);
-    assertThatOutcomeIs("A-a");
-
-    sendLeftAt("B", 65);
-    assertThatOutcomeIs("B-a");
-
-    sendRightAt("b", 66);
-    assertThatOutcomeIs("A-b", "B-b");
-
-    sendLeftAt("C", 69);
-    assertThatOutcomeIs("C-a", "C-b");
-
-    sendRightAt("c", 70);
-    assertThatOutcomeIs("A-c", "B-c", "C-c");
-
-    sendRightAt("d", 74);
-    assertThatOutcomeIs("B-d", "C-d"); // !
-
-    sendLeftAt("D", 75);
-    assertThatOutcomeIs("D-b", "D-c", "D-d");
-
-    sendLeftAt("E", 100);
-    assertThatOutcomeIs();
-
-    sendLeftAt("F", 120);
-    assertThatOutcomeIs();
-
-    sendRightAt("f", 140);
-    assertThatOutcomeIs();
-
-    sendLeftAt("G", 160);
-    assertThatOutcomeIs();
-  }
-
-
-  static final String LEFT = "TEST-LEFT";
-  static final String RIGHT = "TEST-RIGHT";
-  static final String OUTPUT = "TEST-OUTPUT";
-
-  static final String KEY = "foo";
-
-
-  TestInputTopic<String, String> inLeft;
-  TestInputTopic<String, String> inRight;
-  TestOutputTopic<String, String> out;
-
-
-  void sendLeftAt(String value, int second)
-  {
-    TestRecord<String, String> record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second));
-    log.info(
-        "Sending  LEFT  @ {}: {} = {}",
-        record.getRecordTime().toEpochMilli(),
-        record.key(),
-        record.value());
-    inLeft.pipeInput(record);
-  }
-
-  void sendRightAt(String value, int second)
-  {
-    TestRecord<String, String> record = new TestRecord<>(KEY, value, Instant.ofEpochSecond(second));
-    log.info(
-        "Sending  RIGHT @ {}: {} = {}",
-        record.getRecordTime().toEpochMilli(),
-        record.key(),
-        record.value());
-    inRight.pipeInput(record);
-  }
-
-  void assertThatOutcomeIs(String... expected)
-  {
-    assertThat(outcome()).containsExactly(expected);
-  }
-
-  Stream<String> outcome()
-  {
-    return out
-        .readRecordsToList()
-        .stream()
-        .peek(record -> log.info(
-            "Received join  @ {}: {} = {}",
-            record.getRecordTime().toEpochMilli(),
-            record.key(),
-            record.value()))
-        .map(record -> record.value());
-  }
-}