@Test public void testMultiThreadUploadThreadPoolExceptionRethrow() throws IOException { SpecifiedException expectedException = new SpecifiedException("throw exception while multi thread upload states."); CheckpointStreamFactory.CheckpointStateOutputStream outputStream = createFailingCheckpointStateOutputStream(expectedException); CheckpointStreamFactory checkpointStreamFactory = (CheckpointedStateScope scope) -> outputStream; File file = temporaryFolder.newFile(String.valueOf(UUID.randomUUID())); generateRandomFileContent(file.getPath(), 20); Map<StateHandleID, Path> filePaths = new HashMap<>(1); filePaths.put(new StateHandleID("mockHandleID"), new Path(file.getPath())); try (RocksDBStateUploader rocksDBStateUploader = new RocksDBStateUploader(5)) { rocksDBStateUploader.uploadFilesToCheckpointFs(filePaths, checkpointStreamFactory, new CloseableRegistry()); fail(); } catch (Exception e) { assertEquals(expectedException, e); } }
public Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs( @Nonnull Map<StateHandleID, Path> files, CheckpointStreamFactory checkpointStreamFactory, CloseableRegistry closeableRegistry) throws Exception { Map<StateHandleID, StreamStateHandle> handles = new HashMap<>(); Map<StateHandleID, CompletableFuture<StreamStateHandle>> futures = createUploadFutures(files, checkpointStreamFactory, closeableRegistry); try { FutureUtils.waitForAll(futures.values()).get(); for (Map.Entry<StateHandleID, CompletableFuture<StreamStateHandle>> entry : futures.entrySet()) { handles.put(entry.getKey(), entry.getValue().get()); } } catch (ExecutionException e) { Throwable throwable = ExceptionUtils.stripExecutionException(e); throwable = ExceptionUtils.stripException(throwable, RuntimeException.class); if (throwable instanceof IOException) { throw (IOException) throwable; } else { throw new FlinkRuntimeException("Failed to download data for state handles.", e); } } return handles; }
RocksDBStateUploader extends RocksDBStateDataTransfer { public Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs( @Nonnull Map<StateHandleID, Path> files, CheckpointStreamFactory checkpointStreamFactory, CloseableRegistry closeableRegistry) throws Exception { Map<StateHandleID, StreamStateHandle> handles = new HashMap<>(); Map<StateHandleID, CompletableFuture<StreamStateHandle>> futures = createUploadFutures(files, checkpointStreamFactory, closeableRegistry); try { FutureUtils.waitForAll(futures.values()).get(); for (Map.Entry<StateHandleID, CompletableFuture<StreamStateHandle>> entry : futures.entrySet()) { handles.put(entry.getKey(), entry.getValue().get()); } } catch (ExecutionException e) { Throwable throwable = ExceptionUtils.stripExecutionException(e); throwable = ExceptionUtils.stripException(throwable, RuntimeException.class); if (throwable instanceof IOException) { throw (IOException) throwable; } else { throw new FlinkRuntimeException("Failed to download data for state handles.", e); } } return handles; } }
RocksDBStateUploader extends RocksDBStateDataTransfer { public Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs( @Nonnull Map<StateHandleID, Path> files, CheckpointStreamFactory checkpointStreamFactory, CloseableRegistry closeableRegistry) throws Exception { Map<StateHandleID, StreamStateHandle> handles = new HashMap<>(); Map<StateHandleID, CompletableFuture<StreamStateHandle>> futures = createUploadFutures(files, checkpointStreamFactory, closeableRegistry); try { FutureUtils.waitForAll(futures.values()).get(); for (Map.Entry<StateHandleID, CompletableFuture<StreamStateHandle>> entry : futures.entrySet()) { handles.put(entry.getKey(), entry.getValue().get()); } } catch (ExecutionException e) { Throwable throwable = ExceptionUtils.stripExecutionException(e); throwable = ExceptionUtils.stripException(throwable, RuntimeException.class); if (throwable instanceof IOException) { throw (IOException) throwable; } else { throw new FlinkRuntimeException("Failed to download data for state handles.", e); } } return handles; } RocksDBStateUploader(int numberOfSnapshottingThreads); }
RocksDBStateUploader extends RocksDBStateDataTransfer { public Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs( @Nonnull Map<StateHandleID, Path> files, CheckpointStreamFactory checkpointStreamFactory, CloseableRegistry closeableRegistry) throws Exception { Map<StateHandleID, StreamStateHandle> handles = new HashMap<>(); Map<StateHandleID, CompletableFuture<StreamStateHandle>> futures = createUploadFutures(files, checkpointStreamFactory, closeableRegistry); try { FutureUtils.waitForAll(futures.values()).get(); for (Map.Entry<StateHandleID, CompletableFuture<StreamStateHandle>> entry : futures.entrySet()) { handles.put(entry.getKey(), entry.getValue().get()); } } catch (ExecutionException e) { Throwable throwable = ExceptionUtils.stripExecutionException(e); throwable = ExceptionUtils.stripException(throwable, RuntimeException.class); if (throwable instanceof IOException) { throw (IOException) throwable; } else { throw new FlinkRuntimeException("Failed to download data for state handles.", e); } } return handles; } RocksDBStateUploader(int numberOfSnapshottingThreads); Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs(
		@Nonnull Map<StateHandleID, Path> files,
		CheckpointStreamFactory checkpointStreamFactory,
		CloseableRegistry closeableRegistry); }
RocksDBStateUploader extends RocksDBStateDataTransfer { public Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs( @Nonnull Map<StateHandleID, Path> files, CheckpointStreamFactory checkpointStreamFactory, CloseableRegistry closeableRegistry) throws Exception { Map<StateHandleID, StreamStateHandle> handles = new HashMap<>(); Map<StateHandleID, CompletableFuture<StreamStateHandle>> futures = createUploadFutures(files, checkpointStreamFactory, closeableRegistry); try { FutureUtils.waitForAll(futures.values()).get(); for (Map.Entry<StateHandleID, CompletableFuture<StreamStateHandle>> entry : futures.entrySet()) { handles.put(entry.getKey(), entry.getValue().get()); } } catch (ExecutionException e) { Throwable throwable = ExceptionUtils.stripExecutionException(e); throwable = ExceptionUtils.stripException(throwable, RuntimeException.class); if (throwable instanceof IOException) { throw (IOException) throwable; } else { throw new FlinkRuntimeException("Failed to download data for state handles.", e); } } return handles; } RocksDBStateUploader(int numberOfSnapshottingThreads); Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs(
		@Nonnull Map<StateHandleID, Path> files,
		CheckpointStreamFactory checkpointStreamFactory,
		CloseableRegistry closeableRegistry);  }