This Confluence has been LDAP enabled, if you are an ASF Committer, please use your LDAP Credentials to login. Any problems file an INFRA jira ticket please.

Child pages
  • KIP-508: Make Suppression State Queriable
Skip to end of metadata
Go to start of metadata


Current stateUnder Discussion

Discussion thread: thread


Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).


Sometimes, the user may want to query the intermediate state of the suppression. However, as of 2.3, the KTable#suppress operator (added 2.1) lacks this functionality.

Proposed Changes

This KIP proposes to make the suppression buffer state queriable, and add a variant method to specify the queryableStoreName: KTable#suppress(Suppressed, String). User can query the state of suppression buffer like the following:

    // A queryable interface to the suppression buffer, named "suppression-buffer".
    final ReadOnlyKeyValueStore<String, String> myMapStore ="suppression-buffer", QueryableStoreTypes.suppressionBuffer(Serdes.String(), Serdes.String()));

Public Interfaces

Add a new method, KTable#suppress(Suppressed, String)

public interface KTable<K, V> {


     * Suppress some updates from this changelog stream, determined by the supplied {@link Suppressed} configuration.
     * <p>
     * This controls what updates downstream table and stream operations will receive.
     * @param suppressed            Configuration object determining what, if any, updates to suppress
     * @param queryableStoreName    A queryableStoreName of suppression buffer
     * @return A new {@code KTable} with the desired suppression characteristics.
    KTable<K, V> suppress(final Suppressed<? super K> suppressed, final String queryableStoreName);


Add a new QueryableStoreType, SuppressionBufferType

public final class QueryableStoreTypes {


    public static <K, V> QueryableStoreType<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> timestampedKeyValueStore() {
        return new TimestampedKeyValueStoreType<>();


    public static class SuppressionBufferType<K, V> extends QueryableStoreTypeMatcher<ReadOnlyKeyValueStore<K, V>> {

        private final Serde<K> keySerde;
        private final Serde<V> valueSerde;

        SuppressionBufferType(final Serde<K> keySerde, final Serde<V> valueSerde) {
            super(new HashSet<>(Arrays.asList(
            this.keySerde = keySerde;
            this.valueSerde = valueSerde;

        public ReadOnlyKeyValueStore<K, V> create(final StateStoreProvider storeProvider,
                                                  final String storeName) {
            return new CompositeReadOnlyKeyValueStore<>(storeProvider, this, storeName);

        public Serde<K> keySerde() {
            return keySerde;

        public Serde<V> valueSerde() {
            return valueSerde;


Compatibility, Deprecation, and Migration Plan


Rejected Alternatives

Adding KTable#suppress(Suppressed, Materialized)

Since KTable#suppress does not change the Key, Value type of the KTable instance, Materialized#with[KeySerde, ValueSerde] should be ignored. Since the underlying data structure, InMemoryTimeOrderedKeyValueBuffer, does not support caching or logging, Materialized#withCaching[Enabled, Disabled] and Materialized#withLogging[Enabled, Disabled] options are also ignored. For these reasons, a Materialized instance is too big for parameter, unless InMemoryTimeOrderedKeyValueBuffer follows the other StateStore classes.

  • No labels