AbstractStateManager — Base State Manager

AbstractStateManager is the base implementation of the StateManager contract for state managers that use offset checkpointing.

Note	`AbstractStateManager` is a Java abstract class and cannot be created directly. It is created indirectly for the concrete AbstractStateManagers.

AbstractStateManager uses .checkpoint for the name of the checkpoint offset file.

Table 1. AbstractStateManagers
AbstractStateManager	Description
GlobalStateManagerImpl
ProcessorStateManager

Creating AbstractStateManager Instance

AbstractStateManager takes the following to be created:

Base directory
eosEnabled flag

AbstractStateManager initializes the internal properties.

Note	`AbstractStateManager` is a Java abstract class and cannot be created directly. It is created indirectly for the concrete AbstractStateManagers.

`reinitializeStateStoresForPartitions` Method

void reinitializeStateStoresForPartitions(
  Logger log,
  Map<String, StateStore> stateStores,
  Map<String, String> storeToChangelogTopic,
  Collection<TopicPartition> partitions,
  InternalProcessorContext processorContext)

reinitializeStateStoresForPartitions…FIXME

Note	`reinitializeStateStoresForPartitions` is used when: `GlobalStateManagerImpl` is requested to reinitializeStateStoresForPartitions `ProcessorStateManager` is requested to reinitializeStateStoresForPartitions

`inverseOneToOneMap` Internal Method

Map<String, String> inverseOneToOneMap(Map<String, String> origin)

inverseOneToOneMap…FIXME

Note	`inverseOneToOneMap` is used exclusively when `AbstractStateManager` is requested to reinitializeStateStoresForPartitions.

`converterForStore` Static Method

RecordConverter converterForStore(StateStore store)

converterForStore…FIXME

Note	`converterForStore` is used when…FIXME

Internal Properties

Name Description

Name	Description
`checkpoint`	OffsetCheckpoint with the .checkpoint file in the base directory. Initialized to a new OffsetCheckpoint when `AbstractStateManager` is created Used when: `AbstractStateManager` is requested to reinitializeStateStoresForPartitions (to write the checkpointableOffsets with exactly-once support off) `GlobalStateManagerImpl` is requested to initialize and checkpoint `ProcessorStateManager` is created and requested to checkpoint
`checkpointableOffsets`	Collection of Kafka TopicPartitions with offsets (`Map<TopicPartition, Long>`) Entries (partitions with offsets) added when: `GlobalStateManagerImpl` is requested to initialize, restoreState and checkpoint `ProcessorStateManager` is created and requested to checkpoint
`globalStores`	Global state stores by name (managed by GlobalStateManagerImpl mostly) New state stores added when: `GlobalStateManagerImpl` is requested to register a state store `ProcessorStateManager` is requested to registerGlobalStateStores State stores removed when: `GlobalStateManagerImpl` is requested to close
`stores`	State stores by name (managed by ProcessorStateManager) New state stores added when: `ProcessorStateManager` is requested to register a state store State stores removed when: `ProcessorStateManager` is requested to close

checkpoint

OffsetCheckpoint with the .checkpoint file in the base directory.

Initialized to a new OffsetCheckpoint when AbstractStateManager is created

Used when:

AbstractStateManager is requested to reinitializeStateStoresForPartitions (to write the checkpointableOffsets with exactly-once support off)
GlobalStateManagerImpl is requested to initialize and checkpoint
ProcessorStateManager is created and requested to checkpoint

checkpointableOffsets

Collection of Kafka TopicPartitions with offsets (Map<TopicPartition, Long>)

Entries (partitions with offsets) added when:

GlobalStateManagerImpl is requested to initialize, restoreState and checkpoint
ProcessorStateManager is created and requested to checkpoint

globalStores

Global state stores by name (managed by GlobalStateManagerImpl mostly)

New state stores added when:

GlobalStateManagerImpl is requested to register a state store
ProcessorStateManager is requested to registerGlobalStateStores

State stores removed when:

GlobalStateManagerImpl is requested to close

stores

State stores by name (managed by ProcessorStateManager)

New state stores added when:

ProcessorStateManager is requested to register a state store

State stores removed when:

ProcessorStateManager is requested to close

AbstractStateManager