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

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

}