@Test public void testMultiThreadRestoreThreadPoolExceptionRethrow() { SpecifiedException expectedException = new SpecifiedException("throw exception while multi thread restore."); StreamStateHandle stateHandle = new StreamStateHandle() { @Override public FSDataInputStream openInputStream() throws IOException { throw expectedException; } @Override public void discardState() { } @Override public long getStateSize() { return 0; } }; Map<StateHandleID, StreamStateHandle> stateHandles = new HashMap<>(1); stateHandles.put(new StateHandleID("state1"), stateHandle); IncrementalRemoteKeyedStateHandle incrementalKeyedStateHandle = new IncrementalRemoteKeyedStateHandle( UUID.randomUUID(), KeyGroupRange.EMPTY_KEY_GROUP_RANGE, 1, stateHandles, stateHandles, stateHandle); try (RocksDBStateDownloader rocksDBStateDownloader = new RocksDBStateDownloader(5)) { rocksDBStateDownloader.transferAllStateDataToDirectory( incrementalKeyedStateHandle, new Path(temporaryFolder.newFolder().toURI()), new CloseableRegistry()); fail(); } catch (Exception e) { assertEquals(expectedException, e); } }
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);  }