Status

Current state: Discarded (covered by KIP-182)

Discussion thread: here

JIRA: TBD

PR: here

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

 

Note: This KIP may become obsolete as the discussion on "Streams DSL/StateStore Refactoring" supersedes it.

Motivation

The recommended practice to create a StateStoreSupplier is as per the following example:

StateStoreSupplier countStore = Stores.create("Counts")
    .withKeys(Serdes.String())
    .withValues(Serdes.Long())
    .persistent()
    .build();

However, StateStoreSupplier is a generic interface that takes the StateStore type as a parameter:

public interface StateStoreSupplier<T extends StateStore>

In the above example that type parameter is lost as the build() method returns a raw type.

As StateStoreSupplier is passed to count/reduce/aggregate etc. methods on KGroupedStream or KGroupedTable, the compiler cannot detect if a supplier for the wrong kind of store is provided.

The other parameters to those methods, such as Serdes, Reducers, etc are type-parameterised by the key and value types allowing compile-time type checks.

The StateStoreSupplier argument stands out as a raw type. Making it type-parameterised will help detect at compile time errors such as when someone refactors their app to use a different typo of aggregations (e.g. TimeWindowed vs SessionWindowed) and forgets to change the StateStoreSupplier passed in.

Public Interfaces

  • KGroupedStream - deprecate the following methods:

    • KTable<K, Long> count(final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <W extends Window> KTable<Windowed<K>, Long> count(final Windows<W> windows,
                                                                                                     final StateStoreSupplier<WindowStore> storeSupplier);

    • KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows,
                                                                  final StateStoreSupplier<SessionStore> storeSupplier);

    • KTable<K, V> reduce(final Reducer<V> reducer,
                                         final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <W extends Window> KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                                                                  final Windows<W> windows,
                                                                                                  final StateStoreSupplier<WindowStore> storeSupplier);

    • KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                              final SessionWindows sessionWindows,

                                                              final StateStoreSupplier<SessionStore> storeSupplier);

    • <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                                                           final Aggregator<? super K, ? super V, VR> aggregator,

                                                           final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <W extends Window, VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                                                                                                 final Aggregator<? super K, ? super V, VR> aggregator,
                                                                                                                 final Windows<W> windows,
                                                                                                                 final StateStoreSupplier<WindowStore> storeSupplier);

    • <T> KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,

                                                                          final Aggregator<? super K, ? super V, T> aggregator,

                                                                          final Merger<? super K, T> sessionMerger,

                                                                          final SessionWindows sessionWindows,

                                                                          final Serde<T> aggValueSerde,

                                                                          final StateStoreSupplier<SessionStore> storeSupplier);

  • KGroupedStream - add the following replacement methods:

    • KTable<K, Long> count(final TypedStateStoreSupplier<KeyValueStore<K, Long>> storeSupplier);

    • <W extends Window> KTable<Windowed<K>, Long> count(final Windows<W> windows,
                                                                                                     final TypedStateStoreSupplier<WindowStore<K, Long>> storeSupplier);

    • KTable<Windowed<K>, Long> count(final SessionWindows sessionWindows,
                                                                  final TypedStateStoreSupplier<SessionStore<K, Long>> storeSupplier);

    • KTable<K, V> reduce(final Reducer<V> reducer,
                                         final TypedStateStoreSupplier<KeyValueStore<K ,V>> storeSupplier);

    • <W extends Window> KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                                                                  final Windows<W> windows,
                                                                                                  final TypedStateStoreSupplier<WindowStore<K, V>> storeSupplier);

    • KTable<Windowed<K>, V> reduce(final Reducer<V> reducer,
                                                              final SessionWindows sessionWindows,

                                                              final TypedStateStoreSupplier<SessionStore<K, V>> storeSupplier);

    • <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                                                           final Aggregator<? super K, ? super V, VR> aggregator,

                                                           final TypedStateStoreSupplier<KeyValueStore<K, VR>> storeSupplier);

    • <W extends Window, VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                                                                                                 final Aggregator<? super K, ? super V, VR> aggregator,
                                                                                                                 final Windows<W> windows,
                                                                                                                 final TypedStateStoreSupplier<WindowStore<K, VR>> storeSupplier);

    • <T> KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer,

                                                                          final Aggregator<? super K, ? super V, T> aggregator,

                                                                          final Merger<? super K, T> sessionMerger,

                                                                          final SessionWindows sessionWindows,

                                                                          final Serde<T> aggValueSerde,

                                                                          final TypedStateStoreSupplier<SessionStore<K, T>> storeSupplier);

  • KGroupedTable - deprecate the following methods:

    • KTable<K, Long> count(final StateStoreSupplier<KeyValueStore> storeSupplier);

    • KTable<K, V> reduce(final Reducer<V> adder,
                                         final Reducer<V> subtractor,
                                         final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                                                           final Aggregator<? super K, ? super V, VR> adder,
                                                           final Aggregator<? super K, ? super V, VR> subtractor,
                                                           final StateStoreSupplier<KeyValueStore> storeSupplier);

  • KGroupedTable - add the following replacement methods:

    • KTable<K, Long> count(final TypedStateStoreSupplier<KeyValueStore<K, Long>> storeSupplier);

    • KTable<K, V> reduce(final Reducer<V> adder,
                                         final Reducer<V> subtractor,
                                         final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);

    • <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                                                           final Aggregator<? super K, ? super V, VR> adder,
                                                           final Aggregator<? super K, ? super V, VR> subtractor,
                                                           final TypedStateStoreSupplier<KeyValueStore<K, VR>> storeSupplier);

  • KTable - deprecate the following methods:
    • KTable<K, V> filter(final Predicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier);
    • KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <VR> KTable<K, VR> mapValues(final ValueMapper<? super V, ? extends VR> mapper,
                                                             final Serde<VR> valueSerde,
                                                             final StateStoreSupplier<KeyValueStore> storeSupplier);

    • KTable<K, V> through(final String topic,
                                          final StateStoreSupplier<KeyValueStore> storeSupplier);

    • KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                          final String topic,
                                          final StateStoreSupplier<KeyValueStore> storeSupplier);

    • KTable<K, V> through(final Serde<K> keySerde, Serde<V> valSerde,
                                          final String topic,
                                          final StateStoreSupplier<KeyValueStore> storeSupplier);

    • KTable<K, V> through(final Serde<K> keySerde,
                                          final Serde<V> valSerde,
                                          final StreamPartitioner<? super K, ? super V> partitioner,
                                          final String topic,
                                          final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <VO, VR> KTable<K, VR> join(final KTable<K, VO> other,
                                                        final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                                        final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other,
                                                             final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                                             final StateStoreSupplier<KeyValueStore> storeSupplier);

    • <VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                                                 final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                                                 final StateStoreSupplier<KeyValueStore> storeSupplier);

  • KTable - add the following replacement methods:
    • KTable<K, V> filter(final Predicate<? super K, ? super V> predicate, final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);
    • KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate, final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);

    • <VR> KTable<K, VR> mapValues(final ValueMapper<? super V, ? extends VR> mapper,
                                                             final Serde<VR> valueSerde,
                                                             final TypedStateStoreSupplier<KeyValueStore<K, VR>> storeSupplier);

    • KTable<K, V> through(final String topic,
                                          final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);

    • KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                                          final String topic,
                                          final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);

    • KTable<K, V> through(final Serde<K> keySerde, Serde<V> valSerde,
                                          final String topic,
                                          final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);

    • KTable<K, V> through(final Serde<K> keySerde,
                                          final Serde<V> valSerde,
                                          final StreamPartitioner<? super K, ? super V> partitioner,
                                          final String topic,
                                          final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier);

    • <VO, VR> KTable<K, VR> join(final KTable<K, VO> other,
                                                        final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                                        final TypedStateStoreSupplier<KeyValueStore<K, VR>> storeSupplier);

    • <VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other,
                                                             final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                                             final TypedStateStoreSupplier<KeyValueStore<K, VR>> storeSupplier);

    • <VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                                                 final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
                                                                 final TypedStateStoreSupplier<KeyValueStore<K, VR>> storeSupplier);

  • KStreamBuilder - deprecate the following method:

    • <K, V> GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                                       final Serde<V> valSerde,
                                                                       final String topic,
                                                                       final StateStoreSupplier<KeyValueStore> storeSupplier)

  • KStreamBuilder - add the following replacement method:
    • <K, V> GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                                       final Serde<V> valSerde,
                                                                       final String topic,
                                                                       final TypedStateStoreSupplier<KeyValueStore<K, V>> storeSupplier)
  • Deprecate Stores class
  • Add a new replacement TypedStores class with the following extra public interfaces in addition to those equivalent to those in Stores class:
      • public interface PersistentWindowFactory<K, V> {


/**
* Caching should be enabled on the created store.
* @return the factory to create a persistent window store
*/
PersistentWindowFactory<K, V> enableCaching();

/**
* Indicates that a changelog should not be created for the key-value store
* @return the factory to create a persistent window store
*/
PersistentWindowFactory<K, V> disableLogging();

/**
* Indicates that a changelog should be created for the store. The changelog will be created
* with the provided cleanupPolicy and configs.
*
* Note: Any unrecognized configs will be ignored.
* @param config any configs that should be applied to the changelog
* @return the factory to create a persistent window store
*/
PersistentWindowFactory<K, V> enableLogging(final Map<String, String> config);


/**
* Return the instance of StateStoreSupplier of new window store.
* @return the key-value store; never null
*/
TypedStateStoreSupplier<WindowStore<K, V>> build();
}
      • public interface PersistentSessionFactory<K, V> {
/**
* Indicates that a changelog should be created for the store. The changelog will be created
* with the provided cleanupPolicy and configs.
*
* Note: Any unrecognized configs will be ignored.
* @param config any configs that should be applied to the changelog
* @return the factory to create a persistent key-value store
*/
PersistentSessionFactory<K, V> enableLogging(final Map<String, String> config);
/**
* Indicates that a changelog should not be created for the key-value store
* @return the factory to create a persistent session store
*/
PersistentSessionFactory<K, V> disableLogging();
/**
* Caching should be enabled on the created store.
* @return the factory to create a persistent session store
*/
PersistentSessionFactory<K, V> enableCaching();
/**
* Return the instance of StateStoreSupplier of new session store.
* @return the key-value store; never null
*/
TypedStateStoreSupplier<SessionStore<K, V>> build();
 
}
      • differences in PersistentKeyValueFactory in TypedStores versus that in Stores:

        • PersistentWindowFactory<K, V> windowed(final long windowSize, long retentionPeriod, int numSegments, boolean retainDuplicates);

        • PersistentSessionFactory<K, V> sessionWindowed(final long retentionPeriod);

        • TypedStateStoreSupplier<KeyValueStore<K, V>> build();

      • differences in InMemoryKeyValueFactory in TypedStores versus that in Stores:

        • TypedStateStoreSupplier<KeyValueStore<K, V>> build();

 

Proposed Changes

Pull Request to demonstrate the changes: https://github.com/apache/kafka/pull/2992/files

 

The new usage would be e.g.:

TypedStateStoreSupplier<KeyValueStore<String, Long>> countStore = TypedStores.create("Counts")
    .withKeys(Serdes.String())
    .withValues(Serdes.Long())
    .persistent()
    .build();
TypedStateStoreSupplier<WindowStore<String, Long>> windowedStore = TypedStores.create("WindowedCounts")
    .withKeys(Serdes.String())
    .withValues(Serdes.Long())
    .persistent()
	.windowed(1000, 10000, 10, false)
    .build();
TypedStateStoreSupplier<SessionStore<String, Long>> sessionStore = TypedStores.create("SessionWindowedCounts")
    .withKeys(Serdes.String())
    .withValues(Serdes.Long())
    .persistent()
	.sessionWindowed(60000)
    .build();

 

Compatibility, Deprecation, and Migration Plan

Changes are intended to be backwards-compatible.

Test Plan

Only re-run of existing tests is envisaged at this time.

Rejected Alternatives

Add type parameters to current method parameters. This has been rejected as a backwards-incompatible change.

<K, V>

  • No labels