Rewrite tests to use new KIP-470 API #301

mjsax · 2019-10-29T04:38:42Z

No description provided.

mjsax · 2019-10-29T04:39:13Z

src/main/java/io/confluent/examples/streams/microservices/EmailService.java

@@ -84,7 +84,7 @@ private KafkaStreams processStreams(final String bootstrapServers, final String
    final GlobalKTable<Long, Customer> customers = builder.globalTable(CUSTOMERS.name(),
        Consumed.with(CUSTOMERS.keySerde(), CUSTOMERS.valueSerde()));

-    final Joined<String, Order, Payment> serdes = Joined
+    final StreamJoined<String, Order, Payment> serdes = StreamJoined


KIP-479 cleanup

Thanks for the KIP-479 cleanups!

mjsax · 2019-10-29T04:39:21Z

src/main/java/io/confluent/examples/streams/microservices/EmailService.java

@@ -135,7 +135,7 @@ public void stop() {
    void sendEmail(EmailTuple details);
  }

-  public class EmailTuple {
+  public static class EmailTuple {


side cleanup

mjsax · 2019-10-29T04:39:26Z

src/main/java/io/confluent/examples/streams/microservices/ValidationsAggregatorService.java

@@ -47,13 +47,13 @@
      ORDERS.valueSerde());
  private final Grouped<String, OrderValidation> serdes3 = Grouped
      .with(ORDER_VALIDATIONS.keySerde(), ORDER_VALIDATIONS.valueSerde());
-  private final Joined<String, Long, Order> serdes4 = Joined
+  private final StreamJoined<String, Long, Order> serdes4 = StreamJoined


KIP-479 cleanup

mjsax · 2019-10-29T04:40:51Z

src/test/java/io/confluent/examples/streams/GenericAvroIntegrationTest.java

@@ -72,7 +70,7 @@ public void shouldRoundTripGenericAvroDataThroughKafka() throws Exception {
    record.put("user", "alice");
    record.put("is_new", true);
    record.put("content", "lorem ipsum");
-    final List<GenericRecord> inputValues = Collections.singletonList(record);
+    final List<Object> inputValues = Collections.singletonList(record);


new KIP-470 API cannot resolve GenericRecord as type, because KafkaAvroSerializer uses Object...

Huh, that's a bummer.

mjsax · 2019-10-29T04:41:34Z

src/test/java/io/confluent/examples/streams/IntegrationTestUtils.java

-   * @param valueSerializer    The {@link Serializer} corresponding to the value type
-   * @param <V>                Value type of the data records
-   */
-  static <K, V> void produceValuesSynchronously(final String topic,


With the new KIP-470 API, we don't need those helpers any longer.

mjsax · 2019-10-29T04:41:50Z

src/test/java/io/confluent/examples/streams/MapFunctionLambdaIntegrationTest.java

@@ -58,29 +59,26 @@ public void shouldUppercaseTheInput() {
    final String inputTopic = "inputTopic";
    final String outputTopic = "outputTopic";
    final KStream<byte[], String> input = builder.stream(inputTopic);
-    final KStream<byte[], String> uppercased = input.mapValues(s -> s.toUpperCase());
+    final KStream<byte[], String> uppercased = input.mapValues((ValueMapper<String, String>) String::toUpperCase);


side cleanup

mjsax · 2019-10-29T04:42:19Z

src/test/java/io/confluent/examples/streams/SessionWindowsExampleTest.java

    MockSchemaRegistry.dropScope(SCHEMA_REGISTRY_SCOPE);
  }

  @Test
-  public void shouldCountPlayEventsBySession() throws Exception {
+  public void shouldCountPlayEventsBySession() {


side cleanup -- this method does not throw

mjsax · 2019-10-29T04:43:34Z

src/test/java/io/confluent/examples/streams/StateStoresInTheDSLIntegrationTest.java

+      .keyValueStoreBuilder(
+        Stores.persistentKeyValueStore("WordCountsStore"),
+        Serdes.String(),
+        Serdes.Long())


fix indentation only

mjsax · 2019-10-29T04:43:48Z

src/test/java/io/confluent/examples/streams/StreamToStreamJoinIntegrationTest.java

@@ -99,7 +99,7 @@ public void shouldJoinTwoStreams() {
      // In this specific example, we don't need to define join serdes explicitly because the key, left value, and
      // right value are all of type String, which matches our default serdes configured for the application.  However,
      // we want to showcase the use of `Joined.with(...)` in case your code needs a different type setup.
-      Joined.with(
+      StreamJoined.with(


KIP-479 cleanup

mjsax · 2019-10-29T04:44:13Z

src/test/java/io/confluent/examples/streams/StreamToTableJoinIntegrationTest.java

-        // Compute the total per region by summing the individual click counts per region.
-        .groupByKey(Grouped.with(stringSerde, longSerde))
-        .reduce(Long::sum);
+      // Join the stream against the table.


fix indentation only

mjsax · 2019-10-29T04:44:48Z

src/test/java/io/confluent/examples/streams/TableToTableJoinIntegrationTest.java

-      (regionValue, lastLoginValue) -> regionValue + "/" + lastLoginValue,
-      Materialized.as(storeName))
+                     (regionValue, lastLoginValue) -> regionValue + "/" + lastLoginValue,
+                     Materialized.as(storeName))


fix indentation only

mjsax · 2019-10-29T04:45:26Z

src/test/java/io/confluent/examples/streams/WordCountLambdaExampleTest.java

-   *
-   * @return ProducerRecord containing word as key and count as value
-   */
-  private ProducerRecord<String, Long> readOutput() {


With KIP-470, we don't need those helpers any longer

mjsax · 2019-10-29T04:47:29Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

@@ -61,220 +61,243 @@

    @Test
    public void shouldSumNumbersOnSameDay() {
-        final List<MyEvent> inputValues = Arrays.asList(


Replacing MyEvent with new KIP-470 `TestRecord:

requires to pass in null key

required to pass timestamp as Instant

mjsax · 2019-10-29T04:48:07Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

-                new MyEvent(7, ZonedDateTime.of(2019, 1, 1, 16, 31, 0, 0, zone)),
-                dummyEventToForceSuppression()
-        );
-        final List<ExpectedResult> expectedValues = Arrays.asList(


Replacing ExpectedResult with simple KeyValue pair

mjsax · 2019-10-29T04:49:00Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

        );
-        verify(inputValues, expectedValues, zone);
+      final List<KeyValue<Windowed<Integer>, Integer>> expectedValues =
+        Collections.singletonList(KeyValue.pair(toWindowed(1,


Using toWindowed to get Windowed<K> for the key passing in key, start/end ts

mjsax · 2019-10-29T04:51:28Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

-
-    }
-
-    private class MyEvent {


MyEvent replace with TestRecord from KIP-470

mjsax · 2019-10-29T04:51:48Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

-        }
-    }
-
-    private class ExpectedResult {


ExpectedResult replace with KeyValue pair

mjsax · 2019-10-29T04:52:17Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

-        }
-    }
-
-    private void verifyResults(final List<ExpectedResult> expectedValues, final TopologyTestDriver testDriver) {


not needed any longer -- using KIP-470 API now

mjsax · 2019-10-29T04:52:30Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

    }

-    private void injectFakeData(final List<MyEvent> inputValues, final TopologyTestDriver testDriver) {


not needed any longer -- using KIP-470 API now

mjsax · 2019-10-29T04:54:00Z

src/test/scala/io/confluent/examples/streams/IntegrationTestScalaUtils.scala

@@ -85,25 +86,28 @@ object IntegrationTestScalaUtils {
  def produceValuesSynchronously[V](topic: String, values: Seq[V], driver: TopologyTestDriver)
                                   (implicit valueSerializer: Serializer[V]): Unit = {
    import collection.JavaConverters._
-    IntegrationTestUtils.produceValuesSynchronously(topic, values.asJava, driver, valueSerializer)


Because we removed those helper, we call the TopologyTestDriver method directly -- it's still useful to keep the Scala helpers for explicit Serde support

vvcephei

Wow, thanks @mjsax ! I had one high-level suggestion, and a few smaller comments.
-John

vvcephei · 2019-10-29T18:13:48Z

src/test/java/io/confluent/examples/streams/AggregateTest.java

+      testDriver.createInputTopic(inputTopic,
+                                  new IntegrationTestUtils.NothingSerde<>(),
+                                  new StringSerializer())
+        .pipeValueList(inputValues);


Tentatively, I think it might set a better example for users if we establish a pattern like:

try (testDriver = new TopologyTestDriver(...)) { final TestInputTopic<...> inputTopic = testDriver.createInputTopic(...); final TestOutputTopic<...> outputTopic = testDriver.createOutputTopic(...); inputTopic.pipeValueList(inputValues); assertThat(outputTopic.readKeyValuesToMap()).isEqualTo(expectedOutput); }

In other words, I feel it leads to more readable tests if we declare the inputs and outputs right up front, and then push data through and assert the results later on.

vvcephei · 2019-10-29T18:15:15Z

src/test/java/io/confluent/examples/streams/GenericAvroIntegrationTest.java

@@ -72,7 +70,7 @@ public void shouldRoundTripGenericAvroDataThroughKafka() throws Exception {
    record.put("user", "alice");
    record.put("is_new", true);
    record.put("content", "lorem ipsum");
-    final List<GenericRecord> inputValues = Collections.singletonList(record);
+    final List<Object> inputValues = Collections.singletonList(record);


Huh, that's a bummer.

vvcephei · 2019-10-29T18:18:20Z

src/test/java/io/confluent/examples/streams/HandlingCorruptedInputRecordsIntegrationTest.java

+      topologyTestDriver.createInputTopic(inputTopic,
+                                          new IntegrationTestUtils.NothingSerde<>(),
+                                          new LongSerializer())
+        .pipeValueList(inputValues);


It seems like declaring the same input topic twice may disguise the intent of this test. Do you think declaring it once with a ByteArray value serializer, and then piping in invalid and valid data arrays would be clearer?

vvcephei · 2019-10-29T18:18:32Z

src/test/java/io/confluent/examples/streams/IntegrationTestUtils.java

-   * @param valueSerializer    The {@link Serializer} corresponding to the value type
-   * @param <V>                Value type of the data records
-   */
-  static <K, V> void produceValuesSynchronously(final String topic,


vvcephei · 2019-10-29T18:20:13Z

src/test/java/io/confluent/examples/streams/SessionWindowsExampleTest.java

-      playEventSerializer,
-      start
-    );
+    topologyTestDriver.createInputTopic(SessionWindowsExample.PLAY_EVENTS,


maybe we can also create the input and output topics in the @Before method so we can tidy up these tests.

vvcephei · 2019-10-29T18:23:39Z

src/test/java/io/confluent/examples/streams/window/CustomWindowTest.java

+      streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
+      // The commit interval for flushing records to state stores and downstream must be lower than
+      // this integration test's timeout (30 secs) to ensure we observe the expected processing results.
+      streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);


This config has no effect now, right?

Ack. Seems to be a left over from moving from "EmbeddedKafka" to TDD.

bbejeck

Thanks for the clean-up @mjsax, just one minor comment but LGTM modulo @vvcephei comments.

bbejeck · 2019-10-29T19:21:00Z

src/main/java/io/confluent/examples/streams/microservices/EmailService.java

@@ -84,7 +84,7 @@ private KafkaStreams processStreams(final String bootstrapServers, final String
    final GlobalKTable<Long, Customer> customers = builder.globalTable(CUSTOMERS.name(),
        Consumed.with(CUSTOMERS.keySerde(), CUSTOMERS.valueSerde()));

-    final Joined<String, Order, Payment> serdes = Joined
+    final StreamJoined<String, Order, Payment> serdes = StreamJoined


Thanks for the KIP-479 cleanups!

bbejeck · 2019-10-29T19:26:51Z

src/test/java/io/confluent/examples/streams/StreamToStreamJoinIntegrationTest.java

@@ -99,7 +99,7 @@ public void shouldJoinTwoStreams() {
      // In this specific example, we don't need to define join serdes explicitly because the key, left value, and
      // right value are all of type String, which matches our default serdes configured for the application.  However,
      // we want to showcase the use of `Joined.with(...)` in case your code needs a different type setup.


nit: update comment to use StreamJoined as well.

mjsax · 2019-10-29T22:06:31Z

Updated this. (Added some side cleanup to use hamcrest for testing in all test)

vvcephei

Thanks so much, @mjsax ; this looks awesome!

mjsax · 2019-10-30T20:28:02Z

Rebasing this to new branch 5.4.x and also resolve merge conflicts.

mjsax requested a review from a team October 29, 2019 04:38

mjsax commented Oct 29, 2019

View reviewed changes

vvcephei reviewed Oct 29, 2019

View reviewed changes

bbejeck approved these changes Oct 29, 2019

View reviewed changes

vvcephei approved these changes Oct 30, 2019

View reviewed changes

mjsax changed the base branch from master to 5.4.x October 30, 2019 20:23

mjsax added 3 commits October 30, 2019 13:28

Rewrite tests to use new KIP-470 API

fce049a

Github comments

aa10538

Github comment

37d4225

mjsax force-pushed the kip-470-rewrites branch from f9e39b6 to 37d4225 Compare October 30, 2019 20:29

mjsax merged commit 1f62f73 into 5.4.x Oct 30, 2019

mjsax deleted the kip-470-rewrites branch October 30, 2019 22:28

		}

		private void injectFakeData(final List<MyEvent> inputValues, final TopologyTestDriver testDriver) {

Rewrite tests to use new KIP-470 API #301

Rewrite tests to use new KIP-470 API #301

Conversation

mjsax commented Oct 29, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Oct 29, 2019

vvcephei left a comment

Choose a reason for hiding this comment

mjsax commented Oct 30, 2019