@Override
    protected CompletableFuture<Void> respondToRequest(
            ChannelHandlerContext ctx,
            HttpRequest httpRequest,
            HandlerRequest<EmptyRequestBody> handlerRequest,
            RestfulGateway gateway)
            throws RestHandlerException {
        final ResourceID taskManagerId =
                handlerRequest.getPathParameter(TaskManagerIdPathParameter.class);

        String filename = getFileName(handlerRequest);
        final Tuple2<ResourceID, String> taskManagerIdAndFileName =
                new Tuple2<>(taskManagerId, filename);
        final CompletableFuture<TransientBlobKey> blobKeyFuture;
        try {
            blobKeyFuture = fileBlobKeys.get(taskManagerIdAndFileName);
        } 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 {
                                HandlerUtils.transferFile(ctx, file, httpRequest);
                            } catch (FlinkException e) {
                                throw new CompletionException(
                                        new FlinkException(
                                                "Could not transfer file to client.", e));
                            }
                        },
                        ctx.executor());

        return resultFuture
                .handle(
                        (Void ignored, Throwable throwable) -> {
                            if (throwable != null) {
                                return handleException(ctx, httpRequest, throwable, taskManagerId);
                            }
                            return CompletableFuture.<Void>completedFuture(null);
                        })
                .thenCompose(Function.identity());
    }
@Test
    void testFileServing() throws Exception {
        final Time cacheEntryDuration = Time.milliseconds(1000L);

        final Queue<CompletableFuture<TransientBlobKey>> requestFileUploads = new ArrayDeque<>(1);

        requestFileUploads.add(CompletableFuture.completedFuture(transientBlobKey1));

        final TestingTaskManagerFileHandler testingTaskManagerFileHandler =
                createTestTaskManagerFileHandler(
                        cacheEntryDuration, requestFileUploads, EXPECTED_TASK_MANAGER_ID);

        final File outputFile = TempDirUtils.newFile(temporaryFolder.toPath());
        final TestingChannelHandlerContext testingContext =
                new TestingChannelHandlerContext(outputFile);

        testingTaskManagerFileHandler.respondToRequest(
                testingContext, HTTP_REQUEST, handlerRequest, null);

        assertThat(outputFile).isNotEmpty();
        assertThat(FileUtils.readFileUtf8(outputFile)).isEqualTo(fileContent1);
    }