Package org.apache.flink.state.changelog

Class ChangelogKeyedStateBackend<K>

  • Type Parameters:
    K - The key by which state is keyed.
    All Implemented Interfaces:
    Closeable, AutoCloseable, org.apache.flink.api.common.state.CheckpointListener, org.apache.flink.api.common.state.InternalCheckpointListener, org.apache.flink.runtime.state.CheckpointableKeyedStateBackend<K>, org.apache.flink.runtime.state.KeyedStateBackend<K>, org.apache.flink.runtime.state.KeyedStateFactory, org.apache.flink.runtime.state.PriorityQueueSetFactory, org.apache.flink.runtime.state.Snapshotable<org.apache.flink.runtime.state.SnapshotResult<org.apache.flink.runtime.state.KeyedStateHandle>>, org.apache.flink.runtime.state.TestableKeyedStateBackend<K>, org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget, org.apache.flink.util.Disposable
    @Internal
public class ChangelogKeyedStateBackend<K>
extends Object
implements org.apache.flink.runtime.state.CheckpointableKeyedStateBackend<K>, org.apache.flink.api.common.state.CheckpointListener, org.apache.flink.runtime.state.TestableKeyedStateBackend<K>, org.apache.flink.api.common.state.InternalCheckpointListener, org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget
    A KeyedStateBackend that keeps state on the underlying delegated keyed state backend as well as on the state change log.

    • Nested Class Summary

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

        org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K extends Object>

    • Field Summary

      • Fields inherited from interface org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget

        NO_OP

    • Constructor Summary

      Constructors 
      Constructor Description
      ChangelogKeyedStateBackend​(org.apache.flink.runtime.state.AbstractKeyedStateBackend<K> keyedStateBackend, String subtaskName, org.apache.flink.api.common.ExecutionConfig executionConfig, org.apache.flink.runtime.state.ttl.TtlTimeProvider ttlTimeProvider, org.apache.flink.metrics.MetricGroup metricGroup, org.apache.flink.runtime.state.changelog.StateChangelogWriter<? extends org.apache.flink.runtime.state.changelog.ChangelogStateHandle> stateChangelogWriter, Collection<org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle> initialState, org.apache.flink.runtime.state.CheckpointStorageWorkerView checkpointStorageWorkerView)  
      ChangelogKeyedStateBackend​(org.apache.flink.runtime.state.AbstractKeyedStateBackend<K> keyedStateBackend, String subtaskName, org.apache.flink.api.common.ExecutionConfig executionConfig, org.apache.flink.runtime.state.ttl.TtlTimeProvider ttlTimeProvider, org.apache.flink.state.changelog.ChangelogStateBackendMetricGroup metricGroup, org.apache.flink.runtime.state.changelog.StateChangelogWriter<? extends org.apache.flink.runtime.state.changelog.ChangelogStateHandle> stateChangelogWriter, Collection<org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle> initialState, org.apache.flink.runtime.state.CheckpointStorageWorkerView checkpointStorageWorkerView, ChangelogStateFactory changelogStateFactory)  

    • Method Summary

      All Methods Instance Methods Concrete Methods 
      Modifier and Type Method Description
      <N,​S extends org.apache.flink.api.common.state.State,​T>
      void      		 applyToAllKeys​(N namespace, org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer, org.apache.flink.api.common.state.StateDescriptor<S,​T> stateDescriptor, org.apache.flink.runtime.state.KeyedStateFunction<K,​S> function)  
      void close()  
      <T extends org.apache.flink.runtime.state.heap.HeapPriorityQueueElement & org.apache.flink.runtime.state.PriorityComparable<? super T> & org.apache.flink.runtime.state.Keyed<?>>
      org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue<T>      		 create​(String stateName, org.apache.flink.api.common.typeutils.TypeSerializer<T> byteOrderedElementSerializer)  
      <N,​SV,​SEV,​S extends org.apache.flink.api.common.state.State,​IS extends S>
      IS      		 createOrUpdateInternalState​(org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer, org.apache.flink.api.common.state.StateDescriptor<S,​SV> stateDesc, org.apache.flink.runtime.state.StateSnapshotTransformer.StateSnapshotTransformFactory<SEV> snapshotTransformFactory)  
      boolean deregisterKeySelectionListener​(org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K> listener)  
      void dispose()  
      ChangelogRestoreTarget<K> getChangelogRestoreTarget()  
      K getCurrentKey()  
      org.apache.flink.runtime.state.KeyedStateBackend<K> getDelegatedKeyedStateBackend​(boolean recursive)  
      org.apache.flink.runtime.state.KeyGroupRange getKeyGroupRange()  
      <N> java.util.stream.Stream<K> getKeys​(String state, N namespace)  
      <N> java.util.stream.Stream<org.apache.flink.api.java.tuple.Tuple2<K,​N>> getKeysAndNamespaces​(String state)  
      org.apache.flink.api.common.typeutils.TypeSerializer<K> getKeySerializer()  
      <N,​S extends org.apache.flink.api.common.state.State,​T>
      S      		 getOrCreateKeyedState​(org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer, org.apache.flink.api.common.state.StateDescriptor<S,​T> stateDescriptor)  
      <N,​S extends org.apache.flink.api.common.state.State>
      S      		 getPartitionedState​(N namespace, org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer, org.apache.flink.api.common.state.StateDescriptor<S,​?> stateDescriptor)  
      void handleMaterializationFailureOrCancellation​(long materializationID, org.apache.flink.runtime.state.changelog.SequenceNumber upTo, Throwable cause)  
      void handleMaterializationResult​(org.apache.flink.runtime.state.SnapshotResult<org.apache.flink.runtime.state.KeyedStateHandle> materializedSnapshot, long materializationID, org.apache.flink.runtime.state.changelog.SequenceNumber upTo)
      This method is not thread safe.
      Optional<org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationRunnable> initMaterialization()
      Initialize state materialization so that materialized data can be persisted durably and included into the checkpoint.
      boolean isSafeToReuseKVState()  
      void notifyCheckpointAborted​(long checkpointId)  
      void notifyCheckpointComplete​(long checkpointId)  
      void notifyCheckpointSubsumed​(long checkpointId)  
      int numKeyValueStateEntries()  
      void registerCloseable​(Closeable closeable)  
      void registerKeySelectionListener​(org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K> listener)  
      org.apache.flink.runtime.state.SavepointResources<K> savepoint()  
      void setCurrentKey​(K newKey)  
      RunnableFuture<org.apache.flink.runtime.state.SnapshotResult<org.apache.flink.runtime.state.KeyedStateHandle>> snapshot​(long checkpointId, long timestamp, org.apache.flink.runtime.state.CheckpointStreamFactory streamFactory, org.apache.flink.runtime.checkpoint.CheckpointOptions checkpointOptions)  

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

        isStateImmutableInStateBackend

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

        createOrUpdateInternalState, createOrUpdateInternalState

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

        create

    • Constructor Detail

      • ChangelogKeyedStateBackend

        public ChangelogKeyedStateBackend​(org.apache.flink.runtime.state.AbstractKeyedStateBackend<K> keyedStateBackend,
                                  String subtaskName,
                                  org.apache.flink.api.common.ExecutionConfig executionConfig,
                                  org.apache.flink.runtime.state.ttl.TtlTimeProvider ttlTimeProvider,
                                  org.apache.flink.metrics.MetricGroup metricGroup,
                                  org.apache.flink.runtime.state.changelog.StateChangelogWriter<? extends org.apache.flink.runtime.state.changelog.ChangelogStateHandle> stateChangelogWriter,
                                  Collection<org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle> initialState,
                                  org.apache.flink.runtime.state.CheckpointStorageWorkerView checkpointStorageWorkerView)

      • ChangelogKeyedStateBackend

        public ChangelogKeyedStateBackend​(org.apache.flink.runtime.state.AbstractKeyedStateBackend<K> keyedStateBackend,
                                  String subtaskName,
                                  org.apache.flink.api.common.ExecutionConfig executionConfig,
                                  org.apache.flink.runtime.state.ttl.TtlTimeProvider ttlTimeProvider,
                                  org.apache.flink.state.changelog.ChangelogStateBackendMetricGroup metricGroup,
                                  org.apache.flink.runtime.state.changelog.StateChangelogWriter<? extends org.apache.flink.runtime.state.changelog.ChangelogStateHandle> stateChangelogWriter,
                                  Collection<org.apache.flink.runtime.state.changelog.ChangelogStateBackendHandle> initialState,
                                  org.apache.flink.runtime.state.CheckpointStorageWorkerView checkpointStorageWorkerView,
                                  ChangelogStateFactory changelogStateFactory)

    • Method Detail

      • getKeyGroupRange

        public org.apache.flink.runtime.state.KeyGroupRange getKeyGroupRange()
        Specified by:
        getKeyGroupRange in interface org.apache.flink.runtime.state.CheckpointableKeyedStateBackend<K>

      • setCurrentKey

        public void setCurrentKey​(K newKey)
        Specified by:
        setCurrentKey in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • getCurrentKey

        public K getCurrentKey()
        Specified by:
        getCurrentKey in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • getKeySerializer

        public org.apache.flink.api.common.typeutils.TypeSerializer<K> getKeySerializer()
        Specified by:
        getKeySerializer in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • getKeys

        public <N> java.util.stream.Stream<K> getKeys​(String state,
                                              N namespace)
        Specified by:
        getKeys in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • getKeysAndNamespaces

        public <N> java.util.stream.Stream<org.apache.flink.api.java.tuple.Tuple2<K,​N>> getKeysAndNamespaces​(String state)
        Specified by:
        getKeysAndNamespaces in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • dispose

        public void dispose()
        Specified by:
        dispose in interface org.apache.flink.util.Disposable
        Specified by:
        dispose in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • registerKeySelectionListener

        public void registerKeySelectionListener​(org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K> listener)
        Specified by:
        registerKeySelectionListener in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • deregisterKeySelectionListener

        public boolean deregisterKeySelectionListener​(org.apache.flink.runtime.state.KeyedStateBackend.KeySelectionListener<K> listener)
        Specified by:
        deregisterKeySelectionListener in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • applyToAllKeys

        public <N,​S extends org.apache.flink.api.common.state.State,​T> void applyToAllKeys​(N namespace,
                                                                                               org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer,
                                                                                               org.apache.flink.api.common.state.StateDescriptor<S,​T> stateDescriptor,
                                                                                               org.apache.flink.runtime.state.KeyedStateFunction<K,​S> function)
                                                                                        throws Exception
        Specified by:
        applyToAllKeys in interface org.apache.flink.runtime.state.KeyedStateBackend<K>
        Throws:
        Exception

      • getPartitionedState

        public <N,​S extends org.apache.flink.api.common.state.State> S getPartitionedState​(N namespace,
                                                                                         org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer,
                                                                                         org.apache.flink.api.common.state.StateDescriptor<S,​?> stateDescriptor)
                                                                                  throws Exception
        Specified by:
        getPartitionedState in interface org.apache.flink.runtime.state.KeyedStateBackend<K>
        Throws:
        Exception

      • snapshot

        @Nonnull
public RunnableFuture<org.apache.flink.runtime.state.SnapshotResult<org.apache.flink.runtime.state.KeyedStateHandle>> snapshot​(long checkpointId,
                                                                                                                               long timestamp,
                                                                                                                               @Nonnull
                                                                                                                               org.apache.flink.runtime.state.CheckpointStreamFactory streamFactory,
                                                                                                                               @Nonnull
                                                                                                                               org.apache.flink.runtime.checkpoint.CheckpointOptions checkpointOptions)
                                                                                                                        throws Exception
        Specified by:
        snapshot in interface org.apache.flink.runtime.state.Snapshotable<K>
        Throws:
        Exception

      • create

        @Nonnull
public <T extends org.apache.flink.runtime.state.heap.HeapPriorityQueueElement & org.apache.flink.runtime.state.PriorityComparable<? super T> & org.apache.flink.runtime.state.Keyed<?>> org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue<T> create​(@Nonnull
                                                                                                                                                                                                                                                                  String stateName,
                                                                                                                                                                                                                                                                  @Nonnull
                                                                                                                                                                                                                                                                  org.apache.flink.api.common.typeutils.TypeSerializer<T> byteOrderedElementSerializer)
        Specified by:
        create in interface org.apache.flink.runtime.state.PriorityQueueSetFactory

      • numKeyValueStateEntries

        @VisibleForTesting
public int numKeyValueStateEntries()
        Specified by:
        numKeyValueStateEntries in interface org.apache.flink.runtime.state.TestableKeyedStateBackend<K>

      • isSafeToReuseKVState

        public boolean isSafeToReuseKVState()
        Specified by:
        isSafeToReuseKVState in interface org.apache.flink.runtime.state.KeyedStateBackend<K>

      • savepoint

        @Nonnull
public org.apache.flink.runtime.state.SavepointResources<K> savepoint()
                                                               throws Exception
        Specified by:
        savepoint in interface org.apache.flink.runtime.state.CheckpointableKeyedStateBackend<K>
        Throws:
        Exception

      • notifyCheckpointComplete

        public void notifyCheckpointComplete​(long checkpointId)
                              throws Exception
        Specified by:
        notifyCheckpointComplete in interface org.apache.flink.api.common.state.CheckpointListener
        Throws:
        Exception

      • notifyCheckpointAborted

        public void notifyCheckpointAborted​(long checkpointId)
                             throws Exception
        Specified by:
        notifyCheckpointAborted in interface org.apache.flink.api.common.state.CheckpointListener
        Throws:
        Exception

      • getOrCreateKeyedState

        public <N,​S extends org.apache.flink.api.common.state.State,​T> S getOrCreateKeyedState​(org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer,
                                                                                                   org.apache.flink.api.common.state.StateDescriptor<S,​T> stateDescriptor)
                                                                                            throws Exception
        Specified by:
        getOrCreateKeyedState in interface org.apache.flink.runtime.state.KeyedStateBackend<K>
        Throws:
        Exception

      • createOrUpdateInternalState

        @Nonnull
public <N,​SV,​SEV,​S extends org.apache.flink.api.common.state.State,​IS extends S> IS createOrUpdateInternalState​(@Nonnull
                                                                                                                                        org.apache.flink.api.common.typeutils.TypeSerializer<N> namespaceSerializer,
                                                                                                                                        @Nonnull
                                                                                                                                        org.apache.flink.api.common.state.StateDescriptor<S,​SV> stateDesc,
                                                                                                                                        @Nonnull
                                                                                                                                        org.apache.flink.runtime.state.StateSnapshotTransformer.StateSnapshotTransformFactory<SEV> snapshotTransformFactory)
                                                                                                                                 throws Exception
        Specified by:
        createOrUpdateInternalState in interface org.apache.flink.runtime.state.KeyedStateFactory
        Throws:
        Exception

      • initMaterialization

        public Optional<org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationRunnable> initMaterialization()
                                                                                                                   throws Exception
        Initialize state materialization so that materialized data can be persisted durably and included into the checkpoint.

        This method is not thread safe. It should be called either under a lock or through task mailbox executor.

        Specified by:
        initMaterialization in interface org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget
        Returns:
        a tuple of - future snapshot result from the underlying state backend - a SequenceNumber identifying the latest change in the changelog
        Throws:
        Exception

      • handleMaterializationResult

        public void handleMaterializationResult​(org.apache.flink.runtime.state.SnapshotResult<org.apache.flink.runtime.state.KeyedStateHandle> materializedSnapshot,
                                        long materializationID,
                                        org.apache.flink.runtime.state.changelog.SequenceNumber upTo)
        This method is not thread safe. It should be called either under a lock or through task mailbox executor.
        Specified by:
        handleMaterializationResult in interface org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget

      • handleMaterializationFailureOrCancellation

        public void handleMaterializationFailureOrCancellation​(long materializationID,
                                                       org.apache.flink.runtime.state.changelog.SequenceNumber upTo,
                                                       Throwable cause)
        Specified by:
        handleMaterializationFailureOrCancellation in interface org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget

      • getDelegatedKeyedStateBackend

        public org.apache.flink.runtime.state.KeyedStateBackend<K> getDelegatedKeyedStateBackend​(boolean recursive)
        Specified by:
        getDelegatedKeyedStateBackend in interface org.apache.flink.runtime.state.TestableKeyedStateBackend<K>

      • notifyCheckpointSubsumed

        public void notifyCheckpointSubsumed​(long checkpointId)
                              throws Exception
        Specified by:
        notifyCheckpointSubsumed in interface org.apache.flink.api.common.state.InternalCheckpointListener
        Throws:
        Exception