[FLINK-38344][runtime-web] Fix the bug that The local files of the HistoryServer may risk never being deleted.

(cherry picked from commit 39a4628)

Author: RocMarshal

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java

@@ -197,15 +197,7 @@ public HistoryServer(
         webRefreshIntervalMillis =
                 config.get(HistoryServerOptions.HISTORY_SERVER_WEB_REFRESH_INTERVAL).toMillis();
 
-        String webDirectory = config.get(HistoryServerOptions.HISTORY_SERVER_WEB_DIR);
-        if (webDirectory == null) {
-            webDirectory =
-                    System.getProperty("java.io.tmpdir")
-                            + File.separator
-                            + "flink-web-history-"
-                            + UUID.randomUUID();
-        }
-        webDir = new File(webDirectory);
+        webDir = clearWebDir(config);
 
         boolean cleanupExpiredArchives =
                 config.get(HistoryServerOptions.HISTORY_SERVER_CLEANUP_EXPIRED_JOBS);
@@ -257,6 +249,34 @@ public HistoryServer(
                         HistoryServer.this::stop, HistoryServer.class.getSimpleName(), LOG);
     }
 
+    private File clearWebDir(Configuration config) throws IOException {
+        String webDirectory = config.get(HistoryServerOptions.HISTORY_SERVER_WEB_DIR);
+        if (webDirectory == null) {
+            webDirectory =
+                    System.getProperty("java.io.tmpdir")
+                            + File.separator
+                            + "flink-web-history-"
+                            + UUID.randomUUID();
+        }
+        final File webDir = new File(webDirectory);
+        LOG.info("Clear the web directory {}", webDir);
+        if (webDir.exists() && webDir.isDirectory() && webDir.listFiles() != null) {
+            // Reset the current working directory to eliminate the risk of local file leakage.
+            // This is because when the current process is forcibly terminated by an external
+            // command,
+            // the hook methods for cleaning up local files will not be called.
+            for (File subFile : webDir.listFiles()) {
+                FileUtils.deleteFileOrDirectory(subFile);
+            }
+        }
+        return webDir;
+    }
+
+    @VisibleForTesting
+    File getWebDir() {
+        return webDir;
+    }
+
     @VisibleForTesting
     int getWebPort() {
         return netty.getServerPort();

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/HistoryServerTest.java

@@ -270,6 +270,35 @@ void testRemainExpiredJob() throws Exception {
         runArchiveExpirationTest(false);
     }
 
+    @Test
+    void testClearWebDir() throws Exception {
+        // Test the path configured by 'historyserver.web.tmpdir' is clean.
+        Configuration historyServerConfig =
+                createTestConfiguration(
+                        HistoryServerOptions.HISTORY_SERVER_CLEANUP_EXPIRED_JOBS.defaultValue());
+        historyServerConfig.set(
+                HistoryServerOptions.HISTORY_SERVER_WEB_DIR, hsDirectory.toURI().toString());
+        HistoryServer hs = new HistoryServer(historyServerConfig);
+        assertInitializedHistoryServerWebDir(hs.getWebDir());
+
+        // Test the path configured by 'historyserver.web.tmpdir' is dirty.
+        new File(hsDirectory.toURI() + "/dirtyEmptySubDir").mkdir();
+        new File(hsDirectory.toURI() + "/dirtyEmptySubFile.json").createNewFile();
+        new File(hsDirectory.toURI() + "/overviews/dirtyEmptySubDir").mkdir();
+        new File(hsDirectory.toURI() + "/overviews/dirtyEmptySubFile.json").createNewFile();
+        new File(hsDirectory.toURI() + "/jobs/dirtyEmptySubDir").mkdir();
+        new File(hsDirectory.toURI() + "/jobs/dirtyEmptySubFile.json").createNewFile();
+        hs = new HistoryServer(historyServerConfig);
+        assertInitializedHistoryServerWebDir(hs.getWebDir());
+    }
+
+    private void assertInitializedHistoryServerWebDir(File historyWebDir) {
+
+        assertThat(historyWebDir.list()).containsExactlyInAnyOrder("overviews", "jobs");
+        assertThat(new File(historyWebDir, "overviews")).exists().isDirectory().isEmptyDirectory();
+        assertThat(new File(historyWebDir, "jobs").list()).containsExactly("overview.json");
+    }
+
     private void runArchiveExpirationTest(boolean cleanupExpiredJobs) throws Exception {
         int numExpiredJobs = cleanupExpiredJobs ? 1 : 0;
         int numJobs = 3;