### Question:
AbstractTaskManagerFileHandler extends AbstractHandler<RestfulGateway, EmptyRequestBody, M> { @Override protected 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); if (cause instanceof RestHandlerException) { throw (RestHandlerException) cause; } else { throw new RestHandlerException("Could not retrieve file blob key future.", HttpResponseStatus.INTERNAL_SERVER_ERROR, e); } } 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()); resultFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { log.debug("Failed to transfer file from TaskExecutor {}.", taskManagerId, throwable); fileBlobKeys.invalidate(taskManagerId); final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable); final ErrorResponseBody errorResponseBody; final HttpResponseStatus httpResponseStatus; if (strippedThrowable instanceof UnknownTaskExecutorException) { errorResponseBody = new ErrorResponseBody("Unknown TaskExecutor " + taskManagerId + '.'); httpResponseStatus = HttpResponseStatus.NOT_FOUND; } else { errorResponseBody = new ErrorResponseBody("Internal server error: " + throwable.getMessage() + '.'); httpResponseStatus = INTERNAL_SERVER_ERROR; } HandlerUtils.sendErrorResponse( ctx, httpRequest, errorResponseBody, httpResponseStatus, responseHeaders); } }); } protected  AbstractTaskManagerFileHandler(
			@Nonnull CompletableFuture<String> localAddressFuture,
			@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);   }### Answer:
@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))); }