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;
  
  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.SuppressedInternal;
  
  import java.time.Duration;
  import java.util.Collections;
  import java.util.Map;
  
  public interface Suppressed<K> extends NamedOperation<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> {
  
      }
  
      /**
       * Marker interface for a buffer configuration that will strictly enforce size constraints
       * (bytes and/or number of records) on the buffer, so it is suitable for reducing duplicate
       * results downstream, but does not promise to eliminate them entirely.
       */
      interface EagerBufferConfig extends BufferConfig<EagerBufferConfig> {
  
      }
  
      interface BufferConfig<BC extends BufferConfig<BC>> {
          /**
           * Create a size-constrained buffer in terms of the maximum number of keys it will store.
           */
          static EagerBufferConfig maxRecords(final long recordLimit) {
              return new EagerBufferConfigImpl(recordLimit, Long.MAX_VALUE, Collections.emptyMap());
          }
  
          /**
           * 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 EagerBufferConfig maxBytes(final long byteLimit) {
              return new EagerBufferConfigImpl(Long.MAX_VALUE, byteLimit, Collections.emptyMap());
          }
  
          /**
           * 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).
           *
           * <p>As a result, the buffer will consume as much memory as it needs, dictated by the time bound.
           *
           * <p>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.
           *
           * <p>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.
           *
           * <p>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).
           *
           * <p>As a result, the buffer will consume as much memory as it needs, dictated by the time bound.
           *
           * <p>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.
           *
           * <p>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.
           *
           * <p>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
           *
           * <p>This buffer is "strict" in the sense that it will enforce the time bound or shut down.
           * It will never emit early.
           */
          StrictBufferConfig shutDownWhenFull();
  
          /**
           * Set the buffer to just emit the oldest records when any of its constraints are violated.
           *
           * <p>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.
           */
          EagerBufferConfig emitEarlyWhenFull();
  
          /**
           * Disable the changelog for this suppression's internal buffer.
           * This will turn off fault-tolerance for the suppression, and will result in data loss in the event of a rebalance.
           * By default, the changelog is enabled.
           *
           * @return this
           */
          BC withLoggingDisabled();
  
          /**
           * Indicates that a changelog topic should be created containing the currently suppressed
           * records. Due to the short-lived nature of records in this topic it is likely more
           * compactable than changelog topics for KTables.
           *
           * @param config Configs that should be applied to the changelog. Note: Any unrecognized
           *               configs will be ignored.
           * @return this
           */
          BC withLoggingEnabled(final Map<String, String> config);
      }
  
      /**
       * Configure the suppression to emit only the "final results" from the window.
       *
       * <p>By default, all Streams operators emit results whenever new results are available.
       * This includes windowed operations.
       *
       * <p>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.
       *
       * <p>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 out-of-order 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 queryable 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.
       * @return a "final results" mode suppression configuration
       */
      static Suppressed<Windowed<?>> untilWindowCloses(final StrictBufferConfig bufferConfig) {
          return new FinalResultsSuppressionBuilder<>(null, 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 meantime, 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 SuppressedInternal<>(null, timeToWaitForMoreEvents, bufferConfig, null, false);
      }
  
      /**
       * Use the specified name for the suppression node in the topology.
       *
       * <p>This can be used to insert a suppression without changing the rest of the topology names
       * (and therefore not requiring an application reset).
       *
       * <p>Note however, that once a suppression has buffered some records, removing it from the topology would cause
       * the loss of those records.
       *
       * <p>A suppression can be "disabled" with the configuration {@code untilTimeLimit(Duration.ZERO, ...}.
       *
       * @param name The name to be used for the suppression node and changelog topic
       * @return The same configuration with the addition of the given {@code name}.
       */
      @Override
      Suppressed<K> withName(final String name);
  }
  