KAFKA-7223: Part 3 preview

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/GroupedStreamAggregateBuilder.java

@@ -67,7 +67,9 @@ class GroupedStreamAggregateBuilder<K, V> {
     <KR, T> KTable<KR, T> build(final KStreamAggProcessorSupplier<K, KR, V, T> aggregateSupplier,
                                 final String functionName,
                                 final StoreBuilder<? extends StateStore> storeBuilder,
-                                final boolean isQueryable) {
+                                final boolean isQueryable,
+                                final Serde<KR> keySerde,
+                                final Serde<T> valueSerde) {
 
         final String aggFunctionName = builder.newProcessorName(functionName);
 
@@ -98,6 +100,8 @@ <KR, T> KTable<KR, T> build(final KStreamAggProcessorSupplier<K, KR, V, T> aggre
         return new KTableImpl<>(builder,
                                 aggFunctionName,
                                 aggregateSupplier,
+                                keySerde,
+                                valueSerde,
                                 sourceName.equals(this.name) ? sourceNodes : Collections.singleton(sourceName),
                                 storeBuilder.name(),
                                 isQueryable,

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

@@ -92,7 +92,8 @@ public KTable<K, V> reduce(final Reducer<V> reducer,
         return doAggregate(
             new KStreamReduce<>(materializedInternal.storeName(), reducer),
             REDUCE_NAME,
-            materializedInternal
+            materializedInternal,
+            materializedInternal.keySerde()
         );
     }
 
@@ -114,7 +115,8 @@ public <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
         return doAggregate(
             new KStreamAggregate<>(materializedInternal.storeName(), initializer, aggregator),
             AGGREGATE_NAME,
-            materializedInternal
+            materializedInternal,
+            materializedInternal.keySerde()
         );
     }
 
@@ -156,7 +158,9 @@ private KTable<K, Long> doCount(final Materialized<K, Long, KeyValueStore<Bytes,
         return doAggregate(
             new KStreamAggregate<>(materializedInternal.storeName(), aggregateBuilder.countInitializer, aggregateBuilder.countAggregator),
             AGGREGATE_NAME,
-            materializedInternal);
+            materializedInternal,
+            materializedInternal.keySerde()
+        );
     }
 
     @Override
@@ -191,12 +195,15 @@ public SessionWindowedKStream<K, V> windowedBy(final SessionWindows windows) {
 
     private <KR, T> KTable<KR, T> doAggregate(final KStreamAggProcessorSupplier<K, KR, V, T> aggregateSupplier,
                                               final String functionName,
-                                              final MaterializedInternal<K, T, KeyValueStore<Bytes, byte[]>> materializedInternal) {
+                                              final MaterializedInternal<K, T, KeyValueStore<Bytes, byte[]>> materializedInternal,
+                                              final Serde<KR> resultKeySerde) {
         return aggregateBuilder.build(
             aggregateSupplier,
             functionName,
             new KeyValueStoreMaterializer<>(materializedInternal).materialize(),
-            materializedInternal.isQueryable()
+            materializedInternal.isQueryable(),
+            resultKeySerde,
+            materializedInternal.valueSerde()
         );
     }
 }

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

@@ -93,6 +93,8 @@ private <T> KTable<K, T> doAggregate(final ProcessorSupplier<K, Change<V>> aggre
         return new KTableImpl<>(builder,
                                 funcName,
                                 aggregateSupplier,
+                                materialized.keySerde(),
+                                materialized.valueSerde(),
                                 Collections.singleton(sourceName),
                                 materialized.storeName(),
                                 materialized.isQueryable(),

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

@@ -365,7 +365,11 @@ public KTable<K, V> suppress(final Suppressed<K> suppressed) {
         final String name = builder.newProcessorName(SUPPRESS_NAME);
 
         final ProcessorSupplier<K, Change<V>> suppressionSupplier =
-            () -> new KTableSuppressProcessor<>(buildSuppress(suppressed));
+            () -> new KTableSuppressProcessor<>(
+                buildSuppress(suppressed),
+                keySerde,
+                valSerde == null ? null : new FullChangeSerde<>(valSerde)
+            );
 
         final ProcessorParameters<K, Change<V>> processorParameters = new ProcessorParameters<>(
             suppressionSupplier,

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

@@ -28,6 +28,7 @@
 import org.apache.kafka.streams.kstream.SessionWindowedKStream;
 import org.apache.kafka.streams.kstream.SessionWindows;
 import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.WindowedSerdes;
 import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
 import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
 import org.apache.kafka.streams.state.SessionStore;
@@ -100,7 +101,9 @@ private KTable<Windowed<K>, Long> doCount(final Materialized<K, Long, SessionSto
             ),
             AGGREGATE_NAME,
             materialize(materializedInternal),
-            materializedInternal.isQueryable()
+            materializedInternal.isQueryable(),
+            getWindowedSerde(materializedInternal.keySerde()),
+            materializedInternal.valueSerde()
         );
     }
 
@@ -134,7 +137,9 @@ public KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
             ),
             REDUCE_NAME,
             materialize(materializedInternal),
-            materializedInternal.isQueryable()
+            materializedInternal.isQueryable(),
+            getWindowedSerde(materializedInternal.keySerde()),
+            materializedInternal.valueSerde()
         );
     }
 
@@ -170,7 +175,9 @@ public <VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
             ),
             AGGREGATE_NAME,
             materialize(materializedInternal),
-            materializedInternal.isQueryable()
+            materializedInternal.isQueryable(),
+            getWindowedSerde(materializedInternal.keySerde()),
+            materializedInternal.valueSerde()
         );
     }
 
@@ -221,4 +228,8 @@ private Merger<K, V> mergerForAggregator(final Aggregator<K, V, V> aggregator) {
     private Aggregator<K, V, V> aggregatorForReducer(final Reducer<V> reducer) {
         return (aggKey, value, aggregate) -> aggregate == null ? value : reducer.apply(aggregate, value);
     }
+
+    private static <T> Serde<Windowed<T>> getWindowedSerde(final Serde<T> rawSerde) {
+        return rawSerde == null ? null : new WindowedSerdes.TimeWindowedSerde<>(rawSerde);
+    }
 }

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

@@ -27,6 +27,7 @@
 import org.apache.kafka.streams.kstream.TimeWindowedKStream;
 import org.apache.kafka.streams.kstream.Window;
 import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.WindowedSerdes;
 import org.apache.kafka.streams.kstream.Windows;
 import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
 import org.apache.kafka.streams.state.StoreBuilder;
@@ -100,11 +101,14 @@ private KTable<Windowed<K>, Long> doCount(final Materialized<K, Long, WindowStor
             ),
             AGGREGATE_NAME,
             materialize(materializedInternal),
-            materializedInternal.isQueryable()
+            materializedInternal.isQueryable(),
+            getWindowedSerde(materializedInternal.keySerde()),
+            materializedInternal.valueSerde()
         );
     }
 
 
+
     @Override
     public <VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                                   final Aggregator<? super K, ? super V, VR> aggregator) {
@@ -132,7 +136,10 @@ public <VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
             ),
             AGGREGATE_NAME,
             materialize(materializedInternal),
-            materializedInternal.isQueryable());
+            materializedInternal.isQueryable(),
+            getWindowedSerde(materializedInternal.keySerde()),
+            materializedInternal.valueSerde()
+        );
     }
 
     @Override
@@ -159,7 +166,9 @@ public KTable<Windowed<K>, V> reduce(final Reducer<V> reducer, final Materialize
             new KStreamWindowReduce<>(windows, materializedInternal.storeName(), reducer),
             REDUCE_NAME,
             materialize(materializedInternal),
-            materializedInternal.isQueryable()
+            materializedInternal.isQueryable(),
+            getWindowedSerde(materializedInternal.keySerde()),
+            materializedInternal.valueSerde()
         );
     }
 
@@ -226,4 +235,8 @@ private <VR> StoreBuilder<WindowStore<K, VR>> materialize(final MaterializedInte
         }
         return builder;
     }
+
+    private static <T> Serde<Windowed<T>> getWindowedSerde(final Serde<T> rawSerde) {
+        return rawSerde == null ? null : new WindowedSerdes.TimeWindowedSerde<>(rawSerde);
+    }
 }

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

@@ -21,7 +21,7 @@
 import static org.apache.kafka.streams.kstream.internals.suppress.BufferFullStrategy.SHUT_DOWN;
 
 abstract class BufferConfigImpl<BC extends Suppressed.BufferConfig<BC>> implements Suppressed.BufferConfig<BC> {
-    public abstract long maxKeys();
+    public abstract long maxRecords();
 
     public abstract long maxBytes();
 
@@ -39,12 +39,12 @@ public Suppressed.StrictBufferConfig withNoBound() {
 
     @Override
     public Suppressed.StrictBufferConfig shutDownWhenFull() {
-        return new StrictBufferConfigImpl(maxKeys(), maxBytes(), SHUT_DOWN);
+        return new StrictBufferConfigImpl(maxRecords(), maxBytes(), SHUT_DOWN);
     }
 
     @Override
     public Suppressed.BufferConfig emitEarlyWhenFull() {
-        return new EagerBufferConfigImpl(maxKeys(), maxBytes());
+        return new EagerBufferConfigImpl(maxRecords(), maxBytes());
     }
 
     @Override