[FLINK-37504] Refactor + Unit test LocalFSWatchSingleton

Author: JTaky

flink-core/src/main/java/org/apache/flink/core/security/watch/LocalFSDirectoryWatcher.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.security.watch;
+
+import java.io.IOException;
+import java.nio.file.Path;
+
+public interface LocalFSDirectoryWatcher {
+
+    void registerDirectory(Path[] dirsToWatch, LocalFSWatchServiceListener listener)
+            throws IOException;
+}

flink-core/src/main/java/org/apache/flink/core/security/watch/LocalFSWatchService.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.core.security.watch;
 
+import org.apache.flink.annotation.VisibleForTesting;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,6 +28,7 @@
 import java.nio.file.WatchKey;
 import java.nio.file.WatchService;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
 import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE;
@@ -35,8 +38,11 @@
 public class LocalFSWatchService extends Thread {
     private static final Logger LOG = LoggerFactory.getLogger(LocalFSWatchService.class);
 
+    @VisibleForTesting AtomicBoolean running = new AtomicBoolean(false);
+
     public void run() {
         try {
+            running.set(true);
             while (true) {
                 for (Map.Entry<WatchService, LocalFSWatchServiceListener> entry :
                         LocalFSWatchSingleton.getInstance().watchers.entrySet()) {
@@ -47,8 +53,6 @@ public void run() {
                     }
                     LOG.debug("Watch key arrived");
                     for (WatchEvent<?> watchEvent : watchKey.pollEvents()) {
-                        System.out.println(watchEvent.kind());
-                        System.out.println(watchEvent.context());
                         if (watchEvent.kind() == OVERFLOW) {
                             LOG.error("Filesystem events may have been lost or discarded");
                             Thread.yield();
@@ -67,6 +71,7 @@ public void run() {
             }
         } catch (Exception e) {
             LOG.error("Filesystem watcher received exception and stopped: ", e);
+            running.set(false);
             throw new RuntimeException(e);
         }
     }

flink-core/src/main/java/org/apache/flink/core/security/watch/LocalFSWatchServiceListener.java

@@ -1,14 +1,13 @@
-package org.apache.flink.core.security.watch;
-
 /*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ *    http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -17,14 +16,79 @@
  * limitations under the License.
  */
 
+package org.apache.flink.core.security.watch;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicReference;
 
 public interface LocalFSWatchServiceListener {
+
+    enum ReloadState {
+        CLEAN, // Context is up to date
+        DIRTY, // Context needs reloading
+        RELOADING // Context is currently being reloaded
+    }
+
+    @FunctionalInterface
+    interface ContextLoader {
+
+        void loadContext() throws Exception;
+    }
+
+    Logger LOG = LoggerFactory.getLogger(LocalFSWatchServiceListener.class);
+
+    /**
+     * Get the current reload state. Implementations should provide their own state management.
+     *
+     * @return the current reload state
+     */
+    AtomicReference<ReloadState> getReloadStateReference();
+
     default void onWatchStarted(Path realDirectoryPath) {}
 
     default void onFileOrDirectoryCreated(Path relativePath) {}
 
     default void onFileOrDirectoryDeleted(Path relativePath) {}
 
-    default void onFileOrDirectoryModified(Path relativePath) {}
+    default void onFileOrDirectoryModified(Path relativePath) {
+        getReloadStateReference().compareAndSet(ReloadState.CLEAN, ReloadState.DIRTY);
+    }
+
+    default boolean reloadContextIfNeeded(ContextLoader loader) {
+        AtomicReference<ReloadState> reloadState = getReloadStateReference();
+        // Only one thread can transition from DIRTY to RELOADING
+        if (reloadState.compareAndSet(ReloadState.DIRTY, ReloadState.RELOADING)) {
+            try {
+                loader.loadContext();
+                // Successfully loaded, mark as clean
+                reloadState.set(ReloadState.CLEAN);
+                return true;
+            } catch (Exception e) {
+                LOG.warn("Failed to reload context", e);
+                // Failed to load, mark as dirty for retry
+                reloadState.set(ReloadState.DIRTY);
+            }
+        }
+        return false;
+        // If state is CLEAN, do nothing
+        // If state is RELOADING, another thread is handling it, so we can proceed with current
+        // context
+    }
+
+    /**
+     * Abstract base class that provides a default implementation of LocalFSWatchServiceListener
+     * with instance-level reload state management.
+     */
+    abstract class AbstractLocalFSWatchServiceListener implements LocalFSWatchServiceListener {
+        private final AtomicReference<ReloadState> reloadState =
+                new AtomicReference<>(ReloadState.CLEAN);
+
+        @Override
+        public final AtomicReference<ReloadState> getReloadStateReference() {
+            return reloadState;
+        }
+    }
 }

flink-core/src/main/java/org/apache/flink/core/security/watch/LocalFSWatchSingleton.java

@@ -28,18 +28,18 @@
 import static java.nio.file.StandardWatchEventKinds.ENTRY_DELETE;
 import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
 
-public final class LocalFSWatchSingleton {
+public final class LocalFSWatchSingleton implements LocalFSDirectoryWatcher {
     // The field must be declared volatile so that double check lock would work
     // correctly.
-    private static volatile LocalFSWatchSingleton instance;
+    private static volatile LocalFSDirectoryWatcher instance;
 
     ConcurrentHashMap<WatchService, LocalFSWatchServiceListener> watchers =
             new ConcurrentHashMap<>();
 
     private LocalFSWatchSingleton() {}
 
     public static LocalFSWatchSingleton getInstance() {
-        LocalFSWatchSingleton result = instance;
+        LocalFSDirectoryWatcher result = instance;
         if (result != null) {
             return result;
         }
@@ -51,15 +51,16 @@ public static LocalFSWatchSingleton getInstance() {
         }
     }
 
-    public void registerPath(Path[] pathsToWatch, LocalFSWatchServiceListener callback)
+    @Override
+    public void registerDirectory(Path[] dirsToWatch, LocalFSWatchServiceListener listener)
             throws IOException {
 
         WatchService watcher = FileSystems.getDefault().newWatchService();
-        for (Path pathToWatch : pathsToWatch) {
+        for (Path pathToWatch : dirsToWatch) {
             Path realDirectoryPath = pathToWatch.toRealPath();
             realDirectoryPath.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
         }
-        callback.onWatchStarted(pathsToWatch[0]);
-        watchers.put(watcher, callback);
+        listener.onWatchStarted(dirsToWatch[0]);
+        watchers.put(watcher, listener);
     }
 }