@Test public void testMultiThreadRestoreCorrectly() throws Exception { Random random = new Random(); int contentNum = 6; byte[][] contents = new byte[contentNum][]; for (int i = 0; i < contentNum; ++i) { contents[i] = new byte[random.nextInt(100000) + 1]; random.nextBytes(contents[i]); } List<StreamStateHandle> handles = new ArrayList<>(contentNum); for (int i = 0; i < contentNum; ++i) { handles.add(new ByteStreamStateHandle(String.format("state%d", i), contents[i])); } Map<StateHandleID, StreamStateHandle> sharedStates = new HashMap<>(contentNum); Map<StateHandleID, StreamStateHandle> privateStates = new HashMap<>(contentNum); for (int i = 0; i < contentNum; ++i) { sharedStates.put(new StateHandleID(String.format("sharedState%d", i)), handles.get(i)); privateStates.put(new StateHandleID(String.format("privateState%d", i)), handles.get(i)); } IncrementalRemoteKeyedStateHandle incrementalKeyedStateHandle = new IncrementalRemoteKeyedStateHandle( UUID.randomUUID(), KeyGroupRange.of(0, 1), 1, sharedStates, privateStates, handles.get(0)); Path dstPath = new Path(temporaryFolder.newFolder().toURI()); try (RocksDBStateDownloader rocksDBStateDownloader = new RocksDBStateDownloader(5)) { rocksDBStateDownloader.transferAllStateDataToDirectory(incrementalKeyedStateHandle, dstPath, new CloseableRegistry()); } for (int i = 0; i < contentNum; ++i) { assertStateContentEqual(contents[i], new Path(dstPath, String.format("sharedState%d", i))); } }
public void transferAllStateDataToDirectory( IncrementalRemoteKeyedStateHandle restoreStateHandle, Path dest, CloseableRegistry closeableRegistry) throws Exception { final Map<StateHandleID, StreamStateHandle> sstFiles = restoreStateHandle.getSharedState(); final Map<StateHandleID, StreamStateHandle> miscFiles = restoreStateHandle.getPrivateState(); downloadDataForAllStateHandles(sstFiles, dest, closeableRegistry); downloadDataForAllStateHandles(miscFiles, dest, closeableRegistry); }
RocksDBStateDownloader extends RocksDBStateDataTransfer { public void transferAllStateDataToDirectory( IncrementalRemoteKeyedStateHandle restoreStateHandle, Path dest, CloseableRegistry closeableRegistry) throws Exception { final Map<StateHandleID, StreamStateHandle> sstFiles = restoreStateHandle.getSharedState(); final Map<StateHandleID, StreamStateHandle> miscFiles = restoreStateHandle.getPrivateState(); downloadDataForAllStateHandles(sstFiles, dest, closeableRegistry); downloadDataForAllStateHandles(miscFiles, dest, closeableRegistry); } }
RocksDBStateDownloader extends RocksDBStateDataTransfer { public void transferAllStateDataToDirectory( IncrementalRemoteKeyedStateHandle restoreStateHandle, Path dest, CloseableRegistry closeableRegistry) throws Exception { final Map<StateHandleID, StreamStateHandle> sstFiles = restoreStateHandle.getSharedState(); final Map<StateHandleID, StreamStateHandle> miscFiles = restoreStateHandle.getPrivateState(); downloadDataForAllStateHandles(sstFiles, dest, closeableRegistry); downloadDataForAllStateHandles(miscFiles, dest, closeableRegistry); } RocksDBStateDownloader(int restoringThreadNum); }
RocksDBStateDownloader extends RocksDBStateDataTransfer { public void transferAllStateDataToDirectory( IncrementalRemoteKeyedStateHandle restoreStateHandle, Path dest, CloseableRegistry closeableRegistry) throws Exception { final Map<StateHandleID, StreamStateHandle> sstFiles = restoreStateHandle.getSharedState(); final Map<StateHandleID, StreamStateHandle> miscFiles = restoreStateHandle.getPrivateState(); downloadDataForAllStateHandles(sstFiles, dest, closeableRegistry); downloadDataForAllStateHandles(miscFiles, dest, closeableRegistry); } RocksDBStateDownloader(int restoringThreadNum); void transferAllStateDataToDirectory(
		IncrementalRemoteKeyedStateHandle restoreStateHandle,
		Path dest,
		CloseableRegistry closeableRegistry); }
RocksDBStateDownloader extends RocksDBStateDataTransfer { public void transferAllStateDataToDirectory( IncrementalRemoteKeyedStateHandle restoreStateHandle, Path dest, CloseableRegistry closeableRegistry) throws Exception { final Map<StateHandleID, StreamStateHandle> sstFiles = restoreStateHandle.getSharedState(); final Map<StateHandleID, StreamStateHandle> miscFiles = restoreStateHandle.getPrivateState(); downloadDataForAllStateHandles(sstFiles, dest, closeableRegistry); downloadDataForAllStateHandles(miscFiles, dest, closeableRegistry); } RocksDBStateDownloader(int restoringThreadNum); void transferAllStateDataToDirectory(
		IncrementalRemoteKeyedStateHandle restoreStateHandle,
		Path dest,
		CloseableRegistry closeableRegistry);  }