KAFKA-7223: add tests in preparation for suppression

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

@@ -0,0 +1,117 @@
+/*
+ * 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.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();
+        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) {
+                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 buffer.array();
+            }
+
+            @Override
+            public void close() {
+                innerSerializer.close();
+            }
+        };
+    }
+
+    @Override
+    public Deserializer<Change<T>> deserializer() {
+        final Deserializer<T> innerDeserializer = inner.deserializer();
+        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) {
+                final ByteBuffer buffer = ByteBuffer.wrap(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<>(oldValue, newValue);
+            }
+
+            @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/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
@@ -55,10 +65,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();
         }