content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.streams.kstream.internals;
  
  import org.apache.kafka.common.serialization.Deserializer;
  import org.apache.kafka.common.serialization.LongDeserializer;
  import org.apache.kafka.common.serialization.Serde;
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.apache.kafka.common.utils.Bytes;
  import org.apache.kafka.common.utils.Utils;
  import org.apache.kafka.streams.KeyValue;
  import org.apache.kafka.streams.KeyValueTimestamp;
  import org.apache.kafka.streams.StreamsBuilder;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.TestInputTopic;
  import org.apache.kafka.streams.Topology;
  import org.apache.kafka.streams.TopologyTestDriver;
  import org.apache.kafka.streams.kstream.Consumed;
  import org.apache.kafka.streams.kstream.Grouped;
  import org.apache.kafka.streams.kstream.KGroupedStream;
  import org.apache.kafka.streams.kstream.KTable;
  import org.apache.kafka.streams.kstream.Materialized;
  import org.apache.kafka.streams.kstream.Named;
  import org.apache.kafka.streams.kstream.Produced;
  import org.apache.kafka.streams.kstream.SessionWindows;
  import org.apache.kafka.streams.kstream.SlidingWindows;
  import org.apache.kafka.streams.kstream.Suppressed;
  import org.apache.kafka.streams.kstream.TimeWindows;
  import org.apache.kafka.streams.kstream.Windowed;
  import org.apache.kafka.streams.state.KeyValueStore;
  import org.apache.kafka.streams.state.SessionStore;
  import org.apache.kafka.streams.state.WindowStore;
  import org.apache.kafka.streams.test.TestRecord;
  import org.apache.kafka.test.TestUtils;
  
  import org.junit.jupiter.api.Test;
  
  import java.time.Duration;
  import java.util.Comparator;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Properties;
  
  import static java.time.Duration.ZERO;
  import static java.time.Duration.ofMillis;
  import static java.util.Arrays.asList;
  import static java.util.Collections.emptyList;
  import static java.util.Collections.singletonList;
  import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxBytes;
  import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxRecords;
  import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.unbounded;
  import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
  import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
  import static org.hamcrest.CoreMatchers.equalTo;
  import static org.hamcrest.MatcherAssert.assertThat;
  
  @SuppressWarnings("deprecation")
  public class SuppressScenarioTest {
      private static final StringDeserializer STRING_DESERIALIZER = new StringDeserializer();
      private static final StringSerializer STRING_SERIALIZER = new StringSerializer();
      private static final Serde<String> STRING_SERDE = Serdes.String();
      private static final LongDeserializer LONG_DESERIALIZER = new LongDeserializer();
      private final Properties config = Utils.mkProperties(Utils.mkMap(
          Utils.mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath())
      ));
  
      @Test
      public void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final KTable<String, Long> valueCounts = builder
              .table(
                  "input",
                  Consumed.with(STRING_SERDE, STRING_SERDE),
                  Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
                      .withCachingDisabled()
                      .withLoggingDisabled()
              )
              .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
              .count();
  
          valueCounts
              .suppress(untilTimeLimit(ZERO, unbounded()))
              .toStream()
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
  
          valueCounts
              .toStream()
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
  
          final Topology topology = builder.build();
  
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v2", 1L);
              inputTopic.pipeInput("k2", "v1", 2L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("v1", 1L, 0L),
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L),
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("v1", 1L, 0L),
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L),
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
              inputTopic.pipeInput("x", "x", 3L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(
                      new KeyValueTimestamp<>("x", 1L, 3L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(
                      new KeyValueTimestamp<>("x", 1L, 3L)
                  )
              );
              inputTopic.pipeInput("x", "y", 4L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("x", 0L, 4L),
                      new KeyValueTimestamp<>("y", 1L, 4L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("x", 0L, 4L),
                      new KeyValueTimestamp<>("y", 1L, 4L)
                  )
              );
          }
      }
  
      @Test
      public void shouldSuppressIntermediateEventsWithTimeLimit() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<String, Long> valueCounts = builder
              .table(
                  "input",
                  Consumed.with(STRING_SERDE, STRING_SERDE),
                  Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
                      .withCachingDisabled()
                      .withLoggingDisabled()
              )
              .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
              .count();
          valueCounts
              .suppress(untilTimeLimit(ofMillis(2L), unbounded()))
              .toStream()
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .toStream()
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v2", 1L);
              inputTopic.pipeInput("k2", "v1", 2L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("v1", 1L, 0L),
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L),
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("v1", 1L, 2L))
              );
              // inserting a dummy "tick" record just to advance stream time
              inputTopic.pipeInput("tick", "tick", 3L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("tick", 1L, 3L))
              );
              // the stream time is now 3, so it's time to emit this record
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("v2", 1L, 1L))
              );
  
  
              inputTopic.pipeInput("tick", "tock", 4L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("tick", 0L, 4L),
                      new KeyValueTimestamp<>("tock", 1L, 4L)
                  )
              );
              // tick is still buffered, since it was first inserted at time 3, and it is only time 4 right now.
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  emptyList()
              );
          }
      }
  
      @Test
      public void shouldSuppressIntermediateEventsWithRecordLimit() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<String, Long> valueCounts = builder
              .table(
                  "input",
                  Consumed.with(STRING_SERDE, STRING_SERDE),
                  Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
                      .withCachingDisabled()
                      .withLoggingDisabled()
              )
              .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
              .count(Materialized.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .suppress(untilTimeLimit(ofMillis(Long.MAX_VALUE), maxRecords(1L).emitEarlyWhenFull()))
              .toStream()
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .toStream()
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          System.out.println(topology.describe());
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v2", 1L);
              inputTopic.pipeInput("k2", "v1", 2L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("v1", 1L, 0L),
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L),
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      // consecutive updates to v1 get suppressed into only the latter.
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L)
                      // the last update won't be evicted until another key comes along.
                  )
              );
              inputTopic.pipeInput("x", "x", 3L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(
                      new KeyValueTimestamp<>("x", 1L, 3L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(
                      // now we see that last update to v1, but we won't see the update to x until it gets evicted
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
          }
      }
  
      @Test
      public void shouldSuppressIntermediateEventsWithBytesLimit() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<String, Long> valueCounts = builder
              .table(
                  "input",
                  Consumed.with(STRING_SERDE, STRING_SERDE),
                  Materialized.<String, String, KeyValueStore<Bytes, byte[]>>with(STRING_SERDE, STRING_SERDE)
                      .withCachingDisabled()
                      .withLoggingDisabled()
              )
              .groupBy((k, v) -> new KeyValue<>(v, k), Grouped.with(STRING_SERDE, STRING_SERDE))
              .count();
          valueCounts
              // this is a bit brittle, but I happen to know that the entries are a little over 100 bytes in size.
              .suppress(untilTimeLimit(ofMillis(Long.MAX_VALUE), maxBytes(200L).emitEarlyWhenFull()))
              .toStream()
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .toStream()
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          System.out.println(topology.describe());
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v2", 1L);
              inputTopic.pipeInput("k2", "v1", 2L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("v1", 1L, 0L),
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L),
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      // consecutive updates to v1 get suppressed into only the latter.
                      new KeyValueTimestamp<>("v1", 0L, 1L),
                      new KeyValueTimestamp<>("v2", 1L, 1L)
                      // the last update won't be evicted until another key comes along.
                  )
              );
              inputTopic.pipeInput("x", "x", 3L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(
                      new KeyValueTimestamp<>("x", 1L, 3L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(
                      // now we see that last update to v1, but we won't see the update to x until it gets evicted
                      new KeyValueTimestamp<>("v1", 1L, 2L)
                  )
              );
          }
      }
  
      @Test
      public void shouldSupportFinalResultsForTimeWindows() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<Windowed<String>, Long> valueCounts = builder
              .stream("input", Consumed.with(STRING_SERDE, STRING_SERDE))
              .groupBy((String k, String v) -> k, Grouped.with(STRING_SERDE, STRING_SERDE))
              .windowedBy(TimeWindows.ofSizeAndGrace(ofMillis(2L), ofMillis(1L)))
              .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("counts").withCachingDisabled());
          valueCounts
              .suppress(untilWindowCloses(unbounded()))
              .toStream()
              .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .toStream()
              .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          System.out.println(topology.describe());
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 1L);
              inputTopic.pipeInput("k1", "v1", 2L);
              inputTopic.pipeInput("k1", "v1", 1L);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 5L);
              // note this last record gets dropped because it is out of the grace period
              inputTopic.pipeInput("k1", "v1", 0L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@0/2]", 1L, 0L),
                      new KeyValueTimestamp<>("[k1@0/2]", 2L, 1L),
                      new KeyValueTimestamp<>("[k1@2/4]", 1L, 2L),
                      new KeyValueTimestamp<>("[k1@0/2]", 3L, 1L),
                      new KeyValueTimestamp<>("[k1@0/2]", 4L, 1L),
                      new KeyValueTimestamp<>("[k1@4/6]", 1L, 5L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@0/2]", 4L, 1L),
                      new KeyValueTimestamp<>("[k1@2/4]", 1L, 2L)
                  )
              );
          }
      }
  
      @Test
      public void shouldSupportFinalResultsForTimeWindowsWithLargeJump() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<Windowed<String>, Long> valueCounts = builder
              .stream("input", Consumed.with(STRING_SERDE, STRING_SERDE))
              .groupBy((String k, String v) -> k, Grouped.with(STRING_SERDE, STRING_SERDE))
              .windowedBy(TimeWindows.ofSizeAndGrace(ofMillis(2L), ofMillis(2L)))
              .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("counts").withCachingDisabled().withKeySerde(STRING_SERDE));
          valueCounts
              .suppress(untilWindowCloses(unbounded()))
              .toStream()
              .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .toStream()
              .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          System.out.println(topology.describe());
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 1L);
              inputTopic.pipeInput("k1", "v1", 2L);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 3L);
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 4L);
              // this update should get dropped, since the previous event advanced the stream time and closed the window.
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 30L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@0/2]", 1L, 0L),
                      new KeyValueTimestamp<>("[k1@0/2]", 2L, 1L),
                      new KeyValueTimestamp<>("[k1@2/4]", 1L, 2L),
                      new KeyValueTimestamp<>("[k1@0/2]", 3L, 1L),
                      new KeyValueTimestamp<>("[k1@2/4]", 2L, 3L),
                      new KeyValueTimestamp<>("[k1@0/2]", 4L, 1L),
                      new KeyValueTimestamp<>("[k1@4/6]", 1L, 4L),
                      new KeyValueTimestamp<>("[k1@30/32]", 1L, 30L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@0/2]", 4L, 1L),
                      new KeyValueTimestamp<>("[k1@2/4]", 2L, 3L),
                      new KeyValueTimestamp<>("[k1@4/6]", 1L, 4L)
                  )
              );
          }
      }
  
      @Test
      public void shouldSupportFinalResultsForSlidingWindows() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<Windowed<String>, Long> valueCounts = builder
                  .stream("input", Consumed.with(STRING_SERDE, STRING_SERDE))
                  .groupBy((String k, String v) -> k, Grouped.with(STRING_SERDE, STRING_SERDE))
                  .windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(ofMillis(5L), ofMillis(15L)))
                  .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("counts").withCachingDisabled().withKeySerde(STRING_SERDE));
          valueCounts
                  .suppress(untilWindowCloses(unbounded()))
                  .toStream()
                  .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
                  .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
                  .toStream()
                  .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
                  .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          System.out.println(topology.describe());
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              inputTopic.pipeInput("k1", "v1", 10L);
              inputTopic.pipeInput("k1", "v1", 11L);
              inputTopic.pipeInput("k1", "v1", 10L);
              inputTopic.pipeInput("k1", "v1", 13L);
              inputTopic.pipeInput("k1", "v1", 10L);
              inputTopic.pipeInput("k1", "v1", 24L);
              // this update should get dropped, since the previous event advanced the stream time and closed the window.
              inputTopic.pipeInput("k1", "v1", 5L);
              inputTopic.pipeInput("k1", "v1", 7L);
              // final record to advance stream time and flush windows
              inputTopic.pipeInput("k1", "v1", 90L);
              final Comparator<TestRecord<String, Long>> comparator =
                  Comparator.comparing((TestRecord<String, Long> o) -> o.getKey())
                      .thenComparing((TestRecord<String, Long> o) -> o.timestamp());
  
              final List<TestRecord<String, Long>> actual = drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER);
              actual.sort(comparator);
              verify(
                  actual,
                  asList(
                      // right window for k1@10 created when k1@11 is processed
                      new KeyValueTimestamp<>("[k1@11/16]", 1L, 11L),
                      // right window for k1@10 updated when k1@13 is processed
                      new KeyValueTimestamp<>("[k1@11/16]", 2L, 13L),
                      // right window for k1@11 created when k1@13 is processed
                      new KeyValueTimestamp<>("[k1@12/17]", 1L, 13L),
                      // left window for k1@24 created when k1@24 is processed
                      new KeyValueTimestamp<>("[k1@19/24]", 1L, 24L),
                      // left window for k1@10 created when k1@10 is processed
                      new KeyValueTimestamp<>("[k1@5/10]", 1L, 10L),
                      // left window for k1@10 updated when k1@10 is processed
                      new KeyValueTimestamp<>("[k1@5/10]", 2L, 10L),
                      // left window for k1@10 updated when k1@10 is processed
                      new KeyValueTimestamp<>("[k1@5/10]", 3L, 10L),
                      // left window for k1@10 updated when k1@5 is processed
                      new KeyValueTimestamp<>("[k1@5/10]", 4L, 10L),
                      // left window for k1@10 updated when k1@7 is processed
                      new KeyValueTimestamp<>("[k1@5/10]", 5L, 10L),
                      // left window for k1@11 created when k1@11 is processed
                      new KeyValueTimestamp<>("[k1@6/11]", 2L, 11L),
                      // left window for k1@11 updated when k1@10 is processed
                      new KeyValueTimestamp<>("[k1@6/11]", 3L, 11L),
                      // left window for k1@11 updated when k1@10 is processed
                      new KeyValueTimestamp<>("[k1@6/11]", 4L, 11L),
                      // left window for k1@11 updated when k1@7 is processed
                      new KeyValueTimestamp<>("[k1@6/11]", 5L, 11L),
                      // left window for k1@13 created when k1@13 is processed
                      new KeyValueTimestamp<>("[k1@8/13]", 4L, 13L),
                      // left window for k1@13 updated when k1@10 is processed
                      new KeyValueTimestamp<>("[k1@8/13]", 5L, 13L),
                      // right window for k1@90 created when k1@90 is processed
                      new KeyValueTimestamp<>("[k1@85/90]", 1L, 90L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@5/10]", 5L, 10L),
                      new KeyValueTimestamp<>("[k1@6/11]", 5L, 11L),
                      new KeyValueTimestamp<>("[k1@8/13]", 5L, 13L),
                      new KeyValueTimestamp<>("[k1@11/16]", 2L, 13L),
                      new KeyValueTimestamp<>("[k1@12/17]", 1L, 13L),
                      new KeyValueTimestamp<>("[k1@19/24]", 1L, 24L)
                  )
              );
          }
      }
  
      @Test
      public void shouldSupportFinalResultsForSessionWindows() {
          final StreamsBuilder builder = new StreamsBuilder();
          final KTable<Windowed<String>, Long> valueCounts = builder
              .stream("input", Consumed.with(STRING_SERDE, STRING_SERDE))
              .groupBy((String k, String v) -> k, Grouped.with(STRING_SERDE, STRING_SERDE))
              .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(ofMillis(5L)))
              .count(Materialized.<String, Long, SessionStore<Bytes, byte[]>>as("counts").withCachingDisabled());
          valueCounts
              .suppress(untilWindowCloses(unbounded()))
              .toStream()
              .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
              .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long()));
          valueCounts
              .toStream()
              .map((final Windowed<String> k, final Long v) -> new KeyValue<>(k.toString(), v))
              .to("output-raw", Produced.with(STRING_SERDE, Serdes.Long()));
          final Topology topology = builder.build();
          System.out.println(topology.describe());
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopic =
                      driver.createInputTopic("input", STRING_SERIALIZER, STRING_SERIALIZER);
              // first window
              inputTopic.pipeInput("k1", "v1", 0L);
              inputTopic.pipeInput("k1", "v1", 5L);
              // arbitrarily disordered records are admitted, because the *window* is not closed until stream-time > window-end + grace
              inputTopic.pipeInput("k1", "v1", 1L);
              // any record in the same partition advances stream time (note the key is different)
              inputTopic.pipeInput("k2", "v1", 11L);
              // late event for first window - this should get dropped from all streams, since the first window is now closed.
              inputTopic.pipeInput("k1", "v1", 5L);
              // just pushing stream time forward to flush the other events through.
              inputTopic.pipeInput("k1", "v1", 30L);
              verify(
                  drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@0/0]", 1L, 0L),
                      new KeyValueTimestamp<>("[k1@0/0]", null, 0L),
                      new KeyValueTimestamp<>("[k1@0/5]", 2L, 5L),
                      new KeyValueTimestamp<>("[k1@0/5]", null, 5L),
                      new KeyValueTimestamp<>("[k1@0/5]", 3L, 5L),
                      new KeyValueTimestamp<>("[k2@11/11]", 1L, 11L),
                      new KeyValueTimestamp<>("[k1@30/30]", 1L, 30L)
                  )
              );
              verify(
                  drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("[k1@0/5]", 3L, 5L),
                      new KeyValueTimestamp<>("[k2@11/11]", 1L, 11L)
                  )
              );
          }
      }
  
      @Test
      public void shouldWorkBeforeGroupBy() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          builder
              .table("topic", Consumed.with(Serdes.String(), Serdes.String()))
              .suppress(untilTimeLimit(ofMillis(10), unbounded()))
              .groupBy(KeyValue::pair, Grouped.with(Serdes.String(), Serdes.String()))
              .count()
              .toStream()
              .to("output", Produced.with(Serdes.String(), Serdes.Long()));
  
          try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), config)) {
              final TestInputTopic<String, String> inputTopic =
                  driver.createInputTopic("topic", STRING_SERIALIZER, STRING_SERIALIZER);
  
              inputTopic.pipeInput("A", "a", 0L);
              inputTopic.pipeInput("tick", "tick", 10L);
  
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, LONG_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("A", 1L, 0L))
              );
          }
      }
  
      @Test
      public void shouldWorkBeforeJoinRight() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final KTable<String, String> left = builder
              .table("left", Consumed.with(Serdes.String(), Serdes.String()));
  
          final KTable<String, String> right = builder
              .table("right", Consumed.with(Serdes.String(), Serdes.String()))
              .suppress(untilTimeLimit(ofMillis(10), unbounded()));
  
          left
              .outerJoin(right, (l, r) -> String.format("(%s,%s)", l, r))
              .toStream()
              .to("output", Produced.with(Serdes.String(), Serdes.String()));
  
          try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), config)) {
              final TestInputTopic<String, String> inputTopicRight =
                  driver.createInputTopic("right", STRING_SERIALIZER, STRING_SERIALIZER);
              final TestInputTopic<String, String> inputTopicLeft =
                      driver.createInputTopic("left", STRING_SERIALIZER, STRING_SERIALIZER);
  
              inputTopicRight.pipeInput("B", "1", 0L);
              inputTopicRight.pipeInput("A", "1", 0L);
              // buffered, no output
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  emptyList()
              );
  
  
              inputTopicRight.pipeInput("tick", "tick", 10L);
              // flush buffer
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("A", "(null,1)", 0L),
                      new KeyValueTimestamp<>("B", "(null,1)", 0L)
                  )
              );
  
  
              inputTopicRight.pipeInput("A", "2", 11L);
              // buffered, no output
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  emptyList()
              );
  
  
              inputTopicLeft.pipeInput("A", "a", 12L);
              // should join with previously emitted right side
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("A", "(a,1)", 12L))
              );
  
  
              inputTopicLeft.pipeInput("B", "b", 12L);
              // should view through to the parent KTable, since B is no longer buffered
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("B", "(b,1)", 12L))
              );
  
  
              inputTopicLeft.pipeInput("A", "b", 13L);
              // should join with previously emitted right side
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("A", "(b,1)", 13L))
              );
  
  
              inputTopicRight.pipeInput("tick", "tick1", 21L);
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("tick", "(null,tick1)", 21), // just a testing artifact
                      new KeyValueTimestamp<>("A", "(b,2)", 13L)
                  )
              );
          }
  
      }
  
  
      @Test
      public void shouldWorkBeforeJoinLeft() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final KTable<String, String> left = builder
              .table("left", Consumed.with(Serdes.String(), Serdes.String()))
              .suppress(untilTimeLimit(ofMillis(10), unbounded()));
  
          final KTable<String, String> right = builder
              .table("right", Consumed.with(Serdes.String(), Serdes.String()));
  
          left
              .outerJoin(right, (l, r) -> String.format("(%s,%s)", l, r))
              .toStream()
              .to("output", Produced.with(Serdes.String(), Serdes.String()));
  
          final Topology topology = builder.build();
          try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
              final TestInputTopic<String, String> inputTopicRight =
                  driver.createInputTopic("right", STRING_SERIALIZER, STRING_SERIALIZER);
              final TestInputTopic<String, String> inputTopicLeft =
                      driver.createInputTopic("left", STRING_SERIALIZER, STRING_SERIALIZER);
  
              inputTopicLeft.pipeInput("B", "1", 0L);
              inputTopicLeft.pipeInput("A", "1", 0L);
              // buffered, no output
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  emptyList()
              );
  
  
              inputTopicLeft.pipeInput("tick", "tick", 10L);
              // flush buffer
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("A", "(1,null)", 0L),
                      new KeyValueTimestamp<>("B", "(1,null)", 0L)
                  )
              );
  
  
              inputTopicLeft.pipeInput("A", "2", 11L);
              // buffered, no output
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  emptyList()
              );
  
  
              inputTopicRight.pipeInput("A", "a", 12L);
              // should join with previously emitted left side
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("A", "(1,a)", 12L))
              );
  
  
              inputTopicRight.pipeInput("B", "b", 12L);
              // should view through to the parent KTable, since B is no longer buffered
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("B", "(1,b)", 12L))
              );
  
  
              inputTopicRight.pipeInput("A", "b", 13L);
              // should join with previously emitted left side
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  singletonList(new KeyValueTimestamp<>("A", "(1,b)", 13L))
              );
  
  
              inputTopicLeft.pipeInput("tick", "tick1", 21L);
              verify(
                  drainProducerRecords(driver, "output", STRING_DESERIALIZER, STRING_DESERIALIZER),
                  asList(
                      new KeyValueTimestamp<>("tick", "(tick1,null)", 21), // just a testing artifact
                      new KeyValueTimestamp<>("A", "(2,b)", 13L)
                  )
              );
          }
  
      }
  
      @Test
      public void shouldWorkWithCogroupedTimeWindows() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final KGroupedStream<String, String> stream1 = builder.stream("one", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
          final KGroupedStream<String, String> stream2 = builder.stream("two", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
          stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
              .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(15)))
              .aggregate(() -> "", Named.as("test"), Materialized.as("store"))
              .suppress(Suppressed.untilWindowCloses(unbounded()))
              .toStream();
      }
  
      @Test
      public void shouldWorkWithCogroupedSlidingWindows() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final KGroupedStream<String, String> stream1 = builder.stream("one", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
          final KGroupedStream<String, String> stream2 = builder.stream("two", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
          stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
              .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(15)))
              .aggregate(() -> "", Named.as("test"), Materialized.as("store"))
              .suppress(Suppressed.untilWindowCloses(unbounded()))
              .toStream();
      }
  
      @Test
      public void shouldWorkWithCogroupedSessionWindows() {
          final StreamsBuilder builder = new StreamsBuilder();
  
          final KGroupedStream<String, String> stream1 = builder.stream("one", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
          final KGroupedStream<String, String> stream2 = builder.stream("two", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
          stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
              .windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(15), Duration.ofMinutes(5)))
              .aggregate(() -> "", (k, v1, v2) -> "", Named.as("test"), Materialized.as("store"))
              .suppress(Suppressed.untilWindowCloses(unbounded()))
              .toStream();
      }
  
      private static <K, V> void verify(final List<TestRecord<K, V>> results,
                                        final List<KeyValueTimestamp<K, V>> expectedResults) {
          if (results.size() != expectedResults.size()) {
              throw new AssertionError(printRecords(results) + " != " + expectedResults);
          }
          final Iterator<KeyValueTimestamp<K, V>> expectedIterator = expectedResults.iterator();
          for (final TestRecord<K, V> result : results) {
              final KeyValueTimestamp<K, V> expected = expectedIterator.next();
              try {
                  assertThat(result, equalTo(new TestRecord<>(expected.key(), expected.value(), null, expected.timestamp())));
              } catch (final AssertionError e) {
                  throw new AssertionError(printRecords(results) + " != " + expectedResults, e);
              }
          }
      }
  
      private static <K, V> List<TestRecord<K, V>> drainProducerRecords(final TopologyTestDriver driver,
                                                                        final String topic,
                                                                        final Deserializer<K> keyDeserializer,
                                                                        final Deserializer<V> valueDeserializer) {
          return driver.createOutputTopic(topic, keyDeserializer, valueDeserializer).readRecordsToList();
      }
  
      private static <K, V> String printRecords(final List<TestRecord<K, V>> result) {
          final StringBuilder resultStr = new StringBuilder();
          resultStr.append("[\n");
          for (final TestRecord<?, ?> record : result) {
              resultStr.append("  ").append(record).append("\n");
          }
          resultStr.append("]");
          return resultStr.toString();
      }
  }
  