RocksDBStateBackend extends AbstractStateBackend { @Override public CheckpointStreamFactory createStreamFactory(JobID jobId, String operatorIdentifier) throws IOException { return checkpointStreamBackend.createStreamFactory(jobId, operatorIdentifier); } RocksDBStateBackend(String checkpointDataUri);  RocksDBStateBackend(String checkpointDataUri, boolean enableIncrementalCheckpointing);  RocksDBStateBackend(URI checkpointDataUri);  RocksDBStateBackend(URI checkpointDataUri, boolean enableIncrementalCheckpointing);  RocksDBStateBackend(AbstractStateBackend checkpointStreamBackend);  RocksDBStateBackend(AbstractStateBackend checkpointStreamBackend, boolean enableIncrementalCheckpointing); @Override CheckpointStreamFactory createStreamFactory(JobID jobId,
			String operatorIdentifier); @Override CheckpointStreamFactory createSavepointStreamFactory(
			JobID jobId,
			String operatorIdentifier,
			String targetLocation); @Override AbstractKeyedStateBackend<K> createKeyedStateBackend(
			Environment env,
			JobID jobID,
			String operatorIdentifier,
			TypeSerializer<K> keySerializer,
			int numberOfKeyGroups,
			KeyGroupRange keyGroupRange,
			TaskKvStateRegistry kvStateRegistry); void setDbStoragePath(String path); void setDbStoragePaths(String... paths); String[] getDbStoragePaths(); void setPredefinedOptions(PredefinedOptions options); PredefinedOptions getPredefinedOptions(); void setOptions(OptionsFactory optionsFactory); OptionsFactory getOptions(); DBOptions getDbOptions(); ColumnFamilyOptions getColumnOptions(); @Override OperatorStateBackend createOperatorStateBackend(
		Environment env,
		String operatorIdentifier); @Override String toString();  }
@Test public void testSharedIncrementalStateDeRegistration() throws Exception { if (enableIncrementalCheckpointing) { AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE); try { ValueStateDescriptor<String> kvId = new ValueStateDescriptor<>("id", String.class, null); kvId.initializeSerializerUnlessSet(new ExecutionConfig()); ValueState<String> state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); Queue<IncrementalKeyedStateHandle> previousStateHandles = new LinkedList<>(); SharedStateRegistry sharedStateRegistry = spy(new SharedStateRegistry()); for (int checkpointId = 0; checkpointId < 3; ++checkpointId) { reset(sharedStateRegistry); backend.setCurrentKey(checkpointId); state.update("Hello-" + checkpointId); RunnableFuture<KeyedStateHandle> snapshot = backend.snapshot( checkpointId, checkpointId, createStreamFactory(), CheckpointOptions.forCheckpoint()); snapshot.run(); IncrementalKeyedStateHandle stateHandle = (IncrementalKeyedStateHandle) snapshot.get(); Map<StateHandleID, StreamStateHandle> sharedState = new HashMap<>(stateHandle.getSharedState()); stateHandle.registerSharedStates(sharedStateRegistry); for (Map.Entry<StateHandleID, StreamStateHandle> e : sharedState.entrySet()) { verify(sharedStateRegistry).registerReference( stateHandle.createSharedStateRegistryKeyFromFileName(e.getKey()), e.getValue()); } previousStateHandles.add(stateHandle); backend.notifyCheckpointComplete(checkpointId); if (previousStateHandles.size() > 1) { checkRemove(previousStateHandles.remove(), sharedStateRegistry); } } while (!previousStateHandles.isEmpty()) { reset(sharedStateRegistry); checkRemove(previousStateHandles.remove(), sharedStateRegistry); } } finally { IOUtils.closeQuietly(backend); backend.dispose(); } } }