Package org.apache.flink.runtime.state

Class AbstractKeyedStateBackend<K>

java.lang.Object

org.apache.flink.runtime.state.AbstractKeyedStateBackend<K>

Type Parameters:

K - Type of the key by which state is keyed.

All Implemented Interfaces:

Closeable, AutoCloseable, CheckpointListener, InternalCheckpointListener, CheckpointableKeyedStateBackend<K>, InternalKeyContext<K>, KeyedStateBackend<K>, KeyedStateFactory, PriorityQueueSetFactory, Snapshotable<SnapshotResult<KeyedStateHandle>>, TestableKeyedStateBackend<K>, Disposable

Direct Known Subclasses:

ForStSyncKeyedStateBackend, HeapKeyedStateBackend, RocksDBKeyedStateBackend

public abstract class AbstractKeyedStateBackend<K>
extends Object
implements CheckpointableKeyedStateBackend<K>, InternalCheckpointListener, TestableKeyedStateBackend<K>, InternalKeyContext<K>

Base implementation of KeyedStateBackend. The state can be checkpointed to streams using Snapshotable.snapshot(long, long, CheckpointStreamFactory, CheckpointOptions).

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	AbstractKeyedStateBackend.PartitionStateFactory

Nested classes/interfaces inherited from interface org.apache.flink.runtime.state.KeyedStateBackend

KeyedStateBackend.KeySelectionListener<K>

Field Summary

Fields

Modifier and Type	Field	Description
protected CloseableRegistry	cancelStreamRegistry	Registry for all opened streams, so they can be closed if the task using this backend is closed.
protected InternalKeyContext<K>	keyContext	The key context for this backend.
protected StreamCompressionDecorator	keyGroupCompressionDecorator	Decorates the input and output streams to write key-groups compressed.
protected KeyGroupRange	keyGroupRange	Range of key-groups for which this backend is responsible.
protected TypeSerializer<K>	keySerializer	The key serializer.
protected TaskKvStateRegistry	kvStateRegistry	KvStateRegistry helper for this task.
protected LatencyTrackingStateConfig	latencyTrackingStateConfig
protected int	numberOfKeyGroups	The number of key-groups aka max parallelism.
protected TtlTimeProvider	ttlTimeProvider
protected ClassLoader	userCodeClassLoader

Constructor Summary

Constructors

Modifier	Constructor	Description
	AbstractKeyedStateBackend(TaskKvStateRegistry kvStateRegistry, TypeSerializer<K> keySerializer, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, TtlTimeProvider ttlTimeProvider, LatencyTrackingStateConfig latencyTrackingStateConfig, CloseableRegistry cancelStreamRegistry, InternalKeyContext<K> keyContext)
	AbstractKeyedStateBackend(TaskKvStateRegistry kvStateRegistry, TypeSerializer<K> keySerializer, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, TtlTimeProvider ttlTimeProvider, LatencyTrackingStateConfig latencyTrackingStateConfig, CloseableRegistry cancelStreamRegistry, StreamCompressionDecorator keyGroupCompressionDecorator, InternalKeyContext<K> keyContext)
protected	AbstractKeyedStateBackend(AbstractKeyedStateBackend<K> abstractKeyedStateBackend)

Method Summary

Modifier and Type	Method	Description
<N,S extends State,T> void	applyToAllKeys(N namespace, TypeSerializer<N> namespaceSerializer, StateDescriptor<S,T> stateDescriptor, KeyedStateFunction<K,S> function)	Applies the provided KeyedStateFunction to the state with the provided StateDescriptor of all the currently active keys.
<N,S extends State,T> void	applyToAllKeys(N namespace, TypeSerializer<N> namespaceSerializer, StateDescriptor<S,T> stateDescriptor, KeyedStateFunction<K,S> function, AbstractKeyedStateBackend.PartitionStateFactory partitionStateFactory)
void	close()
boolean	deregisterKeySelectionListener(KeyedStateBackend.KeySelectionListener<K> listener)	Stop calling listener registered in KeyedStateBackend.registerKeySelectionListener(org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K>).
void	dispose()	Closes the state backend, releasing all internal resources, but does not delete any persistent checkpoint data.
K	getCurrentKey()	Used by states to access the current key.
int	getCurrentKeyGroupIndex()	Returns the key-group to which the current key belongs.
InternalKeyContext<K>	getKeyContext()
StreamCompressionDecorator	getKeyGroupCompressionDecorator()
KeyGroupRange	getKeyGroupRange()	Returns the key groups which this state backend is responsible for.
TypeSerializer<K>	getKeySerializer()
LatencyTrackingStateConfig	getLatencyTrackingStateConfig()
int	getNumberOfKeyGroups()	Returns the number of key-groups aka max parallelism.
<N,S extends State,V> S	getOrCreateKeyedState(TypeSerializer<N> namespaceSerializer, StateDescriptor<S,V> stateDescriptor)	Creates or retrieves a keyed state backed by this state backend.
<N,S extends State> S	getPartitionedState(N namespace, TypeSerializer<N> namespaceSerializer, StateDescriptor<S,?> stateDescriptor)	TODO: NOTE: This method does a lot of work caching / retrieving states just to update the namespace.
void	notifyCheckpointSubsumed(long checkpointId)	This method is called as a notification once a distributed checkpoint has been subsumed.
int	numKeyValueStatesByName()
void	publishQueryableStateIfEnabled(StateDescriptor<?,?> stateDescriptor, InternalKvState<?,?,?> kvState)
void	registerKeySelectionListener(KeyedStateBackend.KeySelectionListener<K> listener)	State backend will call KeyedStateBackend.KeySelectionListener.keySelected(K) when key context is switched if supported.
boolean	requiresLegacySynchronousTimerSnapshots(SnapshotType checkpointType)
void	setCurrentKey(K newKey)	Sets the current key that is used for partitioned state.
void	setCurrentKeyAndKeyGroup(K newKey, int newKeyGroupIndex)	Act as a fast path for KeyedStateBackend.setCurrentKey(K) when the key group is known.
void	setCurrentKeyGroupIndex(int currentKeyGroupIndex)	Set current key group index of the context.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.runtime.state.CheckpointableKeyedStateBackend

savepoint

Methods inherited from interface org.apache.flink.api.common.state.CheckpointListener

notifyCheckpointAborted, notifyCheckpointComplete

Methods inherited from interface org.apache.flink.runtime.state.KeyedStateBackend

getKeys, getKeysAndNamespaces, isSafeToReuseKVState

Methods inherited from interface org.apache.flink.runtime.state.KeyedStateFactory

createOrUpdateInternalState, createOrUpdateInternalState, createOrUpdateInternalState

Methods inherited from interface org.apache.flink.runtime.state.PriorityQueueSetFactory

create, create

Methods inherited from interface org.apache.flink.runtime.state.Snapshotable

snapshot

Methods inherited from interface org.apache.flink.runtime.state.TestableKeyedStateBackend

getDelegatedKeyedStateBackend, numKeyValueStateEntries

Field Detail

keySerializer

protected final TypeSerializer<K> keySerializer

The key serializer.

numberOfKeyGroups

protected final int numberOfKeyGroups

The number of key-groups aka max parallelism.

keyGroupRange

protected final KeyGroupRange keyGroupRange

Range of key-groups for which this backend is responsible.

kvStateRegistry

protected final TaskKvStateRegistry kvStateRegistry

KvStateRegistry helper for this task.

cancelStreamRegistry

protected CloseableRegistry cancelStreamRegistry

Registry for all opened streams, so they can be closed if the task using this backend is closed.

userCodeClassLoader

protected final ClassLoader userCodeClassLoader

ttlTimeProvider

protected final TtlTimeProvider ttlTimeProvider

latencyTrackingStateConfig

protected final LatencyTrackingStateConfig latencyTrackingStateConfig

keyGroupCompressionDecorator

protected final StreamCompressionDecorator keyGroupCompressionDecorator

Decorates the input and output streams to write key-groups compressed.

keyContext

protected final InternalKeyContext<K> keyContext

The key context for this backend.

Constructor Detail

AbstractKeyedStateBackend

public AbstractKeyedStateBackend(TaskKvStateRegistry kvStateRegistry,
                                 TypeSerializer<K> keySerializer,
                                 ClassLoader userCodeClassLoader,
                                 ExecutionConfig executionConfig,
                                 TtlTimeProvider ttlTimeProvider,
                                 LatencyTrackingStateConfig latencyTrackingStateConfig,
                                 CloseableRegistry cancelStreamRegistry,
                                 InternalKeyContext<K> keyContext)

AbstractKeyedStateBackend

public AbstractKeyedStateBackend(TaskKvStateRegistry kvStateRegistry,
                                 TypeSerializer<K> keySerializer,
                                 ClassLoader userCodeClassLoader,
                                 ExecutionConfig executionConfig,
                                 TtlTimeProvider ttlTimeProvider,
                                 LatencyTrackingStateConfig latencyTrackingStateConfig,
                                 CloseableRegistry cancelStreamRegistry,
                                 StreamCompressionDecorator keyGroupCompressionDecorator,
                                 InternalKeyContext<K> keyContext)

AbstractKeyedStateBackend

protected AbstractKeyedStateBackend(AbstractKeyedStateBackend<K> abstractKeyedStateBackend)

Method Detail

notifyCheckpointSubsumed

public void notifyCheckpointSubsumed(long checkpointId)
                              throws Exception

Description copied from interface: InternalCheckpointListener

This method is called as a notification once a distributed checkpoint has been subsumed.

These notifications are "best effort", meaning they can sometimes be skipped.

This method is very rarely necessary to implement. The "best effort" guarantee, together with the fact that this method should not result in discarding any data (per the "Checkpoint Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources.

Specified by:

notifyCheckpointSubsumed in interface InternalCheckpointListener

Parameters:

checkpointId - The ID of the checkpoint that has been subsumed.

Throws:

Exception - This method can propagate exceptions, which leads to a failure/recovery for the task or job.

dispose

public void dispose()

Closes the state backend, releasing all internal resources, but does not delete any persistent checkpoint data.

Specified by:

dispose in interface Disposable

Specified by:

dispose in interface KeyedStateBackend<K>

setCurrentKey

public void setCurrentKey(K newKey)

Description copied from interface: KeyedStateBackend

Sets the current key that is used for partitioned state.

Specified by:

setCurrentKey in interface InternalKeyContext<K>

Specified by:

setCurrentKey in interface KeyedStateBackend<K>

Parameters:

newKey - The new current key.

See Also:

KeyedStateBackend

setCurrentKeyAndKeyGroup

public void setCurrentKeyAndKeyGroup(K newKey,
                                     int newKeyGroupIndex)

Description copied from interface: KeyedStateBackend

Act as a fast path for KeyedStateBackend.setCurrentKey(K) when the key group is known.

Specified by:

setCurrentKeyAndKeyGroup in interface KeyedStateBackend<K>

registerKeySelectionListener

public void registerKeySelectionListener(KeyedStateBackend.KeySelectionListener<K> listener)

Description copied from interface: KeyedStateBackend

State backend will call KeyedStateBackend.KeySelectionListener.keySelected(K) when key context is switched if supported.

Specified by:

registerKeySelectionListener in interface KeyedStateBackend<K>

deregisterKeySelectionListener

public boolean deregisterKeySelectionListener(KeyedStateBackend.KeySelectionListener<K> listener)

Description copied from interface: KeyedStateBackend

Stop calling listener registered in KeyedStateBackend.registerKeySelectionListener(org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K>).

Specified by:

deregisterKeySelectionListener in interface KeyedStateBackend<K>

Returns:

returns true iff listener was registered before.

getKeySerializer

public TypeSerializer<K> getKeySerializer()

Specified by:

getKeySerializer in interface KeyedStateBackend<K>

Returns:

Serializer of the key.

See Also:

KeyedStateBackend

getCurrentKey

public K getCurrentKey()

Description copied from interface: InternalKeyContext

Used by states to access the current key.

Specified by:

getCurrentKey in interface InternalKeyContext<K>

Specified by:

getCurrentKey in interface KeyedStateBackend<K>

Returns:

Current key.

See Also:

KeyedStateBackend

getCurrentKeyGroupIndex

public int getCurrentKeyGroupIndex()

Description copied from interface: InternalKeyContext

Returns the key-group to which the current key belongs.

Specified by:

getCurrentKeyGroupIndex in interface InternalKeyContext<K>

See Also:

KeyedStateBackend

getNumberOfKeyGroups

public int getNumberOfKeyGroups()

Description copied from interface: InternalKeyContext

Returns the number of key-groups aka max parallelism.

Specified by:

getNumberOfKeyGroups in interface InternalKeyContext<K>

See Also:

KeyedStateBackend

getKeyGroupRange

public KeyGroupRange getKeyGroupRange()

Description copied from interface: CheckpointableKeyedStateBackend

Returns the key groups which this state backend is responsible for.

Specified by:

getKeyGroupRange in interface CheckpointableKeyedStateBackend<K>

Specified by:

getKeyGroupRange in interface InternalKeyContext<K>

See Also:

KeyedStateBackend

applyToAllKeys

public <N,S extends State,T> void applyToAllKeys(N namespace,
                                                             TypeSerializer<N> namespaceSerializer,
                                                             StateDescriptor<S,T> stateDescriptor,
                                                             KeyedStateFunction<K,S> function)
                                                      throws Exception

Description copied from interface: KeyedStateBackend

Applies the provided KeyedStateFunction to the state with the provided StateDescriptor of all the currently active keys.

Specified by:

applyToAllKeys in interface KeyedStateBackend<K>

Type Parameters:

N - The type of the namespace.

S - The type of the state.

Parameters:

namespace - the namespace of the state.

namespaceSerializer - the serializer for the namespace.

stateDescriptor - the descriptor of the state to which the function is going to be applied.

function - the function to be applied to the keyed state.

Throws:

Exception

See Also:

KeyedStateBackend

applyToAllKeys

public <N,S extends State,T> void applyToAllKeys(N namespace,
                                                             TypeSerializer<N> namespaceSerializer,
                                                             StateDescriptor<S,T> stateDescriptor,
                                                             KeyedStateFunction<K,S> function,
                                                             AbstractKeyedStateBackend.PartitionStateFactory partitionStateFactory)
                                                      throws Exception

Throws:

Exception

getOrCreateKeyedState

public <N,S extends State,V> S getOrCreateKeyedState(TypeSerializer<N> namespaceSerializer,
                                                                 StateDescriptor<S,V> stateDescriptor)
                                                          throws Exception

Description copied from interface: KeyedStateBackend

Creates or retrieves a keyed state backed by this state backend.

Specified by:

getOrCreateKeyedState in interface KeyedStateBackend<K>

Type Parameters:

N - The type of the namespace.

S - The type of the state.

Parameters:

namespaceSerializer - The serializer used for the namespace type of the state

stateDescriptor - The identifier for the state. This contains name and can create a default state value.

Returns:

A new key/value state backed by this backend.

Throws:

Exception - Exceptions may occur during initialization of the state and should be forwarded.

See Also:

KeyedStateBackend

publishQueryableStateIfEnabled

public void publishQueryableStateIfEnabled(StateDescriptor<?,?> stateDescriptor,
                                           InternalKvState<?,?,?> kvState)

getPartitionedState

public <N,S extends State> S getPartitionedState(N namespace,
                                                       TypeSerializer<N> namespaceSerializer,
                                                       StateDescriptor<S,?> stateDescriptor)
                                                throws Exception

TODO: NOTE: This method does a lot of work caching / retrieving states just to update the namespace. This method should be removed for the sake of namespaces being lazily fetched from the keyed state backend, or being set on the state directly.

Specified by:

getPartitionedState in interface KeyedStateBackend<K>

Type Parameters:

N - The type of the namespace.

S - The type of the state.

stateDescriptor - The identifier for the state. This contains name and can create a default state value.

Returns:

A new key/value state backed by this backend.

Throws:

Exception - Exceptions may occur during initialization of the state and should be forwarded.

See Also:

KeyedStateBackend

close

public void close()
           throws IOException

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Throws:

IOException

getLatencyTrackingStateConfig

public LatencyTrackingStateConfig getLatencyTrackingStateConfig()

getKeyGroupCompressionDecorator

@VisibleForTesting
public StreamCompressionDecorator getKeyGroupCompressionDecorator()

numKeyValueStatesByName

@VisibleForTesting
public int numKeyValueStatesByName()

requiresLegacySynchronousTimerSnapshots

public boolean requiresLegacySynchronousTimerSnapshots(SnapshotType checkpointType)

getKeyContext

public InternalKeyContext<K> getKeyContext()

setCurrentKeyGroupIndex

public void setCurrentKeyGroupIndex(int currentKeyGroupIndex)

Description copied from interface: InternalKeyContext

Set current key group index of the context.

Specified by:

setCurrentKeyGroupIndex in interface InternalKeyContext<K>

Parameters:

currentKeyGroupIndex - the current key group index to set to.