org.apache.flink.runtime.persistence

Interface StateHandleStore<T extends Serializable,R extends ResourceVersion<R>>

Type Parameters:

T - Type of state

R - Type of ResourceVersion

All Known Implementing Classes:

KubernetesStateHandleStore, ZooKeeperStateHandleStore

public interface StateHandleStore<T extends Serializable,R extends ResourceVersion<R>>

Class which stores state via the provided RetrievableStateStorageHelper and writes the returned state handle to distributed coordination system(e.g. Zookeeper, Kubernetes, etc.).

To avoid concurrent modification issues, the implementation needs to ensure that only the leader could update the state store.

Even some methods name contains the "lock"(e.g. getAndLock(java.lang.String)), it does not mean the implementation has to actually lock a specific state handle. Also it could have an empty implementation for release operation.

Nested Class Summary

Nested Classes

Modifier and Type	Interface and Description
static class	StateHandleStore.AlreadyExistException The key already exists in ConfigMap or the Zookeeper node already exists.
static class	StateHandleStore.NotExistException The key does not exist in ConfigMap or the Zookeeper node does not exists.

Method Summary

Modifier and Type	Method and Description
RetrievableStateHandle<T>	addAndLock(String name, T state) Persist the state to distributed storage(e.g.
void	clearEntries() Only clears all the state handle pointers on Kubernetes or ZooKeeper.
R	exists(String name) Returns resource version of state handle with specific name on the underlying storage.
List<Tuple2<RetrievableStateHandle<T>,String>>	getAllAndLock() Gets all available state handles from the storage.
Collection<String>	getAllHandles() Return a list of all valid name for state handles.
RetrievableStateHandle<T>	getAndLock(String name) Gets the RetrievableStateHandle stored with the given name.
void	release(String name) Releases the lock on the specific state handle so that it could be deleted by other StateHandleStore.
void	releaseAll() Releases all the locks on corresponding state handle so that it could be deleted by other StateHandleStore.
boolean	releaseAndTryRemove(String name) Releases the lock for the given state handle and tries to remove the state handle if it is no longer locked.
void	releaseAndTryRemoveAll() Releases and removes all the states.
void	replace(String name, R resourceVersion, T state) Replaces a state handle in the distributed coordination system and discards the old state handle.

Method Detail

addAndLock

RetrievableStateHandle<T> addAndLock(String name,
                                     T state)
                              throws PossibleInconsistentStateException,
                                     Exception

Persist the state to distributed storage(e.g. S3, HDFS, etc.). And then creates a state handle, stores it in the distributed coordination system(e.g. ZooKeeper, Kubernetes, etc.).

Parameters:

name - Key name in ConfigMap or child path name in ZooKeeper

state - State to be added

Throws:

StateHandleStore.AlreadyExistException - if the name already exists

PossibleInconsistentStateException - if the write operation failed. This indicates that it's not clear whether the new state was successfully written to distributed coordination system or not. No state was discarded. Proper error handling has to be applied on the caller's side.

Exception - if persisting state or writing state handle failed

replace

void replace(String name,
             R resourceVersion,
             T state)
      throws PossibleInconsistentStateException,
             Exception

Replaces a state handle in the distributed coordination system and discards the old state handle.

Parameters:

name - Key name in ConfigMap or child path name in ZooKeeper

resourceVersion - resource version of previous storage object. If the resource version does not match, the replace operation will fail. Since there is an unexpected update operation snuck in.

state - State to be replace with

Throws:

StateHandleStore.NotExistException - if the name does not exist

PossibleInconsistentStateException - if a failure occurred during the update operation for which it's unclear whether the operation actually succeeded or not. No state was discarded. The method's caller should handle this case properly.

Exception - if persisting state or writing state handle failed

exists

R exists(String name)
  throws Exception

Returns resource version of state handle with specific name on the underlying storage.

Parameters:

name - Key name in ConfigMap or child path name in ZooKeeper

Returns:

current resource version on the underlying storage.

Throws:

Exception - if the check existence operation failed

getAndLock

RetrievableStateHandle<T> getAndLock(String name)
                              throws Exception

Gets the RetrievableStateHandle stored with the given name.

Parameters:

name - Key name in ConfigMap or child path name in ZooKeeper

Returns:

The retrieved state handle

Throws:

IOException - if the method failed to deserialize the stored state handle

StateHandleStore.NotExistException - when the name does not exist

Exception - if get state handle failed

getAllAndLock

List<Tuple2<RetrievableStateHandle<T>,String>> getAllAndLock()
                                                      throws Exception

Gets all available state handles from the storage.

Returns:

All retrieved state handles.

Throws:

Exception - if get state handle operation failed

getAllHandles

Collection<String> getAllHandles()
                          throws Exception

Return a list of all valid name for state handles. The result might contain nodes that are marked for deletion.

Returns:

List of valid state handle name. The name is key name in ConfigMap or child path name in ZooKeeper.

Throws:

Exception - if get handle operation failed

releaseAndTryRemove

boolean releaseAndTryRemove(String name)
                     throws Exception

Releases the lock for the given state handle and tries to remove the state handle if it is no longer locked. Also the state on the external storage will be discarded.

Parameters:

name - Key name in ConfigMap or child path name in ZooKeeper

Returns:

true if the state handle is removed (also if it didn't exist in the first place); otherwise false.

Throws:

Exception - if releasing, removing the handles or discarding the state failed

releaseAndTryRemoveAll

void releaseAndTryRemoveAll()
                     throws Exception

Releases and removes all the states. Not only the state handles in the distributed coordination system will be removed, but also the real state data on the distributed storage will be discarded. This method should be implemented in a idempotent manner, i.e. failing calls should be able to be repeated.

Throws:

Exception - if releasing, removing the handles or discarding the state failed. The removal of the states should happen independently from each other, i.e. the method call fails if at least one removal failed. But the removal of any other state should still be performed.

clearEntries

void clearEntries()
           throws Exception

Only clears all the state handle pointers on Kubernetes or ZooKeeper.

Throws:

Exception - if removing the handles failed

release

void release(String name)
      throws Exception

Releases the lock on the specific state handle so that it could be deleted by other StateHandleStore. If no lock exists or the underlying storage does not support, nothing will happen.

Throws:

Exception - if releasing the lock

releaseAll

void releaseAll()
         throws Exception

Releases all the locks on corresponding state handle so that it could be deleted by other StateHandleStore. If no lock exists or the underlying storage does not support, nothing will happen.

Throws:

Exception - if releasing the locks