Package org.apache.kafka.streams.state

Class Stores

java.lang.Object

org.apache.kafka.streams.state.Stores

@Evolving
public class Stores
extends java.lang.Object

Factory for creating state stores in Kafka Streams.

When using the high-level DSL, i.e., StreamsBuilder, users create StoreSuppliers that can be further customized via Materialized. For example, a topic read as KTable can be materialized into an in-memory store with custom key/value serdes and caching disabled:

 StreamsBuilder builder = new StreamsBuilder();
 KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore("queryable-store-name");
 KTable<Long,String> table = builder.table(
   "topicName",
   Materialized.<Long,String>as(storeSupplier)
               .withKeySerde(Serdes.Long())
               .withValueSerde(Serdes.String())
               .withCachingDisabled());
 

When using the Processor API, i.e., Topology, users create StoreBuilders that can be attached to Processors. For example, you can create a windowed RocksDB store with custom changelog topic configuration like:

 Topology topology = new Topology();
 topology.addProcessor("processorName", ...);

 Map<String,String> topicConfig = new HashMap<>();
 StoreBuilder<WindowStore<Integer, Long>> storeBuilder = Stores
   .windowStoreBuilder(
     Stores.persistentWindowStore("queryable-store-name", ...),
     Serdes.Integer(),
     Serdes.Long())
   .withLoggingEnabled(topicConfig);

 topology.addStateStore(storeBuilder, "processorName");
 

Constructor Summary

Constructors

Constructor	Description
Stores()

Method Summary

Modifier and Type	Method	Description
static KeyValueBytesStoreSupplier	inMemoryKeyValueStore(java.lang.String name)	Create an in-memory KeyValueBytesStoreSupplier.
static <K,V> StoreBuilder<KeyValueStore<K,V>>	keyValueStoreBuilder(KeyValueBytesStoreSupplier supplier, Serde<K> keySerde, Serde<V> valueSerde)	Creates a StoreBuilder than can be used to build a KeyValueStore.
static KeyValueBytesStoreSupplier	lruMap(java.lang.String name, int maxCacheSize)	Create a LRU Map KeyValueBytesStoreSupplier.
static KeyValueBytesStoreSupplier	persistentKeyValueStore(java.lang.String name)	Create a persistent KeyValueBytesStoreSupplier.
static SessionBytesStoreSupplier	persistentSessionStore(java.lang.String name, long retentionPeriod)	Create a persistent SessionBytesStoreSupplier.
static WindowBytesStoreSupplier	persistentWindowStore(java.lang.String name, long retentionPeriod, int numSegments, long windowSize, boolean retainDuplicates)	Create a persistent WindowBytesStoreSupplier.
static <K,V> StoreBuilder<SessionStore<K,V>>	sessionStoreBuilder(SessionBytesStoreSupplier supplier, Serde<K> keySerde, Serde<V> valueSerde)	Creates a StoreBuilder that can be used to build a SessionStore.
static <K,V> StoreBuilder<WindowStore<K,V>>	windowStoreBuilder(WindowBytesStoreSupplier supplier, Serde<K> keySerde, Serde<V> valueSerde)	Creates a StoreBuilder that can be used to build a WindowStore.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

Stores

public Stores()

Method Detail

persistentKeyValueStore

public static KeyValueBytesStoreSupplier persistentKeyValueStore(java.lang.String name)

Create a persistent KeyValueBytesStoreSupplier.

Parameters:

name - name of the store (cannot be null)

Returns:

an instance of a KeyValueBytesStoreSupplier that can be used to build a persistent store

inMemoryKeyValueStore

public static KeyValueBytesStoreSupplier inMemoryKeyValueStore(java.lang.String name)

Create an in-memory KeyValueBytesStoreSupplier.

Parameters:

name - name of the store (cannot be null)

Returns:

an instance of a KeyValueBytesStoreSupplier than can be used to build an in-memory store

lruMap

public static KeyValueBytesStoreSupplier lruMap(java.lang.String name,
                                                int maxCacheSize)

Create a LRU Map KeyValueBytesStoreSupplier.

Parameters:

name - name of the store (cannot be null)

maxCacheSize - maximum number of items in the LRU (cannot be negative)

Returns:

an instance of a KeyValueBytesStoreSupplier that can be used to build an LRU Map based store

persistentWindowStore

public static WindowBytesStoreSupplier persistentWindowStore(java.lang.String name,
                                                             long retentionPeriod,
                                                             int numSegments,
                                                             long windowSize,
                                                             boolean retainDuplicates)

Create a persistent WindowBytesStoreSupplier.

Parameters:

name - name of the store (cannot be null)

retentionPeriod - length of time to retain data in the store (cannot be negative)

numSegments - number of db segments (cannot be zero or negative)

windowSize - size of the windows (cannot be negative)

retainDuplicates - whether or not to retain duplicates.

Returns:

an instance of WindowBytesStoreSupplier

persistentSessionStore

public static SessionBytesStoreSupplier persistentSessionStore(java.lang.String name,
                                                               long retentionPeriod)

Create a persistent SessionBytesStoreSupplier.

Parameters:

name - name of the store (cannot be null)

retentionPeriod - length ot time to retain data in the store (cannot be negative)

Returns:

an instance of a SessionBytesStoreSupplier

windowStoreBuilder

public static <K,V> StoreBuilder<WindowStore<K,V>> windowStoreBuilder(WindowBytesStoreSupplier supplier,
                                                                      Serde<K> keySerde,
                                                                      Serde<V> valueSerde)

Creates a StoreBuilder that can be used to build a WindowStore.

Type Parameters:

K - key type

V - value type

Parameters:

supplier - a WindowBytesStoreSupplier (cannot be null)

keySerde - the key serde to use

valueSerde - the value serde to use; if the serialized bytes is null for put operations, it is treated as delete

Returns:

an instance of StoreBuilder than can build a WindowStore

keyValueStoreBuilder

public static <K,V> StoreBuilder<KeyValueStore<K,V>> keyValueStoreBuilder(KeyValueBytesStoreSupplier supplier,
                                                                          Serde<K> keySerde,
                                                                          Serde<V> valueSerde)

Creates a StoreBuilder than can be used to build a KeyValueStore.

Type Parameters:

K - key type

V - value type

Parameters:

supplier - a KeyValueBytesStoreSupplier (cannot be null)

keySerde - the key serde to use

valueSerde - the value serde to use; if the serialized bytes is null for put operations, it is treated as delete

Returns:

an instance of a StoreBuilder that can build a KeyValueStore

sessionStoreBuilder

public static <K,V> StoreBuilder<SessionStore<K,V>> sessionStoreBuilder(SessionBytesStoreSupplier supplier,
                                                                        Serde<K> keySerde,
                                                                        Serde<V> valueSerde)

Creates a StoreBuilder that can be used to build a SessionStore.

Type Parameters:

K - key type

V - value type

Parameters:

supplier - a SessionBytesStoreSupplier (cannot be null)

keySerde - the key serde to use

valueSerde - the value serde to use; if the serialized bytes is null for put operations, it is treated as delete

Returns:

an instance of StoreBuilder than can build a SessionStore