Status

Current state: Under Discussion Accepted

Discussion thread: here 

JIRA:

...

public interface DslStorePlugin {

    KeyValueBytesStoreSupplier keyValueStore(final KeyValuePluginParams params);

    WindowBytesStoreSupplier windowStore(final WindowPluginParams params);

    SessionBytesStoreSupplier sessionStore(final SessionPluginParams params);
}

// the below are all "struct"-like classes with the following fields
class KeyValuePluginParams(String name);
class WindowPluginParams(String name, Duration retentionPeriod, Duration windowSize, boolean retainDuplicates, EmitStrategy emitStrategy);
class SessionPluginParams(String name, Duration retentionPeriod, EmitStrategy emitStrategy);

OOTB Store Type Specs

We will provide a default implementations of the above interface in a class org.apache.kafka.streams.DefaultDslStorePlugin. This default implementation will use either RocksDB or In Memory stores depending on the default.dsl.store  configuration (see rejected alternatives for a discussion here).

Configuration

A user can optionally specify a DslStorePlugin by means of a new config, defined below, which defaults to org.apache.kafka.streams.DefaultDslStorePlugin.

public static final String DSL_STORE_PLUGIN_CLASS_CONFIG = "dsl.store.plugin.class";
public static final String DSL_STORE_PLUGIN_CLASS_DOC = "Defines which store implementations to plug in to DSL operators. Must implement the <code>org.apache.kafka.streams.state.DslStorePlugin</code> interface.";

Example usage of this configuration:

dsl.store.plugin.class = com.my.company.MyCustomStoreTypeSpec

If using anything other than dsl.store.plugin.class = org.apahce.kafka.streams.DefaultDslStorePlugin  the default.dsl.store  configuration will be ignored.

Materialized API

In the existing code, users are able to override the default store type by leveraging the existing Materialized.as and Materialized.withStoreType  methods. We will change the signature of these methods (see below on how we maintain compatibility) to take in a DslStorePlugin  instead of the StoreType  enum:

public class Materialized {

    public static <K, V, S extends StateStore> Materialized<K, V, S> as(final DslStorePlugin storePlugin);

    public Materialized<K, V, S> withStoreType(final DslStorePlugin storePlugin);
}

In order to ensure code compatibility and smooth upgrade paths, we will leave the old enums in place and have them implement DslStorePlugin:

package org.apache.kafka.streams.kstream;

public class Materialized {
	...
	public enum StoreType implements DslStorePlugin {
		// DefaultDslStorePlugin will itself be a delegate to two different implementations
		// which will be stored in static fields ROCKS_DB and IN_MEMORY, choosing which one
		// to delegate to depending on the default.dsl.store configuration
    	ROCKS_DB(DefaultDslStorePlugin.ROCKS_DB),
    	IN_MEMORY(DefaultDslStorePlugin.IN_MEMORY);
	
		private final DslStorePlugin delegatePlugin;

		StoreTypeSpec(final DslStorePlugin delegate) {
			this.delegatePlugin = delegate;
		}

		// delegate all methods to delegatePlugin
	}
	...
}

StreamJoined API

Finally, to address the three problems with configuring state stores for stream-stream joins today, we propose one last new method that will operate similarly to the new Materialized APIs but for StreamJoined:

/**
 * {@code DslStoreSuppliers} defines a grouping of factories to construct
 * stores for each of the types of state store implementations in Kafka
 * Streams. This allows configuration of a default store supplier beyond
 * the builtin defaults of RocksDB and In-Memory.
 *
 * <p>There are various ways that this configuration can be supplied to
 * the application (in order of precedence):
 * <ol>
 *     <li>Passed in directly to a DSL operator via either
 *     {@link org.apache.kafka.streams.kstream.Materialized#as(DslStoreSuppliers)},
 *     {@link org.apache.kafka.streams.kstream.Materialized#withStoreType(DslStoreSuppliers)}, or
 *     {@link org.apache.kafka.streams.kstream.StreamJoined#withDslStoreSuppliers(DslStoreSuppliers)}</li>
 *
 *     <li>Passed in via a Topology configuration override (configured in a
 *     {@link org.apache.kafka.streams.TopologyConfig} and passed into the
 *     {@link org.apache.kafka.streams.StreamsBuilder#StreamsBuilder(TopologyConfig)} constructor</li>
 *
 *     <li>Configured as a global default in {@link org.apache.kafka.streams.StreamsConfig} using
 *     the {@link org.apache.kafka.streams.StreamsConfig#DSL_STORE_SUPPLIERS_CLASS_CONFIG}</li>
 *     configuration.
 * </ol>
 *
 * <p>Kafka Streams is packaged with some pre-existing {@code DslStoreSuppliers}
 * that exist in {@link BuiltInDslStoreSuppliers}
 */
public interface DslStoreSuppliers extends Configurable {

    KeyValueBytesStoreSupplier keyValueStore(final DslKeyValueParams params);

    WindowBytesStoreSupplier windowStore(final DslWindowParams params);

    SessionBytesStoreSupplier sessionStore(final DslSessionParams params);
}

// the below are all "struct"-like classes with the following fields
class DslKeyValueParams(String name);
class DslWindowParams(String name, Duration retentionPeriod, Duration windowSize, boolean retainDuplicates, EmitStrategy emitStrategy, boolean isSlidingWindow, boolean isTimestamped);
class DslSessionParams(String name, Duration retentionPeriod, EmitStrategy emitStrategy);

OOTB Store Type Specs

We will provide a default implementations of the above interfaces:

• org.apache.kafka.streams.RocksDbDslStoreSuppliers
• org.apache.kafka.streams.InMemoryDslStoreSuppliers

Configuration

These interfaces will be specified by means of a new config, defined below. This takes precedence over the old default.dsl.store config, which will be deprecated to provide clear signal on the preferred method. Further discussion on this point is under "Rejected Alternatives":

public static final String DSL_STORE_SUPPLIERS_CLASS_CONFIG = "dsl.store.suppliers.class";
public static final String DSL_STORE_SUPPLIERS_CLASS_DOC = "Defines which store implementations to plug in to DSL operators. Must implement the <code>org.apache.kafka.streams.state.DslStoreSuppliers</code> interface.";
public static final Class DSL_STORE_SUPPLIERS_CLASS_DEFAULT = org.apache.kafka.streams.RocksDbDslStoreSuppliers.class;

// we also add this constant for the old configuration so that it is easily referenceable
// across the code base
public static final String DEFAULT_DSL_STORE = ROCKS_DB;

Example usage of this configuration:

dsl.store.suppliers.class = com.my.company.MyCustomStoreSuppliers

TopologyConfig

We will expose the following method from TopologyConfig :

    /**
     * @return the DslStoreSuppliers if the value was explicitly configured (either by
     *         {@link StreamsConfig#DEFAULT_DSL_STORE} or {@link StreamsConfig#DSL_STORE_SUPPLIERS_CLASS_CONFIG})
     */
    public Optional<DslStoreSuppliers> resolveDslStoreSuppliers() {
        if (isTopologyOverride(DSL_STORE_SUPPLIERS_CLASS_CONFIG, topologyOverrides) || globalAppConfigs.originals().containsKey(DSL_STORE_SUPPLIERS_CLASS_CONFIG)) {

public class StreamJoined {

    /**
     * Creates a StreamJoined instance with the given {@link DslStorePlugin}. The store plugin
     * will be used to get all the state stores in this operation that do not otherwise have an
     * explicitly configured {@link org.apache.kafka.streams.state.DslStorePlugin}.
     *
     * @param storePlugin         the store plugin that will be used for state stores
     * @param <K>                 the key type
     * @param <V1>                this value type
     * @param <V2>return Optional.of(Utils.newInstance(dslStoreSuppliers, DslStoreSuppliers.class));
        } else if (isTopologyOverride(DEFAULT_DSL_STORE_CONFIG, topologyOverrides) || globalAppConfigs.originals().containsKey(DEFAULT_DSL_STORE_CONFIG)) {
     other value type
     * @returnreturn Optional.of(MaterializedInternal.parse(storeType));
        } else {
          {@link StreamJoined} instance
     */return Optional.empty();
    public static <K, V1, V2> StreamJoined<K, V1, V2> with(final DslStorePlugin storePlugin);
    
    /**
     * Configure with the provided {@link DslStorePlugin} for all state stores that are not
     * configured with a {@link org.apache.kafka.streams.state.DslStorePlugin} already.
     * 
     * @param storePlugin  the store plugin to use for plugging in state stores
     * @return             a new {@link StreamJoined} configured with this store plugin
     */
    public StreamJoined<K, V1, V2> withStorePlugin(final DslStorePlugin storePlugin);
}  

Proposed Changes

There will be no functional changes as part of this KIP. Implementations for all of the above interfaces will be provided. Note the following discussion points:

• Terminology: "Store Type" vs "Store Implementation". A store "type" is a top level DSL store type - there are currently only three: KV, Windowed and Session. A "store implementation" is the chosen implementation of the specified store type. Kafka Streams provides RocksDB and in-memory store implementations OOTB.
• Versioned Tables, like timestamped tables, are an implementation detail of the chosen plugin when providing a store type (usually KV stores). In the future, the DefaultDslStorePlugin is free to switch its default implementation of the KV store to return a versioned implementation instead. This KIP does not propose promoting Versioned tables to the top level API here.

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?

N/A

• If we are changing behavior how will we phase out the older behavior?

N/A

• If we need special migration tools, describe them here.

N/A

• When will we remove the existing behavior?

N/A

Test Plan

Existing test coverage for the default.dsl.store configuration will be migrated to use the new configuration and ensure that the default behavior can be changed.

Rejected Alternatives

Deprecate Existing Config

We can deprecate the existing config entirely, and instead provide two out of the box implementations of dsl.store.plugin.class:

• RocksDbStorePlugin
• InMemoryStorePlugin

This makes it a little easier for users to reason about as there is only a single config. It also would make it a bit easier to aadd new store implementations going forward without cluttering the DefaultStorePlugin class. The downside is that it is more cumbersome to configure (an enum is easy to remember and clear in the configs - as opposed to a class). The other downside is that there will be a deprecation period where both are active and if users ignore those deprecations they will have compile time errors when we remove the original config.

Add CUSTOM Enum to Existing Config

 }

This allows us to determine whether or not a DslStoreSuppliers was configured or not, and also respects the hierarchy of resolution described in the javadocs to DslStoreSuppliers.

Materialized API

In the existing code, users are able to override the default store type by leveraging the existing Materialized.as and Materialized.withStoreType  methods. We will change the signature of these methods (see below on how we maintain compatibility) to take in a DslStoreSuppliers  instead of the StoreType  enum:

public class Materialized {

    public static <K, V, S extends StateStore> Materialized<K, V, S> as(final DslStoreSuppliers storeSuppliers);

    public Materialized<K, V, S> withStoreType(final DslStoreSuppliers storeSuppliers);
}

In order to ensure code compatibility and smooth upgrade paths, we will leave the old enums in place and have them implement DslStoreSuppliers:

package org.apache.kafka.streams.kstream;

public class Materialized {
	...
	public enum StoreType implements DslStoreSuppliers {
		// DefaultDslStoreSuppliers will itself be a delegate to two different implementations
		// which will be stored in static fields ROCKS_DB and IN_MEMORY, choosing which one
		// to delegate to depending on the default.dsl.store configuration
    	ROCKS_DB(DefaultDslStoreSuppliers.ROCKS_DB),
    	IN_MEMORY(DefaultDslStoreSuppliers.IN_MEMORY);
	
		private final DslStoreSuppliers delegate;

		StoreTypeSpec(final DslStoreSuppliers delegate) {
			this.delegate = delegate;
		}

		// delegate all methods to delegate
	}
	...
}

StreamJoined API

Finally, to address the three problems with configuring state stores for stream-stream joins today, we propose one last new method that will operate similarly to the new Materialized APIs but for StreamJoined:

public class StreamJoined {

    /**
     * Creates a StreamJoined instance with the given {@link DslStoreSuppliers}. The store plugin
     * will be used to get all the state stores in this operation that do not otherwise have an
     * explicitly configured {@link org.apache.kafka.streams.state.DslStoreSuppliers}.
     *
     * @param storeSuppliers         the store plugin that will be used for state stores
     * @param <K>                 the key type
     * @param <V1>                this value type
     * @param <V2>                other value type
     * @return                    {@link StreamJoined} instance
     */
    public static <K, V1, V2> StreamJoined<K, V1, V2> with(final DslStoreSuppliers storeSuppliers);
    
    /**
     * Configure with the provided {@link DslStoreSuppliers} for all state stores that are not
     * configured with a {@link org.apache.kafka.streams.state.DslStoreSuppliers} already.
     * 
     * @param storeSuppliers  the store plugin to use for plugging in state stores
     * @return             a new {@link StreamJoined} configured with this store plugin
     */
    public StreamJoined<K, V1, V2> withstoreSuppliers(final DslStoreSuppliers storeSuppliers);
}  

Proposed Changes

There will be no functional changes as part of this KIP. Implementations for all of the above interfaces will be provided. Note the following discussion points:

• Terminology: "Store Type" vs "Store Implementation". A store "type" is a top level DSL store type - there are currently only three: KV, Windowed and Session. A "store implementation" is the chosen implementation of the specified store type. Kafka Streams provides RocksDB and in-memory store implementations OOTB.
• Versioned Tables, like timestamped tables, are an implementation detail of the chosen plugin when providing a store type (usually KV stores). In the future, the DefaultDslStoreSuppliers is free to switch its default implementation of the KV store to return a versioned implementation instead. This KIP does not propose promoting Versioned tables to the top level API here.

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?

Existing users will see deprecation warnings if they are using the old default.dsl.store configuration in their code. When 4.0 is released this configuration will no longer be respected.

• If we are changing behavior how will we phase out the older behavior?

N/A

• If we need special migration tools, describe them here.

N/A

• When will we remove the existing behavior?

4.0

Test Plan

Existing test coverage for the default.dsl.store configuration will be migrated to use the new configuration and ensure that the default behavior can be changed.

Rejected Alternatives

Support Both Configs

Don't deprecate the old default.dsl.store config and instead maintain it alongside the new config. The advantage here is obviously that we minimize disruption to anyone using this config already. However I believe the existing userbase is likely to be quite small, as in its current form this feature is only useful to those who (a) have many state stores to the point overriding them separately is cumbersome, and (b) also wishes to override all (or most) of their stores to be in-memory specifically. This seems to be a relatively uncommon pattern already, not least of which being that it's a recipe for OOM disaster unless their applications are extremely small and/or low traffic. In the absence of any evidence that this config is already in use despite such a niche application, it seems best to take this opportunity to deprecate it now, before it gains any more traction, and have everyone converge on the same, single configuration going forward. Whenever there are two configs affecting the same underlying feature, it's inevitable there will be confusion over their interaction/hierarchy which can often lead to misconfiguration. Deprecating the old config is a clear sign to users which of the two should be preferred and will take precedence.

Add CUSTOM Enum to Existing Config

Instead of positioning the new config as a full replacement for default.dsl.store, we could instead introduce it as a complementary config for custom stores specifically. For example, add a 3rd StoreType to the enum definition, such as CUSTOM. If (and only if) the CUSTOM StoreType is used, the value of the new default.dsl.store.type.spec config will be looked at and used to obtain the actual StoreSupplier instances of this custom type. This option has the same advantages as the above, though I would qualify it in the same way. In general, having two configs that do the same thing or are tied to the same feature will be confusing to new (or old!) users. However, between these two alternatives I would personally advocate for this one as the better option, as it at least solves the concern over potential misconfiguration due to unclear interaction between the two configsIn the proposal above, the dsl.store.plugin.class is the first to be resolved, and then only if DefaultDslStorePlugin is used the old default.dsl.store config is used. We could flip the semantics, and require users to specify CUSTOM as the default.dsl.store and only then check the new configuration. This option is slightly more cumbersome over the proposed design, and doesn't have any clear advantages.

Deprecating StoreType Enum, Materialized.as(StoreType) and Materialized.withStoreType(StoreType) APIs

...

public class Materialized {

	// new methods
    public static <K, V, S extends StateStore> Materialized<K, V, S> as(final DslStorePluginDslStoreSuppliers storePluginstoreSuppliers);
    public Materialized<K, V, S> withStorePluginwithStoreSuppliers(final DslStorePluginDslStoreSuppliers storePluginstorestoreSuppliers);

	// deprecate old methods
	@Deprecated
    public static <K, V, S extends StateStore> Materialized<K, V, S> as(final StoreType storeType);
	@Deprecated
    public Materialized<K, V, S> withStoreType(final StoreType storeType); 
}

...