KAFKA-7223: Suppress API with only immediate emit

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

@@ -389,6 +389,16 @@ <VR> KTable<K, VR> mapValues(final ValueMapperWithKey<? super K, ? super V, ? ex
      */
     <KR> KStream<KR, V> toStream(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper);
 
+    /**
+     * Suppress some updates from this changelog stream, determined by the supplied {@link Suppressed} configuration.
+     *
+     * This controls what updates downstream table and stream operations will receive.
+     *
+     * @param suppressed Configuration object determining what, if any, updates to suppress
+     * @return A new KTable with the desired suppression characteristics.
+     */
+    KTable<K, V> suppress(final Suppressed<K> suppressed);
+
     /**
      * Create a new {@code KTable} by transforming the value of each record in this {@code KTable} into a new value
      * (with possibly a new type), with default serializers, deserializers, and state store.

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

@@ -0,0 +1,160 @@
+/*
+ * 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;
+
+import org.apache.kafka.streams.kstream.internals.suppress.EagerBufferConfigImpl;
+import org.apache.kafka.streams.kstream.internals.suppress.FinalResultsSuppressionBuilder;
+import org.apache.kafka.streams.kstream.internals.suppress.StrictBufferConfigImpl;
+import org.apache.kafka.streams.kstream.internals.suppress.SuppressedImpl;
+
+import java.time.Duration;
+
+public interface Suppressed<K> {
+
+    /**
+     * Marker interface for a buffer configuration that is "strict" in the sense that it will strictly
+     * enforce the time bound and never emit early.
+     */
+    interface StrictBufferConfig extends BufferConfig<StrictBufferConfig> {
+
+    }
+
+    interface BufferConfig<BC extends BufferConfig<BC>> {
+        /**
+         * Create a size-constrained buffer in terms of the maximum number of keys it will store.
+         */
+        static BufferConfig<?> maxRecords(final long recordLimit) {
+            return new EagerBufferConfigImpl(recordLimit, Long.MAX_VALUE);
+        }
+
+        /**
+         * Set a size constraint on the buffer in terms of the maximum number of keys it will store.
+         */
+        BC withMaxRecords(final long recordLimit);
+
+        /**
+         * Create a size-constrained buffer in terms of the maximum number of bytes it will use.
+         */
+        static BufferConfig<?> maxBytes(final long byteLimit) {
+            return new EagerBufferConfigImpl(Long.MAX_VALUE, byteLimit);
+        }
+
+        /**
+         * Set a size constraint on the buffer, the maximum number of bytes it will use.
+         */
+        BC withMaxBytes(final long byteLimit);
+
+        /**
+         * Create a buffer unconstrained by size (either keys or bytes).
+         *
+         * As a result, the buffer will consume as much memory as it needs, dictated by the time bound.
+         *
+         * If there isn't enough heap available to meet the demand, the application will encounter an
+         * {@link OutOfMemoryError} and shut down (not guaranteed to be a graceful exit). Also, note that
+         * JVM processes under extreme memory pressure may exhibit poor GC behavior.
+         *
+         * This is a convenient option if you doubt that your buffer will be that large, but also don't
+         * wish to pick particular constraints, such as in testing.
+         *
+         * This buffer is "strict" in the sense that it will enforce the time bound or crash.
+         * It will never emit early.
+         */
+        static StrictBufferConfig unbounded() {
+            return new StrictBufferConfigImpl();
+        }
+
+        /**
+         * Set the buffer to be unconstrained by size (either keys or bytes).
+         *
+         * As a result, the buffer will consume as much memory as it needs, dictated by the time bound.
+         *
+         * If there isn't enough heap available to meet the demand, the application will encounter an
+         * {@link OutOfMemoryError} and shut down (not guaranteed to be a graceful exit). Also, note that
+         * JVM processes under extreme memory pressure may exhibit poor GC behavior.
+         *
+         * This is a convenient option if you doubt that your buffer will be that large, but also don't
+         * wish to pick particular constraints, such as in testing.
+         *
+         * This buffer is "strict" in the sense that it will enforce the time bound or crash.
+         * It will never emit early.
+         */
+        StrictBufferConfig withNoBound();
+
+        /**
+         * Set the buffer to gracefully shut down the application when any of its constraints are violated
+         *
+         * This buffer is "strict" in the sense that it will enforce the time bound or shut down.
+         * It will never emit early.
+         */
+        StrictBufferConfig shutDownWhenFull();
+
+        /**
+         * Sets the buffer to use on-disk storage if it requires more memory than the constraints allow.
+         *
+         * This buffer is "strict" in the sense that it will never emit early.
+         */
+        StrictBufferConfig spillToDiskWhenFull();
+
+        /**
+         * Set the buffer to just emit the oldest records when any of its constraints are violated.
+         *
+         * This buffer is "not strict" in the sense that it may emit early, so it is suitable for reducing
+         * duplicate results downstream, but does not promise to eliminate them.
+         */
+        BufferConfig emitEarlyWhenFull();
+    }
+
+    /**
+     * Configure the suppression to emit only the "final results" from the window.
+     *
+     * By default all Streams operators emit results whenever new results are available.
+     * This includes windowed operations.
+     *
+     * This configuration will instead emit just one result per key for each window, guaranteeing
+     * to deliver only the final result. This option is suitable for use cases in which the business logic
+     * requires a hard guarantee that only the final result is propagated. For example, sending alerts.
+     *
+     * To accomplish this, the operator will buffer events from the window until the window close (that is,
+     * until the end-time passes, and additionally until the grace period expires). Since windowed operators
+     * are required to reject late events for a window whose grace period is expired, there is an additional
+     * guarantee that the final results emitted from this suppression will match any queriable state upstream.
+     *
+     * @param bufferConfig A configuration specifying how much space to use for buffering intermediate results.
+     *                     This is required to be a "strict" config, since it would violate the "final results"
+     *                     property to emit early and then issue an update later.
+     * @param <K> The key type for the KTable to apply this suppression to. "Final results" mode is only available
+     *           on Windowed KTables (this is enforced by the type parameter).
+     * @return a "final results" mode suppression configuration
+     */
+    static <K extends Windowed> Suppressed<K> untilWindowCloses(final StrictBufferConfig bufferConfig) {
+        return new FinalResultsSuppressionBuilder<>(bufferConfig);
+    }
+
+    /**
+     * Configure the suppression to wait {@code timeToWaitForMoreEvents} amount of time after receiving a record
+     * before emitting it further downstream. If another record for the same key arrives in the mean time, it replaces
+     * the first record in the buffer but does <em>not</em> re-start the timer.
+     *
+     * @param timeToWaitForMoreEvents The amount of time to wait, per record, for new events.
+     * @param bufferConfig A configuration specifying how much space to use for buffering intermediate results.
+     * @param <K> The key type for the KTable to apply this suppression to.
+     * @return a suppression configuration
+     */
+    static <K> Suppressed<K> untilTimeLimit(final Duration timeToWaitForMoreEvents, final BufferConfig bufferConfig) {
+        return new SuppressedImpl<>(timeToWaitForMoreEvents, bufferConfig, null);
+    }
+}

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

@@ -38,7 +38,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
-class KStreamSessionWindowAggregate<K, V, Agg> implements KStreamAggProcessorSupplier<K, Windowed<K>, V, Agg> {
+public class KStreamSessionWindowAggregate<K, V, Agg> implements KStreamAggProcessorSupplier<K, Windowed<K>, V, Agg> {
     private static final Logger LOG = LoggerFactory.getLogger(KStreamSessionWindowAggregate.class);
 
     private final String storeName;
@@ -49,11 +49,11 @@ class KStreamSessionWindowAggregate<K, V, Agg> implements KStreamAggProcessorSup
 
     private boolean sendOldValues = false;
 
-    KStreamSessionWindowAggregate(final SessionWindows windows,
-                                  final String storeName,
-                                  final Initializer<Agg> initializer,
-                                  final Aggregator<? super K, ? super V, Agg> aggregator,
-                                  final Merger<? super K, Agg> sessionMerger) {
+    public KStreamSessionWindowAggregate(final SessionWindows windows,
+                                         final String storeName,
+                                         final Initializer<Agg> initializer,
+                                         final Aggregator<? super K, ? super V, Agg> aggregator,
+                                         final Merger<? super K, Agg> sessionMerger) {
         this.windows = windows;
         this.storeName = storeName;
         this.initializer = initializer;
@@ -66,6 +66,10 @@ public Processor<K, V> get() {
         return new KStreamSessionWindowAggregateProcessor();
     }
 
+    public SessionWindows windows() {
+        return windows;
+    }
+
     @Override
     public void enableSendingOldValues() {
         sendOldValues = true;

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

@@ -44,10 +44,10 @@ public class KStreamWindowAggregate<K, V, Agg, W extends Window> implements KStr
 
     private boolean sendOldValues = false;
 
-    KStreamWindowAggregate(final Windows<W> windows,
-                           final String storeName,
-                           final Initializer<Agg> initializer,
-                           final Aggregator<? super K, ? super V, Agg> aggregator) {
+    public KStreamWindowAggregate(final Windows<W> windows,
+                                  final String storeName,
+                                  final Initializer<Agg> initializer,
+                                  final Aggregator<? super K, ? super V, Agg> aggregator) {
         this.windows = windows;
         this.storeName = storeName;
         this.initializer = initializer;
@@ -59,6 +59,10 @@ public Processor<K, V> get() {
         return new KStreamWindowAggregateProcessor();
     }
 
+    public Windows<W> windows() {
+        return windows;
+    }
+
     @Override
     public void enableSendingOldValues() {
         sendOldValues = true;

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

@@ -26,21 +26,30 @@
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Predicate;
 import org.apache.kafka.streams.kstream.Serialized;
+import org.apache.kafka.streams.kstream.Suppressed;
 import org.apache.kafka.streams.kstream.ValueJoiner;
 import org.apache.kafka.streams.kstream.ValueMapper;
 import org.apache.kafka.streams.kstream.ValueMapperWithKey;
 import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+import org.apache.kafka.streams.kstream.Windowed;
 import org.apache.kafka.streams.kstream.internals.graph.KTableKTableJoinNode;
 import org.apache.kafka.streams.kstream.internals.graph.ProcessorGraphNode;
 import org.apache.kafka.streams.kstream.internals.graph.ProcessorParameters;
 import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
 import org.apache.kafka.streams.kstream.internals.graph.TableProcessorNode;
+import org.apache.kafka.streams.kstream.internals.suppress.FinalResultsSuppressionBuilder;
+import org.apache.kafka.streams.kstream.internals.suppress.KTableSuppressProcessor;
+import org.apache.kafka.streams.kstream.internals.suppress.SuppressedImpl;
 import org.apache.kafka.streams.processor.ProcessorSupplier;
 import org.apache.kafka.streams.state.KeyValueStore;
 
+import java.time.Duration;
+import java.util.Collections;
 import java.util.Objects;
 import java.util.Set;
 
+import static org.apache.kafka.streams.kstream.internals.graph.GraphGraceSearchUtil.findAndVerifyWindowGrace;
+
 /**
  * The implementation class of {@link KTable}.
  *
@@ -66,6 +75,8 @@ public class KTableImpl<K, S, V> extends AbstractStream<K> implements KTable<K,
 
     private static final String SELECT_NAME = "KTABLE-SELECT-";
 
+    private static final String SUPPRESS_NAME = "KTABLE-SUPPRESS-";
+
     private static final String TOSTREAM_NAME = "KTABLE-TOSTREAM-";
 
     private static final String TRANSFORMVALUES_NAME = "KTABLE-TRANSFORMVALUES-";
@@ -349,6 +360,53 @@ public <K1> KStream<K1, V> toStream(final KeyValueMapper<? super K, ? super V, ?
         return toStream().selectKey(mapper);
     }
 
+    @Override
+    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));
+
+        final ProcessorParameters<K, Change<V>> processorParameters = new ProcessorParameters<>(
+            suppressionSupplier,
+            name
+        );
+
+        final ProcessorGraphNode<K, Change<V>> node = new ProcessorGraphNode<>(name, processorParameters, false);
+
+        builder.addGraphNode(streamsGraphNode, node);
+
+        return new KTableImpl<K, S, V>(
+            builder,
+            name,
+            suppressionSupplier,
+            keySerde,
+            valSerde,
+            Collections.singleton(this.name),
+            null,
+            false,
+            node
+        );
+    }
+
+    @SuppressWarnings("unchecked")
+    private SuppressedImpl<K> buildSuppress(final Suppressed<K> suppress) {
+        if (suppress instanceof FinalResultsSuppressionBuilder) {
+            final long grace = findAndVerifyWindowGrace(streamsGraphNode);
+
+            final FinalResultsSuppressionBuilder<?> builder = (FinalResultsSuppressionBuilder) suppress;
+
+            final SuppressedImpl<? extends Windowed> finalResultsSuppression =
+                builder.buildFinalResultsSuppression(Duration.ofMillis(grace));
+
+            return (SuppressedImpl<K>) finalResultsSuppression;
+        } else if (suppress instanceof SuppressedImpl) {
+            return (SuppressedImpl<K>) suppress;
+        } else {
+            throw new IllegalArgumentException("Custom subclasses of Suppressed are not allowed.");
+        }
+    }
+
     @Override
     public <V1, R> KTable<K, R> join(final KTable<K, V1> other,
                                      final ValueJoiner<? super V, ? super V1, ? extends R> joiner) {
@@ -492,12 +550,12 @@ private <V1, R> KTable<K, R> buildJoin(final AbstractStream<K> other,
         final ProcessorParameters joinMergeProcessorParameters = new ProcessorParameters(joinMerge, joinMergeName);
 
         kTableJoinNodeBuilder.withJoinMergeProcessorParameters(joinMergeProcessorParameters)
-            .withJoinOtherProcessorParameters(joinOtherProcessorParameters)
-            .withJoinThisProcessorParameters(joinThisProcessorParameters)
-            .withJoinThisStoreNames(valueGetterSupplier().storeNames())
-            .withJoinOtherStoreNames(((KTableImpl) other).valueGetterSupplier().storeNames())
-            .withOtherJoinSideNodeName(((KTableImpl) other).name)
-            .withThisJoinSideNodeName(name);
+                             .withJoinOtherProcessorParameters(joinOtherProcessorParameters)
+                             .withJoinThisProcessorParameters(joinThisProcessorParameters)
+                             .withJoinThisStoreNames(valueGetterSupplier().storeNames())
+                             .withJoinOtherStoreNames(((KTableImpl) other).valueGetterSupplier().storeNames())
+                             .withOtherJoinSideNodeName(((KTableImpl) other).name)
+                             .withThisJoinSideNodeName(name);
 
         final KTableKTableJoinNode kTableKTableJoinNode = kTableJoinNodeBuilder.build();
         builder.addGraphNode(this.streamsGraphNode, kTableKTableJoinNode);
@@ -526,10 +584,10 @@ public <K1, V1> KGroupedTable<K1, V1> groupBy(final KeyValueMapper<? super K, ?
         final String selectName = builder.newProcessorName(SELECT_NAME);
 
         final KTableProcessorSupplier<K, V, KeyValue<K1, V1>> selectSupplier = new KTableRepartitionMap<>(this, selector);
-        final ProcessorParameters processorParameters = new ProcessorParameters<>(selectSupplier, selectName);
+        final ProcessorParameters<K, Change<V>> processorParameters = new ProcessorParameters<>(selectSupplier, selectName);
 
         // select the aggregate key and values (old and new), it would require parent to send old values
-        final ProcessorGraphNode<K1, V1> groupByMapNode = new ProcessorGraphNode<>(
+        final ProcessorGraphNode<K, Change<V>> groupByMapNode = new ProcessorGraphNode<>(
             selectName,
             processorParameters,
             false