...

public interface StoreTypeSpecDslStorePlugin {

    KeyValueBytesStoreSupplier keyValueStore(final KeyValueSupplierParametersKeyValuePluginParams params);

    WindowBytesStoreSupplier windowStore(final WindowSupplierParametersWindowPluginParams params);

    SessionBytesStoreSupplier sessionStore(final SessionSupplierParametersSessionPluginParams params);
}

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

Each StoreTypeSpec  will have Javadoc outlining any guarantees above and beyond the specific guarantees of the generic Store interface. An example is ReadOnlyKeyValueStore, which does not provide read ordering guarantees on range queries, but the RocksDBStoreTypeSpec#KeyValueBytesStoreSupplier would guarantee that keys are read in lexicographical order.

OOTB Store Type Specs

We will provide a default implementations of the above interface in the following classes:

...

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

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":, which defaults to org.apache.kafka.streams.DefaultDslStorePlugin.

public static final String DEFAULT_DSL_STORE_TYPEPLUGIN_SPECCLASS_CONFIG = "default.dsl.store.typeplugin.specclass";
public static final String DEFAULT_DSL_STORE_TYPEPLUGIN_SPECCLASS_DOC = "DefaultDefines statewhich store implementations to specificationplug usedin byto DSL operators. Must implement the <code>org.apache.kafka.streams.state.StoreTypeSpec<DslStorePlugin</code> interface.";

@Deprecated
public static final String DEFAULT_DSL_STORE_CONFIG = "default.dsl.store";

Example usage of this configuration:

// old configuration examples
default.dsl.store = ROCKS_DB
default.dsl.store = IN_MEMORY

// new configuration examples
default.dsl.store.type.spec.plugin.class = orgcom.apachemy.kafka.streams.RocksDbStoreTypeSpec
default.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 StoreTypeSpec DslStorePlugin  instead of the StoreType  enum:

public class Materialized {

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

    public Materialized<K, V, S> withStoreType(final StoreTypeSpecDslStorePlugin typeSpecstorePlugin);
}

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

package org.apache.kafka.streams.kstream;

public class Materialized {
	...
	public enum StoreType implements StoreTypeSpecDslStorePlugin {
		// DefaultDslStorePlugin will itself be a delegate to two different implementations
		ROCKS_DB(new RocksDbStoreTypeSpec())// 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(new InMemoryStoreTypeSpec()DefaultDslStorePlugin.IN_MEMORY);
	
		private final StoreTypeSpecDslStorePlugin delegateSpec;

		StoreTypeSpec(final StoreTypeSpec delegateSpec) {
			this.delegateSpec = delegateSpec;
		}

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

...

public class StreamJoined {

    /**
     * Creates a StreamJoined instance with the given {@link StoreTypeSpecDslStorePlugin}. The store providerplugin
     * 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.StoreSupplierDslStorePlugin}.
     *
     * @param storePlugin  storeTypeSpec       the store type specificationplugin that will be used for all unconfigured 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 StoreTypeSpecDslStorePlugin storeTypeSpecstorePlugin);
    
    /**
     * Configure with the provided {@link StoreTypeSpecDslStorePlugin} for all state stores that are not
     * configured with a {@link org.apache.kafka.streams.state.StoreSupplierDslStorePlugin} already.
     * 
     * @param storeProviderstorePlugin  the store providerplugin to use for allplugging unconfiguredin state stores
     * @return             a new {@link StreamJoined} configured with this store providerplugin
     */
    public StreamJoined<K, V1, V2> withStoreTypeSpec(final StoreTypeSpecDslStorePlugin storeProviderstorePlugin);
}  

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

...

The main alternative to consider here is how to handle the original default.dsl.store config. As noted it's impossible to expand this config as-is without either changing how it's defined or introducing a new config. But there are a few possible approaches with their own merits and flaws:

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 configsis 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

...