@Test public void testFileServing() throws Exception { final Time cacheEntryDuration = Time.milliseconds(1000L); final Queue<CompletableFuture<TransientBlobKey>> requestFileUploads = new ArrayDeque<>(1); requestFileUploads.add(CompletableFuture.completedFuture(transientBlobKey1)); final TestTaskManagerFileHandler testTaskManagerFileHandler = createTestTaskManagerFileHandler(cacheEntryDuration, requestFileUploads, EXPECTED_TASK_MANAGER_ID); final File outputFile = temporaryFolder.newFile(); final TestingChannelHandlerContext testingContext = new TestingChannelHandlerContext(outputFile); testTaskManagerFileHandler.respondToRequest( testingContext, HTTP_REQUEST, handlerRequest, null); assertThat(outputFile.length(), is(greaterThan(0L))); assertThat(FileUtils.readFileUtf8(outputFile), is(equalTo(fileContent1))); }
@Override protected CompletableFuture<Void> respondToRequest(ChannelHandlerContext ctx, HttpRequest httpRequest, HandlerRequest<EmptyRequestBody, M> handlerRequest, RestfulGateway gateway) throws RestHandlerException { final ResourceID taskManagerId = handlerRequest.getPathParameter(TaskManagerIdPathParameter.class); final CompletableFuture<TransientBlobKey> blobKeyFuture; try { blobKeyFuture = fileBlobKeys.get(taskManagerId); } catch (ExecutionException e) { final Throwable cause = ExceptionUtils.stripExecutionException(e); throw new RestHandlerException("Could not retrieve file blob key future.", HttpResponseStatus.INTERNAL_SERVER_ERROR, cause); } final CompletableFuture<Void> resultFuture = blobKeyFuture.thenAcceptAsync( (TransientBlobKey blobKey) -> { final File file; try { file = transientBlobService.getFile(blobKey); } catch (IOException e) { throw new CompletionException(new FlinkException("Could not retrieve file from transient blob store.", e)); } try { transferFile( ctx, file, httpRequest); } catch (FlinkException e) { throw new CompletionException(new FlinkException("Could not transfer file to client.", e)); } }, ctx.executor()); return resultFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { log.error("Failed to transfer file from TaskExecutor {}.", taskManagerId, throwable); fileBlobKeys.invalidate(taskManagerId); final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable); if (strippedThrowable instanceof UnknownTaskExecutorException) { throw new CompletionException( new NotFoundException( String.format("Failed to transfer file from TaskExecutor %s because it was unknown.", taskManagerId), strippedThrowable)); } else { throw new CompletionException( new FlinkException( String.format("Failed to transfer file from TaskExecutor %s.", taskManagerId), strippedThrowable)); } } }); }
AbstractTaskManagerFileHandler extends AbstractHandler<RestfulGateway, EmptyRequestBody, M> { @Override protected CompletableFuture<Void> respondToRequest(ChannelHandlerContext ctx, HttpRequest httpRequest, HandlerRequest<EmptyRequestBody, M> handlerRequest, RestfulGateway gateway) throws RestHandlerException { final ResourceID taskManagerId = handlerRequest.getPathParameter(TaskManagerIdPathParameter.class); final CompletableFuture<TransientBlobKey> blobKeyFuture; try { blobKeyFuture = fileBlobKeys.get(taskManagerId); } catch (ExecutionException e) { final Throwable cause = ExceptionUtils.stripExecutionException(e); throw new RestHandlerException("Could not retrieve file blob key future.", HttpResponseStatus.INTERNAL_SERVER_ERROR, cause); } final CompletableFuture<Void> resultFuture = blobKeyFuture.thenAcceptAsync( (TransientBlobKey blobKey) -> { final File file; try { file = transientBlobService.getFile(blobKey); } catch (IOException e) { throw new CompletionException(new FlinkException("Could not retrieve file from transient blob store.", e)); } try { transferFile( ctx, file, httpRequest); } catch (FlinkException e) { throw new CompletionException(new FlinkException("Could not transfer file to client.", e)); } }, ctx.executor()); return resultFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { log.error("Failed to transfer file from TaskExecutor {}.", taskManagerId, throwable); fileBlobKeys.invalidate(taskManagerId); final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable); if (strippedThrowable instanceof UnknownTaskExecutorException) { throw new CompletionException( new NotFoundException( String.format("Failed to transfer file from TaskExecutor %s because it was unknown.", taskManagerId), strippedThrowable)); } else { throw new CompletionException( new FlinkException( String.format("Failed to transfer file from TaskExecutor %s.", taskManagerId), strippedThrowable)); } } }); } }
AbstractTaskManagerFileHandler extends AbstractHandler<RestfulGateway, EmptyRequestBody, M> { @Override protected CompletableFuture<Void> respondToRequest(ChannelHandlerContext ctx, HttpRequest httpRequest, HandlerRequest<EmptyRequestBody, M> handlerRequest, RestfulGateway gateway) throws RestHandlerException { final ResourceID taskManagerId = handlerRequest.getPathParameter(TaskManagerIdPathParameter.class); final CompletableFuture<TransientBlobKey> blobKeyFuture; try { blobKeyFuture = fileBlobKeys.get(taskManagerId); } catch (ExecutionException e) { final Throwable cause = ExceptionUtils.stripExecutionException(e); throw new RestHandlerException("Could not retrieve file blob key future.", HttpResponseStatus.INTERNAL_SERVER_ERROR, cause); } final CompletableFuture<Void> resultFuture = blobKeyFuture.thenAcceptAsync( (TransientBlobKey blobKey) -> { final File file; try { file = transientBlobService.getFile(blobKey); } catch (IOException e) { throw new CompletionException(new FlinkException("Could not retrieve file from transient blob store.", e)); } try { transferFile( ctx, file, httpRequest); } catch (FlinkException e) { throw new CompletionException(new FlinkException("Could not transfer file to client.", e)); } }, ctx.executor()); return resultFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { log.error("Failed to transfer file from TaskExecutor {}.", taskManagerId, throwable); fileBlobKeys.invalidate(taskManagerId); final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable); if (strippedThrowable instanceof UnknownTaskExecutorException) { throw new CompletionException( new NotFoundException( String.format("Failed to transfer file from TaskExecutor %s because it was unknown.", taskManagerId), strippedThrowable)); } else { throw new CompletionException( new FlinkException( String.format("Failed to transfer file from TaskExecutor %s.", taskManagerId), strippedThrowable)); } } }); } protected  AbstractTaskManagerFileHandler(
			@Nonnull GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			@Nonnull Time timeout,
			@Nonnull Map<String, String> responseHeaders,
			@Nonnull UntypedResponseMessageHeaders<EmptyRequestBody, M> untypedResponseMessageHeaders,
			@Nonnull GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			@Nonnull TransientBlobService transientBlobService,
			@Nonnull Time cacheEntryDuration); }
AbstractTaskManagerFileHandler extends AbstractHandler<RestfulGateway, EmptyRequestBody, M> { @Override protected CompletableFuture<Void> respondToRequest(ChannelHandlerContext ctx, HttpRequest httpRequest, HandlerRequest<EmptyRequestBody, M> handlerRequest, RestfulGateway gateway) throws RestHandlerException { final ResourceID taskManagerId = handlerRequest.getPathParameter(TaskManagerIdPathParameter.class); final CompletableFuture<TransientBlobKey> blobKeyFuture; try { blobKeyFuture = fileBlobKeys.get(taskManagerId); } catch (ExecutionException e) { final Throwable cause = ExceptionUtils.stripExecutionException(e); throw new RestHandlerException("Could not retrieve file blob key future.", HttpResponseStatus.INTERNAL_SERVER_ERROR, cause); } final CompletableFuture<Void> resultFuture = blobKeyFuture.thenAcceptAsync( (TransientBlobKey blobKey) -> { final File file; try { file = transientBlobService.getFile(blobKey); } catch (IOException e) { throw new CompletionException(new FlinkException("Could not retrieve file from transient blob store.", e)); } try { transferFile( ctx, file, httpRequest); } catch (FlinkException e) { throw new CompletionException(new FlinkException("Could not transfer file to client.", e)); } }, ctx.executor()); return resultFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { log.error("Failed to transfer file from TaskExecutor {}.", taskManagerId, throwable); fileBlobKeys.invalidate(taskManagerId); final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable); if (strippedThrowable instanceof UnknownTaskExecutorException) { throw new CompletionException( new NotFoundException( String.format("Failed to transfer file from TaskExecutor %s because it was unknown.", taskManagerId), strippedThrowable)); } else { throw new CompletionException( new FlinkException( String.format("Failed to transfer file from TaskExecutor %s.", taskManagerId), strippedThrowable)); } } }); } protected  AbstractTaskManagerFileHandler(
			@Nonnull GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			@Nonnull Time timeout,
			@Nonnull Map<String, String> responseHeaders,
			@Nonnull UntypedResponseMessageHeaders<EmptyRequestBody, M> untypedResponseMessageHeaders,
			@Nonnull GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			@Nonnull TransientBlobService transientBlobService,
			@Nonnull Time cacheEntryDuration);  }
AbstractTaskManagerFileHandler extends AbstractHandler<RestfulGateway, EmptyRequestBody, M> { @Override protected CompletableFuture<Void> respondToRequest(ChannelHandlerContext ctx, HttpRequest httpRequest, HandlerRequest<EmptyRequestBody, M> handlerRequest, RestfulGateway gateway) throws RestHandlerException { final ResourceID taskManagerId = handlerRequest.getPathParameter(TaskManagerIdPathParameter.class); final CompletableFuture<TransientBlobKey> blobKeyFuture; try { blobKeyFuture = fileBlobKeys.get(taskManagerId); } catch (ExecutionException e) { final Throwable cause = ExceptionUtils.stripExecutionException(e); throw new RestHandlerException("Could not retrieve file blob key future.", HttpResponseStatus.INTERNAL_SERVER_ERROR, cause); } final CompletableFuture<Void> resultFuture = blobKeyFuture.thenAcceptAsync( (TransientBlobKey blobKey) -> { final File file; try { file = transientBlobService.getFile(blobKey); } catch (IOException e) { throw new CompletionException(new FlinkException("Could not retrieve file from transient blob store.", e)); } try { transferFile( ctx, file, httpRequest); } catch (FlinkException e) { throw new CompletionException(new FlinkException("Could not transfer file to client.", e)); } }, ctx.executor()); return resultFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { log.error("Failed to transfer file from TaskExecutor {}.", taskManagerId, throwable); fileBlobKeys.invalidate(taskManagerId); final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable); if (strippedThrowable instanceof UnknownTaskExecutorException) { throw new CompletionException( new NotFoundException( String.format("Failed to transfer file from TaskExecutor %s because it was unknown.", taskManagerId), strippedThrowable)); } else { throw new CompletionException( new FlinkException( String.format("Failed to transfer file from TaskExecutor %s.", taskManagerId), strippedThrowable)); } } }); } protected  AbstractTaskManagerFileHandler(
			@Nonnull GatewayRetriever<? extends RestfulGateway> leaderRetriever,
			@Nonnull Time timeout,
			@Nonnull Map<String, String> responseHeaders,
			@Nonnull UntypedResponseMessageHeaders<EmptyRequestBody, M> untypedResponseMessageHeaders,
			@Nonnull GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			@Nonnull TransientBlobService transientBlobService,
			@Nonnull Time cacheEntryDuration);   }