kafka-streams-ProcessorStateManager.adoc

ProcessorStateManager

ProcessorStateManager is a concrete AbstractStateManager that…​FIXME

ProcessorStateManager is created exclusively when AbstractTask is created.

ProcessorStateManager uses the -changelog suffix for the name of storeChangelogTopic.

ProcessorStateManager uses…​FIXME…​for the base directory.

checkpoint Method

void checkpoint(final Map<TopicPartition, Long> ackedOffsets)

Note	checkpoint is part of Checkpointable Contract to…​FIXME.

checkpoint…​FIXME

Creating ProcessorStateManager Instance

ProcessorStateManager takes the following when created:

• Task ID

• Sources (as Kafka TopicPartitions)

• isStandby Flag

• StateDirectory

• Map<String, String>

• ChangelogReader

• eosEnabled Flag (that corresponds to processing.guarantee configuration property)

• LogContext

ProcessorStateManager initializes the internal registries and counters.

Registering StateStore (and StateRestoreCallback) — register Method

void register(
  final StateStore store,
  final StateRestoreCallback stateRestoreCallback)

Note	register is part of StateManager Contract to…​FIXME.

register…​FIXME

Closing ProcessorStateManager — close Method

void close(final Map<TopicPartition, Long> ackedOffsets) throws ProcessorStateException

Note	close is part of StateManager Contract to…​FIXME.

close…​FIXME

Flushing ProcessorStateManager — flush Method

void flush()

Note	flush is part of StateManager Contract to…​FIXME.

flush…​FIXME

Requesting Global StateStore By Name — getGlobalStore Method

StateStore getGlobalStore(final String name)

Note	getGlobalStore is part of StateManager Contract to…​FIXME.

getGlobalStore…​FIXME

Requesting StateStore By Name — getStore Method

StateStore getStore(final String name)

Note	getStore is part of StateManager Contract to…​FIXME.

getStore…​FIXME

reinitializeStateStoresForPartitions Method

void reinitializeStateStoresForPartitions(
  final Collection<TopicPartition> partitions,
  final InternalProcessorContext processorContext)

Note	reinitializeStateStoresForPartitions is part of StateManager Contract to…​FIXME.

reinitializeStateStoresForPartitions…​FIXME

storeChangelogTopic Static Method

static String storeChangelogTopic(final String applicationId, final String storeName)

storeChangelogTopic simply returns a [applicationId]-[storeName][STATE_CHANGELOG_TOPIC_SUFFIX].

Note

storeChangelogTopic is used when: InternalTopologyBuilder is requested to buildProcessorNode and topicGroups CachingSessionStore, ChangeLoggingKeyValueBytesStore, ChangeLoggingSessionBytesStore, ChangeLoggingWindowBytesStore, InMemoryKeyValueLoggedStore, InMemoryKeyValueStore, MemoryLRUCache, MeteredKeyValueBytesStore, MeteredSessionStore, MeteredWindowStore, RocksDBSegmentedBytesStore, RocksDBSessionStore, RocksDBWindowStore, CachingKeyValueStore and CachingWindowStore are requested to init StoreChangeLogger is created