WindowStoreBuilder

WindowStoreBuilder is a StoreBuilder (indirectly as an AbstractStoreBuilder) that is used to build WindowStores (using a given WindowBytesStoreSupplier).

WindowStoreBuilder takes the following when created:

WindowStoreBuilder can however be created using Stores.windowStoreBuilder.

kafka streams WindowStoreBuilder.png
Figure 1. Creating WindowStoreBuilder using Stores.windowStoreBuilder
import org.apache.kafka.streams.state.Stores
// Using Scala API for Kafka Streams
import org.apache.kafka.streams.scala.Serdes
import java.time.Duration
val storeBuilder = Stores.windowStoreBuilder(
  Stores.persistentWindowStore(
    "queryable-store-name",
    Duration.ofMinutes(1),
    Duration.ofSeconds(30),
    false),
  Serdes.Integer,
  Serdes.Long)
scala> :type storeBuilder
org.apache.kafka.streams.state.StoreBuilder[org.apache.kafka.streams.state.WindowStore[Int,Long]]

import org.apache.kafka.streams.state.internals.WindowStoreBuilder
assert(storeBuilder.isInstanceOf[WindowStoreBuilder[_, _]])

When requested to build a state store, WindowStoreBuilder creates a new MeteredWindowStore (with the state store and the metricsScope from the WindowBytesStoreSupplier).

scala> :type storeBuilder
org.apache.kafka.streams.state.StoreBuilder[org.apache.kafka.streams.state.WindowStore[Int,Long]]

val windowStore = storeBuilder.build
scala> :type windowStore
org.apache.kafka.streams.state.WindowStore[Int,Long]

With caching enabled, build creates a MeteredWindowStore (with a CachingWindowStore).

scala> :type storeBuilder
org.apache.kafka.streams.state.StoreBuilder[org.apache.kafka.streams.state.WindowStore[Int,Long]]

val withCachingWindowStoreBuilder = storeBuilder.withCachingEnabled

val windowStore = withCachingWindowStoreBuilder.build
import org.apache.kafka.streams.state.internals.MeteredWindowStore
assert(windowStore.isInstanceOf[MeteredWindowStore[_, _]])

With logging enabled, build creates a MeteredWindowStore (with a ChangeLoggingWindowBytesStore).

maybeWrapLogging Internal Method

WindowStore<Bytes, byte[]> maybeWrapLogging(
  final WindowStore<Bytes, byte[]> inner)

maybeWrapLogging…​FIXME

Note
 maybeWrapLogging is used when…​FIXME

maybeWrapCaching Internal Method

WindowStore<Bytes, byte[]> maybeWrapCaching(
  final WindowStore<Bytes, byte[]> inner)

maybeWrapCaching…​FIXME

Note
 maybeWrapCaching is used when…​FIXME

retentionPeriod Method

long retentionPeriod()

retentionPeriod…​FIXME

Note
 retentionPeriod is used when…​FIXME

results matching ""

    No results matching ""