...

To better separate concerns, we will deprecate remove the checkpoint storage methods from the StateBackend interface .  We will move and them to a new interface SnapshotStorage and provide two default implementations: JobManagerStorage and FileSystemStorageinterface.

public interface SnapshotStorage extends java.io.Serializable {

  CompletedCheckpointStorageLocation resolveCheckpoint(String externalPointer);

  CheckpointStorage createCheckpointStorage(JobID jobId);
}

Flink will provide two default implementations: JobManagerStorage and FileSystemStorage. JobManagerStorage and FileSystemStorage will maintain the same functionality as the implementations of these methods in MemoryStateBackend and FsStateBackend respectively. We will also provide a SnapshotStorageFactory to retain feature parity with existing functionality.

@PublicEvolving
public interface SnapshotStorageFactory<T extends SnapshotStorage> {

	T createFromConfig(ReadableConfig config, ClassLoader classLoader) throws IllegalConfigurationException, IOException;
}

...

Compatibility, Deprecation, and Migration Plan

To retain full backwards compatibility all deprecated StateBackends The three existing state backends (Memory, Fs, RocksDB) will implement the SnapshotStorage interface. When a state backend that implements this interface is provided, its snapshot storage methods will take precedence over any other provided configuration. This way, all existing state backend implementations will continue to function as they do today and any provided SnapshotStorage will be ignored.

Migration

be deprecated in favor of the new classes. In their JavaDoc and the release notes we will provide guidance on how to migrate to the new api in a compatible way. Because we are using the same internal data structures, users will be able to trivially migrate to the new API, ie, MemoryStateBackend should be replaced with HashMapStateBackend and JobManagerStorage.

All flink-conf configurations will be duplicated for the new state backend or snapshot storage instance, which ever is appropriate. Again, no functionality will be added or dropped in this change. Existing flink-conf keys will also be specified as deprecated keys of on the new state backends to ease migration.

To retain full backwards compatibility all deprecated StateBackends (Memory, Fs, RocksDB) will implement the SnapshotStorage interface. When a state backend that implements this interface is provided, its snapshot storage methods will take precedence over any other provided configuration. This way, all existing state backend implementations will continue to function as they do today and users will not see any changes in semantics.

While two methods will be removed from StateBackend, externally defined state backends will be able to migrate by simply adding `implements SnapshotStorage` to their implementations.