...
| Code Block |
|---|
public class StreamsConfig {
public static final String DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG = "default.store.impl.class";
private static final String DEFAULT_STORE_IMPLEMENTATION_CLASS_DOC = "Store supplier implementation class to use. Default is set as RocksDBStoreImplementation. It can be overwritten dynamically during streaming operation.";
.define(DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG,
Type.CLASS,
RocksDBStoreImplementation.class.getName(),
Importance.MEDIUM,
DEFAULT_STORE_IMPLEMENTATION_CLASS_DOC)
} |
| Code Block |
|---|
|
/**
* Materialize a {@link StateStore} with the store implementation.
*
* @param storeImplementation store implementation used to materialize the store
* alphanumerics, '.', '_' and '-'.
* @param <K> key type of the store
* @param <V> value type of the store
* @param <S> type of the {@link StateStore}
* @return a new {@link Materialized} instance with the given storeName
*/
public static <K, V, S extends StateStore> Materialized<K, V, S> as(final StoreImplementation storeImplementation) |
| Code Block |
|---|
|
/**
* A state store supplier Implementation interface all types of {@link StateStore}.
*
*/
public interface StoreImplementation {
/**
* Create a {@link KeyValueBytesStoreSupplier}.
*
* @param name name of the store (cannot be {@code null})
* @return an instance of a {@link KeyValueBytesStoreSupplier} that can be used
* to build a persistent key-value store
*/
KeyValueBytesStoreSupplier keyValueSupplier(String name);
/**
* Create a {@link WindowBytesStoreSupplier}.
*
* @param name name of the store (cannot be {@code null})
* @param retentionPeriod length of time to retain data in the store (cannot be negative)
* (note that the retention period must be at least long enough to contain the
* windowed data's entire life cycle, from window-start through window-end,
* and for the entire grace period)
* @param windowSize size of the windows (cannot be negative)
* @param retainDuplicates whether or not to retain duplicates. Turning this on will automatically disable
* caching and means that null values will be ignored.
* @return an instance of {@link WindowBytesStoreSupplier}
* @throws IllegalArgumentException if {@code retentionPeriod} or {@code windowSize} can't be represented as {@code long milliseconds}
* @throws IllegalArgumentException if {@code retentionPeriod} is smaller than {@code windowSize}
WindowBytesStoreSupplier windowBytesStoreSupplier(String name,
Duration retentionPeriod,
Duration windowSize,
boolean retainDuplicates);
/**
* Create a {@link SessionBytesStoreSupplier}.
*
* @param name name of the store (cannot be {@code null})
* @param retentionPeriod length of time to retain data in the store (cannot be negative)
* (note that the retention period must be at least as long enough to
* contain the inactivity gap of the session and the entire grace period.)
* @return an instance of a {@link SessionBytesStoreSupplier}
*/
SessionBytesStoreSupplier sessionBytesStoreSupplier(String name,
Duration retentionPeriod);
} |
Proposed Changes
In order to leverage the new configuration, users will need to call StreamsBuilder#build(Properties) instead of StreamsBuilder#build(). This is an already established pattern that users need to follow to enable topology optimization. As we predict that we might need to access the config in the DSL more often in the future, we also propose to deprecate StreamsBuilder#build() method to lift the mental burden for users to know which of both methods that should use.
...