Status

Current state: Under Discussion Accepted

Discussion thread: https://lists.apache.org/thread.html/600996d83d485f2b8daf45037de64a60cebdfac9b234bf3449b6b753@%3Cdev.kafka.apache.org%3E

JIRA: https://issues.apache.org/jira/browse/

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

...

This allows for the same "reads top to bottom" type of clarity as when using Processors (and Transformers) as when using the high-level DSL.

Public Interfaces

Add an interface StateStoreConnector that interface ConnectedStoreProvider that allows the implementor to specify state stores that should be connected to this processor/transformer (defaulting to no stores).

...

ConnectedStoreProvider

public interface StateStoreConnectorConnectedStoreProvider {
    default List<StoreBuilder>Set<StoreBuilder> stateStoresstores() {
        return Collections.emptyList()null;
    }
}

Change all Processor/TransformerSupplier  interfaces to extend from it:

...

public interface TransformerSupplier<K, V, R> extends StateStoreConnectorConnectedStoreProvider {
    ...
}

ValueTransformerSupplier

public interface ValueTransformerSupplier<V, VR> extends StateStoreConnectorConnectedStoreProvider {
    ...
}

ValueTransformerWithKeySupplier

public interface ValueTransformerWithKeySupplier<K, V, VR> extends StateStoreConnectorConnectedStoreProvider {
    ...
}

ProcessorSupplier

public interface ProcessorSupplier<K, V> extends StateStoreConnectorConnectedStoreProvider {
    ...
}

Proposed Changes

The proposal is to enhance the ProcessorSupplier and TransformerSupplier interfaces by allowing them to provide information about what state stores they "own" when constructing a topology using StreamsBuilder, KStream::process, KStream::transform, and  KStream::transformValues, and Topology::addProcessor.

The public interface changes above directly imply what needs to be changed in KStream:  The process etc methods would get state store names from the list of StoreBuilders that the supplier provides(which implements ConnectedStoreProvider ) provides, rather than the var args stateStoreNames.

Additionally, the The process method would add the StoreBuilders to the topology using builder.addStateStore() and connect the store to that processor, rather than requiring the user to do it themselves.  In order to solve the problem of addStateStore potentially being called twice for the same store (because more than one Supplier specifies it), the check for duplicate stores in addStateStores will be relaxed so to allow for duplicates if the same StoreBuilder instance for the same store namename (compared by referenced, not equals() ).

Compatibility, Deprecation, and Migration Plan

Because the added interface methods are default with a reasonable default, those additions are backwards compatible.  However, given that now there would be two ways

A user may continue to "connect" state stores to a low level processor, we would have to specify how they behave together.  I believe the correct decision is to enforce that for a given call of processor by passing stateStoreNames when calling stream.process/transform(...)  or stream.transform(...) it should only be possible to specify state stores one of the two ways, either through the stateStoreNames  argument or by implementing the stateStores method of StateStoreConnector .  Attempting to do both should cause an exception to be thrown making it clear that the user should choose one way or the other.  This may be used in combination with a Supplier  that provides its own state stores by implementing ConnectedStoreProvider::stores() .

If a StoreBuilder  that was manually added is also returned by a ConnectedStoreProvider , there is no issue since adding a state store will now be idempotent.

No migration tools are required since it's a relatively minor library change.

...