class TaskManagerLogHandler extends RedirectHandler<JobManagerGateway> implements WebHandler {

	@Override
	protected void respondAsLeader(final ChannelHandlerContext ctx, final RoutedRequest routedRequest, final JobManagerGateway jobManagerGateway) {
		if (cache == null) {
			CompletableFuture<Integer> blobPortFuture = jobManagerGateway.requestBlobServerPort(timeout);
			cache = blobPortFuture.thenApplyAsync(
				(Integer port) -> {
					try {
						return new TransientBlobCache(config, new InetSocketAddress(jobManagerGateway.getHostname(), port));
					} catch (IOException e) {
						throw new CompletionException(new FlinkException("Could not create TransientBlobCache.", e));
					}
				},
				executor);
		}

		final String taskManagerId = routedRequest.getRouteResult().param(TaskManagersHandler.TASK_MANAGER_ID_KEY);
		final HttpRequest request = routedRequest.getRequest();

		//fetch TaskManager logs if no other process is currently doing it
		if (lastRequestPending.putIfAbsent(taskManagerId, true) == null) {
			try {
				final String unescapedString;

				try {
					unescapedString = URLDecoder.decode(taskManagerId, "UTF-8");
				} catch (UnsupportedEncodingException e) {
					throw new FlinkException("Could not decode task manager id: " + taskManagerId + '.', e);
				}

				final ResourceID resourceId = new ResourceID(unescapedString);
				final CompletableFuture<Optional<Instance>> taskManagerFuture = jobManagerGateway.requestTaskManagerInstance(resourceId, timeout);

				CompletableFuture<TransientBlobKey> blobKeyFuture = taskManagerFuture.thenCompose(
					(Optional<Instance> optTMInstance) -> {
						Instance taskManagerInstance = optTMInstance.orElseThrow(
							() -> new CompletionException(new FlinkException("Could not find instance with " + resourceId + '.')));
						switch (fileMode) {
							case LOG:
								return taskManagerInstance.getTaskManagerGateway().requestTaskManagerLog(timeout);
							case STDOUT:
							default:
								return taskManagerInstance.getTaskManagerGateway().requestTaskManagerStdout(timeout);
						}
					}
				);

				CompletableFuture<String> logPathFuture = blobKeyFuture
					.thenCombineAsync(
						cache,
						(blobKey, blobCache) -> {
							//delete previous log file, if it is different than the current one
							HashMap<String, TransientBlobKey> lastSubmittedFile = fileMode == FileMode.LOG ? lastSubmittedLog : lastSubmittedStdout;
							if (lastSubmittedFile.containsKey(taskManagerId)) {
								// the BlobKey will almost certainly be different but the old file
								// may not exist anymore so we cannot rely on it and need to
								// download the new file anyway, even if the hashes match
								if (!Objects.equals(blobKey, lastSubmittedFile.get(taskManagerId))) {
									if (!blobCache.deleteFromCache(lastSubmittedFile.get(taskManagerId))) {
										throw new CompletionException(new FlinkException("Could not delete file for " + taskManagerId + '.'));
									}
									lastSubmittedFile.put(taskManagerId, blobKey);
								}
							} else {
								lastSubmittedFile.put(taskManagerId, blobKey);
							}
							try {
								return blobCache.getFile(blobKey).getAbsolutePath();
							} catch (IOException e) {
								throw new CompletionException(new FlinkException("Could not retrieve blob for " + blobKey + '.', e));
							}
						},
						executor);

				logPathFuture.exceptionally(
					failure -> {
						display(ctx, request, "Fetching TaskManager log failed.");
						LOG.error("Fetching TaskManager log failed.", failure);
						lastRequestPending.remove(taskManagerId);

						return null;
					});

				logPathFuture.thenAccept(
					filePath -> {
						File file = new File(filePath);
						final RandomAccessFile raf;
						try {
							raf = new RandomAccessFile(file, "r");
						} catch (FileNotFoundException e) {
							display(ctx, request, "Displaying TaskManager log failed.");
							LOG.error("Displaying TaskManager log failed.", e);

							return;
						}
						long fileLength;
						try {
							fileLength = raf.length();
						} catch (IOException ioe) {
							display(ctx, request, "Displaying TaskManager log failed.");
							LOG.error("Displaying TaskManager log failed.", ioe);
							try {
								raf.close();
							} catch (IOException e) {
								LOG.error("Could not close random access file.", e);
							}

							return;
						}
						final FileChannel fc = raf.getChannel();

						HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
						response.headers().set(CONTENT_TYPE, "text/plain");

						if (HttpHeaders.isKeepAlive(request)) {
							response.headers().set(CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
						}
						HttpHeaders.setContentLength(response, fileLength);

						// write the initial line and the header.
						ctx.write(response);

						// write the content.
						ChannelFuture lastContentFuture;
						final GenericFutureListener<Future<? super Void>> completionListener = future -> {
							lastRequestPending.remove(taskManagerId);
							fc.close();
							raf.close();
						};
						if (ctx.pipeline().get(SslHandler.class) == null) {
							ctx.write(
								new DefaultFileRegion(fc, 0, fileLength), ctx.newProgressivePromise())
									.addListener(completionListener);
							lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);

						} else {
							try {
								lastContentFuture = ctx.writeAndFlush(
									new HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)),
									ctx.newProgressivePromise())
									.addListener(completionListener);
							} catch (IOException e) {
								display(ctx, request, "Displaying TaskManager log failed.");
								LOG.warn("Could not write http data.", e);

								return;
							}
							// HttpChunkedInput will write the end marker (LastHttpContent) for us.
						}

						// close the connection, if no keep-alive is needed
						if (!HttpHeaders.isKeepAlive(request)) {
							lastContentFuture.addListener(ChannelFutureListener.CLOSE);
						}
					});
			} catch (Exception e) {
				display(ctx, request, "Error: " + e.getMessage());
				LOG.error("Fetching TaskManager log failed.", e);
				lastRequestPending.remove(taskManagerId);
			}
		} else {
			display(ctx, request, "loading...");
		}
	}

	public  TaskManagerLogHandler(
		GatewayRetriever<JobManagerGateway> retriever,
		Executor executor,
		CompletableFuture<String> localJobManagerAddressPromise,
		Time timeout,
		FileMode fileMode,
		Configuration config);

}