Package org.apache.kafka.streams.kstream

Class Materialized<K,V,S extends StateStore>

java.lang.Object
org.apache.kafka.streams.kstream.Materialized<K,V,S>
Type Parameters:
K - type of record key
V - type of record value
S - type of state store (note: state stores always have key/value types <Bytes,byte[]>
public class Materialized<K,V,S extends StateStore> extends Object
Used to describe how a StateStore should be materialized. You can either provide a custom StateStore backend through one of the provided methods accepting a supplier or use the default RocksDB backends by providing just a store name.

For example, you can read a topic as KTable and force a state store materialization to access the content via Interactive Queries API: 


 StreamsBuilder builder = new StreamsBuilder();
 KTable<Integer, Integer> table = builder.table(
   "topicName",
   Materialized.as("queryable-store-name"));

Correct Usage When Providing Serde:

To configure both the name of the store and the Serde for the key and value, you should use the following pattern: 


 Materialized.<KeyType, ValueType, StateStore>as("MyStoreName")
     .withKeySerde(keySerde)
     .withValueSerde(valueSerde);
This ensures that the store name is retained while configuring the key and value serde.

Warning: If you use the (static) with(Serde, Serde) method after calling as(String), the instance created by as(String) will be replaced by a new Materialized instance, and any configuration set on the first instance (e.g., store name, logging settings) will be lost.

For example, the following code is incorrect because it discards the configuration of the store name (calling static) methods on an object/instance is an anti-pattern): 


 // This will not work as expected:
 Materialized.<KeyType, ValueType, StateStore>as("MyStoreName")
     .with(keySerde, valueSerde);  // The store name "MyStoreName" is lost
Instead, use the proper pattern of chaining withKeySerde and withValueSerde.
See Also:

  • Method Details

    • as

      public static <K, V, S extends StateStore> Materialized<K,V,S> as(DslStoreSuppliers storeSuppliers)
      Materialize a StateStore with the given DslStoreSuppliers.
      Type Parameters:
      K - key type of the store
      V - value type of the store
      S - type of the StateStore
      Parameters:
      storeSuppliers - the type of the state store
      Returns:
      a new Materialized instance with the given storeName

    • as

      public static <K, V, S extends StateStore> Materialized<K,V,S> as(String storeName)
      Materialize a StateStore with the given name.

      This method sets the name of the state store to be used during materialization. You can provide additional configurations like key and value Serdes using withKeySerde(Serde) and withValueSerde(Serde).

      Warning: Do not use with(Serde, Serde) after calling this method, as it creates a new Materialized instance, which discards the store name and any other configurations set earlier.

      Correct usage: 

      
 Materialized.<KeyType, ValueType, StateStore>as("MyStoreName")
     .withKeySerde(keySerde)
     .withValueSerde(valueSerde);

      Incorrect usage (store name is lost): 

      
 Materialized.<KeyType, ValueType, StateStore>as("MyStoreName")
     .with(keySerde, valueSerde);  // Store name is lost
      Type Parameters:
      K - key type of the store
      V - value type of the store
      S - type of the StateStore
      Parameters:
      storeName - the name of the underlying KTable state store; valid characters are ASCII alphanumerics, '.', '_' and '-'.
      Returns:
      a new Materialized instance with the given storeName

    • as

      public static <K, V> Materialized<K,V,WindowStore<org.apache.kafka.common.utils.Bytes,byte[]>> as(WindowBytesStoreSupplier supplier)
      Materialize a WindowStore using the provided WindowBytesStoreSupplier. Important: Custom subclasses are allowed here, but they should respect the retention contract: Window stores are required to retain windows at least as long as (window size + window grace period). Stores constructed via Stores already satisfy this contract.
      Type Parameters:
      K - key type of the store
      V - value type of the store
      Parameters:
      supplier - the WindowBytesStoreSupplier used to materialize the store
      Returns:
      a new Materialized instance with the given supplier

    • as

      public static <K, V> Materialized<K,V,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> as(SessionBytesStoreSupplier supplier)
      Materialize a SessionStore using the provided SessionBytesStoreSupplier. Important: Custom subclasses are allowed here, but they should respect the retention contract: Session stores are required to retain windows at least as long as (session inactivity gap + session grace period). Stores constructed via Stores already satisfy this contract.
      Type Parameters:
      K - key type of the store
      V - value type of the store
      Parameters:
      supplier - the SessionBytesStoreSupplier used to materialize the store
      Returns:
      a new Materialized instance with the given supplier

    • as

      public static <K, V> Materialized<K,V,KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>> as(KeyValueBytesStoreSupplier supplier)
      Materialize a KeyValueStore using the provided KeyValueBytesStoreSupplier.
      Type Parameters:
      K - key type of the store
      V - value type of the store
      Parameters:
      supplier - the KeyValueBytesStoreSupplier used to materialize the store
      Returns:
      a new Materialized instance with the given supplier

    • with

      public static <K, V, S extends StateStore> Materialized<K,V,S> with(Serde<K> keySerde, Serde<V> valueSerde)
      Materialize a StateStore with the provided key and value Serdes.

      Note: If this method is used after as(String), the original Materialized instance will be replaced with a new instance, and any configuration on the first instance (e.g., store name) will be lost. To configure both a store name and key/value serde, use withKeySerde(Serde) and withValueSerde(Serde) instead.

      Correct usage with Serde: 

      
 Materialized.<KeyType, ValueType, StateStore>as("MyStoreName")
     .withKeySerde(keySerde)
     .withValueSerde(valueSerde);

      Incorrect usage (store name will be lost): 

      
 Materialized.<KeyType, ValueType, StateStore>as("MyStoreName")
     .with(keySerde, valueSerde);  // Store name is lost
      Type Parameters:
      K - key type
      V - value type
      S - store type
      Parameters:
      keySerde - the key Serde to use. If the Serde is null, then the default key serde from configs will be used
      valueSerde - the value Serde to use. If the Serde is null, then the default value serde from configs will be used
      Returns:
      a new Materialized instance with the given key and value serdes

    • withValueSerde

      public Materialized<K,V,S> withValueSerde(Serde<V> valueSerde)
      Set the valueSerde the materialized StateStore will use.
      Parameters:
      valueSerde - the value Serde to use. If the Serde is null, then the default value serde from configs will be used. If the serialized bytes is null for put operations, it is treated as delete operation
      Returns:
      itself

    • withKeySerde

      public Materialized<K,V,S> withKeySerde(Serde<K> keySerde)
      Set the keySerde the materialized StateStore will use.
      Parameters:
      keySerde - the key Serde to use. If the Serde is null, then the default key serde from configs will be used
      Returns:
      itself

    • withLoggingEnabled

      public Materialized<K,V,S> withLoggingEnabled(Map<String,String> config)
      Indicates that a changelog should be created for the store. The changelog will be created with the provided configs.

      Note: Any unrecognized configs will be ignored.

      Parameters:
      config - any configs that should be applied to the changelog
      Returns:
      itself

    • withLoggingDisabled

      public Materialized<K,V,S> withLoggingDisabled()
      Disable change logging for the materialized StateStore.
      Returns:
      itself

    • withCachingEnabled

      public Materialized<K,V,S> withCachingEnabled()
      Enable caching for the materialized StateStore.
      Returns:
      itself

    • withCachingDisabled

      public Materialized<K,V,S> withCachingDisabled()
      Disable caching for the materialized StateStore.
      Returns:
      itself

    • withRetention

      public Materialized<K,V,S> withRetention(Duration retention) throws IllegalArgumentException
      Configure retention period for window and session stores. Ignored for key/value stores. Overridden by pre-configured store suppliers (as(SessionBytesStoreSupplier) or as(WindowBytesStoreSupplier)). 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. If not specified, the retention period would be set as the window length (from window-start through window-end) plus the grace period.
      Parameters:
      retention - the retention time
      Returns:
      itself
      Throws:
      IllegalArgumentException - if retention is negative or can't be represented as long milliseconds

    • withStoreType

      public Materialized<K,V,S> withStoreType(DslStoreSuppliers storeSuppliers) throws IllegalArgumentException
      Set the type of the materialized StateStore.
      Parameters:
      storeSuppliers - the store type Materialized.StoreType to use.
      Returns:
      itself
      Throws:
      IllegalArgumentException - if store supplier is also pre-configured