KAFKA-7223: add tests in preparation for suppression

streams/src/main/java/org/apache/kafka/streams/kstream/Suppressed.java

@@ -155,6 +155,6 @@ static <K extends Windowed> Suppressed<K> untilWindowCloses(final StrictBufferCo
      * @return a suppression configuration
      */
     static <K> Suppressed<K> untilTimeLimit(final Duration timeToWaitForMoreEvents, final BufferConfig bufferConfig) {
-        return new SuppressedImpl<>(timeToWaitForMoreEvents, bufferConfig, null);
+        return new SuppressedImpl<>(timeToWaitForMoreEvents, bufferConfig, null, false);
     }
 }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java

@@ -0,0 +1,128 @@
+/*
+ * 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.ByteBufferDeserializer;
+import org.apache.kafka.common.serialization.ByteBufferSerializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+public class FullChangeSerde<T> implements Serde<Change<T>> {
+    private final Serde<T> inner;
+
+    public FullChangeSerde(final Serde<T> inner) {
+        this.inner = requireNonNull(inner);
+    }
+
+    @Override
+    public void configure(final Map<String, ?> configs, final boolean isKey) {
+        inner.configure(configs, isKey);
+    }
+
+    @Override
+    public void close() {
+        inner.close();
+    }
+
+    @Override
+    public Serializer<Change<T>> serializer() {
+        final Serializer<T> innerSerializer = inner.serializer();
+        final ByteBufferSerializer byteBufferSerializer = new ByteBufferSerializer();
+
+        return new Serializer<Change<T>>() {
+            @Override
+            public void configure(final Map<String, ?> configs, final boolean isKey) {
+                innerSerializer.configure(configs, isKey);
+            }
+
+            @Override
+            public byte[] serialize(final String topic, final Change<T> data) {
+                if (data == null) {
+                    return null;
+                }
+                final byte[] oldBytes = data.oldValue == null ? null : innerSerializer.serialize(topic, data.oldValue);
+                final int oldSize = oldBytes == null ? -1 : oldBytes.length;
+                final byte[] newBytes = data.newValue == null ? null : innerSerializer.serialize(topic, data.newValue);
+                final int newSize = newBytes == null ? -1 : newBytes.length;
+
+                final ByteBuffer buffer = ByteBuffer.allocate(
+                    4 + (oldSize == -1 ? 0 : oldSize) + 4 + (newSize == -1 ? 0 : newSize)
+                );
+                buffer.putInt(oldSize);
+                if (oldBytes != null) {
+                    buffer.put(oldBytes);
+                }
+                buffer.putInt(newSize);
+                if (newBytes != null) {
+                    buffer.put(newBytes);
+                }
+                return byteBufferSerializer.serialize(null, buffer);
+            }
+
+            @Override
+            public void close() {
+                innerSerializer.close();
+            }
+        };
+    }
+
+    @Override
+    public Deserializer<Change<T>> deserializer() {
+        final Deserializer<T> innerDeserializer = inner.deserializer();
+        final ByteBufferDeserializer byteBufferDeserializer = new ByteBufferDeserializer();
+        return new Deserializer<Change<T>>() {
+            @Override
+            public void configure(final Map<String, ?> configs, final boolean isKey) {
+                innerDeserializer.configure(configs, isKey);
+            }
+
+            @Override
+            public Change<T> deserialize(final String topic, final byte[] data) {
+                if (data == null) {
+                    return null;
+                }
+                final ByteBuffer buffer = byteBufferDeserializer.deserialize(null, data);
+
+                final int oldSize = buffer.getInt();
+                final byte[] oldBytes = oldSize == -1 ? null : new byte[oldSize];
+                if (oldBytes != null) {
+                    buffer.get(oldBytes);
+                }
+                final T oldValue = oldBytes == null ? null : innerDeserializer.deserialize(topic, oldBytes);
+
+                final int newSize = buffer.getInt();
+                final byte[] newBytes = newSize == -1 ? null : new byte[newSize];
+                if (newBytes != null) {
+                    buffer.get(newBytes);
+                }
+                final T newValue = newBytes == null ? null : innerDeserializer.deserialize(topic, newBytes);
+                return new Change<>(newValue, oldValue);
+            }
+
+            @Override
+            public void close() {
+                innerDeserializer.close();
+            }
+        };
+    }
+}

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java

@@ -364,8 +364,13 @@ public <K1> KStream<K1, V> toStream(final KeyValueMapper<? super K, ? super V, ?
     public KTable<K, V> suppress(final Suppressed<K> suppressed) {
         final String name = builder.newProcessorName(SUPPRESS_NAME);
 
+        // TODO: follow-up pr to forward the k/v serdes
         final ProcessorSupplier<K, Change<V>> suppressionSupplier =
-            () -> new KTableSuppressProcessor<>(buildSuppress(suppressed));
+            () -> new KTableSuppressProcessor<>(
+                buildSuppress(suppressed),
+                null,
+                null
+            );
 
         final ProcessorParameters<K, Change<V>> processorParameters = new ProcessorParameters<>(
             suppressionSupplier,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java

@@ -34,7 +34,8 @@ public SuppressedImpl<K> buildFinalResultsSuppression(final Duration gracePeriod
         return new SuppressedImpl<>(
             gracePeriod,
             bufferConfig,
-            (ProcessorContext context, K key) -> key.window().end()
+            (ProcessorContext context, K key) -> key.window().end(),
+            true
         );
     }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java

@@ -16,19 +16,29 @@
  */
 package org.apache.kafka.streams.kstream.internals.suppress;
 
+import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.streams.kstream.internals.Change;
 import org.apache.kafka.streams.processor.Processor;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
 
 import java.time.Duration;
 
+import static java.util.Objects.requireNonNull;
+
 public class KTableSuppressProcessor<K, V> implements Processor<K, Change<V>> {
     private final SuppressedImpl<K> suppress;
     private InternalProcessorContext internalProcessorContext;
 
-    public KTableSuppressProcessor(final SuppressedImpl<K> suppress) {
-        this.suppress = suppress;
+    private final Serde<K> keySerde;
+    private final Serde<Change<V>> valueSerde;
+
+    public KTableSuppressProcessor(final SuppressedImpl<K> suppress,
+                                   final Serde<K> keySerde,
+                                   final Serde<Change<V>> valueSerde) {
+        this.suppress = requireNonNull(suppress);
+        this.keySerde = keySerde;
+        this.valueSerde = valueSerde;
     }
 
     @Override
@@ -39,12 +49,18 @@ public void init(final ProcessorContext context) {
     @Override
     public void process(final K key, final Change<V> value) {
         if (suppress.getTimeToWaitForMoreEvents() == Duration.ZERO && definedRecordTime(key) <= internalProcessorContext.streamTime()) {
-            internalProcessorContext.forward(key, value);
+            if (shouldForward(value)) {
+                internalProcessorContext.forward(key, value);
+            } // else skip
         } else {
             throw new NotImplementedException();
         }
     }
 
+    private boolean shouldForward(final Change<V> value) {
+        return !(value.newValue == null && suppress.suppressTombstones());
+    }
+
     private long definedRecordTime(final K key) {
         return suppress.getTimeDefinition().time(internalProcessorContext, key);
     }
@@ -55,10 +71,14 @@ public void close() {
 
     @Override
     public String toString() {
-        return "KTableSuppressProcessor{suppress=" + suppress + '}';
+        return "KTableSuppressProcessor{" +
+            "suppress=" + suppress +
+            ", keySerde=" + keySerde +
+            ", valueSerde=" + valueSerde +
+            '}';
     }
 
-    static class NotImplementedException extends RuntimeException {
+    public static class NotImplementedException extends RuntimeException {
         NotImplementedException() {
             super();
         }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedImpl.java

@@ -29,13 +29,16 @@ public class SuppressedImpl<K> implements Suppressed<K> {
     private final BufferConfig bufferConfig;
     private final Duration timeToWaitForMoreEvents;
     private final TimeDefinition<K> timeDefinition;
+    private final boolean suppressTombstones;
 
     public SuppressedImpl(final Duration suppressionTime,
                           final BufferConfig bufferConfig,
-                          final TimeDefinition<K> timeDefinition) {
+                          final TimeDefinition<K> timeDefinition,
+                          final boolean suppressTombstones) {
         this.timeToWaitForMoreEvents = suppressionTime == null ? DEFAULT_SUPPRESSION_TIME : suppressionTime;
         this.timeDefinition = timeDefinition == null ? (context, anyKey) -> context.timestamp() : timeDefinition;
         this.bufferConfig = bufferConfig == null ? DEFAULT_BUFFER_CONFIG : bufferConfig;
+        this.suppressTombstones = suppressTombstones;
     }
 
     interface TimeDefinition<K> {
@@ -73,4 +76,8 @@ public String toString() {
             ", timeDefinition=" + timeDefinition +
             '}';
     }
+
+    boolean suppressTombstones() {
+        return suppressTombstones;
+    }
 }