private void respondWithFile(ChannelHandlerContext ctx, HttpRequest request, String requestPath)
            throws IOException, ParseException, RestHandlerException {

        // make sure we request the "index.html" in case there is a directory request
        if (requestPath.endsWith("/")) {
            requestPath = requestPath + "index.html";
        }

        if (!requestPath.contains(".")) { // we assume that the path ends in either .html or .js
            requestPath = requestPath + ".json";
        }

        // convert to absolute path
        final File file = new File(rootPath, requestPath);

        if (!file.exists()) {
            // file does not exist. Try to load it with the classloader
            ClassLoader cl = HistoryServerStaticFileServerHandler.class.getClassLoader();

            try (InputStream resourceStream = cl.getResourceAsStream("web" + requestPath)) {
                boolean success = false;
                try {
                    if (resourceStream != null) {
                        URL root = cl.getResource("web");
                        URL requested = cl.getResource("web" + requestPath);

                        if (root != null && requested != null) {
                            URI rootURI = new URI(root.getPath()).normalize();
                            URI requestedURI = new URI(requested.getPath()).normalize();

                            // Check that we don't load anything from outside of the
                            // expected scope.
                            if (!rootURI.relativize(requestedURI).equals(requestedURI)) {
                                LOG.debug("Loading missing file from classloader: {}", requestPath);
                                // ensure that directory to file exists.
                                file.getParentFile().mkdirs();
                                Files.copy(resourceStream, file.toPath());

                                success = true;
                            }
                        }
                    }
                } catch (Throwable t) {
                    LOG.error("error while responding", t);
                } finally {
                    if (!success) {
                        LOG.debug("Unable to load requested file {} from classloader", requestPath);
                        throw new NotFoundException("File not found.");
                    }
                }
            }
        }

        StaticFileServerHandler.checkFileValidity(file, rootPath, LOG);

        // cache validation
        final String ifModifiedSince = request.headers().get(IF_MODIFIED_SINCE);
        if (ifModifiedSince != null && !ifModifiedSince.isEmpty()) {
            SimpleDateFormat dateFormatter =
                    new SimpleDateFormat(StaticFileServerHandler.HTTP_DATE_FORMAT, Locale.US);
            Date ifModifiedSinceDate = dateFormatter.parse(ifModifiedSince);

            // Only compare up to the second because the datetime format we send to the client
            // does not have milliseconds
            long ifModifiedSinceDateSeconds = ifModifiedSinceDate.getTime() / 1000;
            long fileLastModifiedSeconds = file.lastModified() / 1000;
            if (ifModifiedSinceDateSeconds == fileLastModifiedSeconds) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            "Responding 'NOT MODIFIED' for file '" + file.getAbsolutePath() + '\'');
                }

                StaticFileServerHandler.sendNotModified(ctx);
                return;
            }
        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("Responding with file '" + file.getAbsolutePath() + '\'');
        }

        // Don't need to close this manually. Netty's DefaultFileRegion will take care of it.
        final RandomAccessFile raf;
        try {
            raf = new RandomAccessFile(file, "r");
        } catch (FileNotFoundException e) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Could not find file {}.", file.getAbsolutePath());
            }
            HandlerUtils.sendErrorResponse(
                    ctx,
                    request,
                    new ErrorResponseBody("File not found."),
                    NOT_FOUND,
                    Collections.emptyMap());
            return;
        }

        try {
            long fileLength = raf.length();

            HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
            StaticFileServerHandler.setContentTypeHeader(response, file);

            // the job overview should be updated as soon as possible
            if (!requestPath.equals("/joboverview.json")) {
                StaticFileServerHandler.setDateAndCacheHeaders(response, file);
            }
            if (HttpUtil.isKeepAlive(request)) {
                response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            }
            HttpUtil.setContentLength(response, fileLength);

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

            // write the content.
            ChannelFuture lastContentFuture;
            if (ctx.pipeline().get(SslHandler.class) == null) {
                ctx.write(
                        new DefaultFileRegion(raf.getChannel(), 0, fileLength),
                        ctx.newProgressivePromise());
                lastContentFuture = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
            } else {
                lastContentFuture =
                        ctx.writeAndFlush(
                                new HttpChunkedInput(new ChunkedFile(raf, 0, fileLength, 8192)),
                                ctx.newProgressivePromise());
                // HttpChunkedInput will write the end marker (LastHttpContent) for us.
            }

            // close the connection, if no keep-alive is needed
            if (!HttpUtil.isKeepAlive(request)) {
                lastContentFuture.addListener(ChannelFutureListener.CLOSE);
            }
        } catch (Exception e) {
            raf.close();
            LOG.error("Failed to serve file.", e);
            throw new RestHandlerException("Internal server error.", INTERNAL_SERVER_ERROR);
        }
    }
@Test
    void testRespondWithFile(@TempDir Path tmpDir) throws Exception {
        final Path webDir = Files.createDirectory(tmpDir.resolve("webDir"));
        final Path uploadDir = Files.createDirectory(tmpDir.resolve("uploadDir"));

        Router router =
                new Router()
                        .addGet("/:*", new HistoryServerStaticFileServerHandler(webDir.toFile()));
        WebFrontendBootstrap webUI =
                new WebFrontendBootstrap(
                        router,
                        LoggerFactory.getLogger(HistoryServerStaticFileServerHandlerTest.class),
                        uploadDir.toFile(),
                        null,
                        "localhost",
                        0,
                        new Configuration());

        int port = webUI.getServerPort();
        try {
            // verify that 404 message is returned when requesting a non-existent file
            Tuple2<Integer, String> notFound404 =
                    HttpUtils.getFromHTTP("http://localhost:" + port + "/hello");
            assertThat(notFound404.f0).isEqualTo(404);
            assertThat(notFound404.f1).contains("not found");

            // verify that a) a file can be loaded using the ClassLoader and b) that the
            // HistoryServer
            // index_hs.html is injected
            Tuple2<Integer, String> index =
                    HttpUtils.getFromHTTP("http://localhost:" + port + "/index.html");
            assertThat(index.f0).isEqualTo(200);
            assertThat(index.f1).contains("Apache Flink Web Dashboard");

            // verify that index.html is appended if the request path ends on '/'
            Tuple2<Integer, String> index2 =
                    HttpUtils.getFromHTTP("http://localhost:" + port + "/");
            assertThat(index2).isEqualTo(index);

            // verify that a 405 message is returned when requesting a directory
            Files.createDirectory(webDir.resolve("dir.json"));
            Tuple2<Integer, String> dirNotFound =
                    HttpUtils.getFromHTTP("http://localhost:" + port + "/dir");
            assertThat(dirNotFound.f0).isEqualTo(405);
            assertThat(dirNotFound.f1).contains("not found");

            // verify that a 403 message is returned when requesting a file outside the webDir
            Files.createFile(tmpDir.resolve("secret"));
            Tuple2<Integer, String> dirOutsideDirectory =
                    HttpUtils.getFromHTTP("http://localhost:" + port + "/../secret");
            assertThat(dirOutsideDirectory.f0).isEqualTo(403);
            assertThat(dirOutsideDirectory.f1).contains("Forbidden");
        } finally {
            webUI.shutdown();
        }
    }