class 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);

}