[FLINK-38592] Add configurable property for file system factories to allow loading multiple factories packaged in the image depending on their priority for the same schema.

Author: Izeren

docs/content.zh/docs/deployment/filesystems/common.md

@@ -63,4 +63,24 @@ fs.<scheme>.limit.stream-timeout: (毫秒，0 表示无穷)
 
 连接数是按每个 TaskManager/文件系统来进行限制的。因为文件系统的创建是按照 scheme 和 authority 进行的，所以不同的 authority 具有独立的连接池，例如 `hdfs://myhdfs:50010/` 和 `hdfs://anotherhdfs:4399/` 会有单独的连接池。
 
+## File System Factory Priority
+
+When multiple `FileSystemFactory` implementations are available for the same URI scheme (for example, during a migration between file system backends), Flink resolves the conflict using a priority mechanism. The factory with the highest priority is selected.
+
+Each factory declares a default priority via its `getPriority()` method (default `0`). You can override the priority for any factory through configuration:
+
+```yaml
+fs.<scheme>.priority.<factoryClassName>: <integer>
+```
+
+Higher values indicate higher priority. When the configuration key is not set, the factory's declared priority is used.
+
+If two factories end up with the same priority, the winner depends on classloading order, which is non-deterministic. Flink logs a warning in this case. Set explicit priorities to ensure deterministic behavior.
+
+**Example:** Suppose two S3 factory implementations are on the classpath. To select the Hadoop-based factory over the Presto-based one (default priority is zero):
+
+```yaml
+fs.s3.priority.org.apache.flink.fs.s3.hadoop.S3FileSystemFactory: 1
+```
+
 {{< top >}}

docs/content.zh/docs/deployment/filesystems/overview.md

@@ -75,6 +75,8 @@ cp ./opt/flink-s3-fs-hadoop-{{< version >}}.jar ./plugins/s3-fs-hadoop/
   - 添加 Service Entry。创建文件 `META-INF/services/org.apache.flink.core.fs.FileSystemFactory`，文件中包含文件系统 Factory 类的类名。
   （更多细节请查看 [Java Service Loader docs](https://docs.oracle.com/javase/8/docs/api/java/util/ServiceLoader.html)）
 
+  - Optionally, override `getPriority()` in your factory to declare a relative priority. This is useful when multiple factories may be present for the same URI scheme (for example, during a migration between FS backends). The factory with the highest priority is selected. The default priority is `0`. See [Common Configurations]({{< ref "docs/deployment/filesystems/common" >}}#file-system-factory-priority) for details on overriding priority via configuration. For new experimental factories it is recommended to override: `getPriority()` to `return -1`. This way you provide production safe defaults for file system migrations.
+
 在插件检索时，文件系统 Factory 类会由一个专用的 Java 类加载器加载，从而避免与其他类或 Flink 组件冲突。在文件系统实例化和文件系统调用时，应使用该类加载器。
 
 <span class="label label-warning">警告</span> 实际上这表示您的实现应避免使用 `Thread.currentThread().getContextClassLoader()` 类加载器。

docs/content/docs/deployment/filesystems/common.md

@@ -69,4 +69,24 @@ Limit enforcement on a per TaskManager/file system basis.
 Because file systems creation occurs per scheme and authority, different
 authorities have independent connection pools. For example `hdfs://myhdfs:50010/` and `hdfs://anotherhdfs:4399/` will have separate pools.
 
-{{< top >}}
+## File System Factory Priority
+
+When multiple `FileSystemFactory` implementations are available for the same URI scheme (for example, during a migration between file system backends), Flink resolves the conflict using a priority mechanism. The factory with the highest priority is selected.
+
+Each factory declares a default priority via its `getPriority()` method (default `0`). You can override the priority for any factory through configuration:
+
+```yaml
+fs.<scheme>.priority.<factoryClassName>: <integer>
+```
+
+Higher values indicate higher priority. When the configuration key is not set, the factory's declared priority is used.
+
+If two factories end up with the same priority, the winner depends on classloading order, which is non-deterministic. Flink logs a warning in this case. Set explicit priorities to ensure deterministic behavior.
+
+**Example:** Suppose two S3 factory implementations are on the classpath. To select the Hadoop-based factory over the Presto-based one (default priority is zero):
+
+```yaml
+fs.s3.priority.org.apache.flink.fs.s3.hadoop.S3FileSystemFactory: 1
+```
+
+{{< top >}}

docs/content/docs/deployment/filesystems/overview.md

@@ -92,6 +92,8 @@ To add a new file system:
   - Add a service entry. Create a file `META-INF/services/org.apache.flink.core.fs.FileSystemFactory` which contains the class name of your file system factory class
   (see the [Java Service Loader docs](https://docs.oracle.com/javase/8/docs/api/java/util/ServiceLoader.html) for more details).
 
+  - Optionally, override `getPriority()` in your factory to declare a relative priority. This is useful when multiple factories may be present for the same URI scheme (for example, during a migration between FS backends). The factory with the highest priority is selected. The default priority is `0`. See [Common Configurations]({{< ref "docs/deployment/filesystems/common" >}}#file-system-factory-priority) for details on overriding priority via configuration. For new experimental factories it is recommended to override: `getPriority()` to `return -1`. This way you provide production safe defaults for file system migrations. 
+
 During plugins discovery, the file system factory class will be loaded by a dedicated Java class loader to avoid class conflicts with other plugins and Flink components.
 The same class loader should be used during file system instantiation and the file system operation calls.
 

flink-core/src/main/java/org/apache/flink/configuration/CoreOptions.java

@@ -567,4 +567,24 @@ public static ConfigOption<Long> fileSystemConnectionLimitStreamInactivityTimeou
                 .longType()
                 .defaultValue(0L);
     }
+
+    /**
+     * Explicitly resolves the conflict between multiple FileSystemFactory implementations when
+     * multiple jars are loaded for the same scheme. Primary use is to allow configuration based
+     * migration between file systems without the need to build separate images.
+     *
+     * <p>Config key pattern: {@code fs.<scheme>.priority.<factoryClassName>}
+     */
+    public static ConfigOption<Integer> fileSystemFactoryPriority(String scheme, String className) {
+        return ConfigOptions.key("fs." + scheme + ".priority." + className)
+                .intType()
+                .noDefaultValue()
+                .withDescription(
+                        "Priority for the filesystem factory '"
+                                + className
+                                + "' when multiple factories register for the '"
+                                + scheme
+                                + "' scheme. Higher priority wins. "
+                                + "When not set, the factory's declared priority is used.");
+    }
 }

flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java

@@ -35,6 +35,8 @@
 import org.apache.flink.core.plugin.PluginManager;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.TemporaryClassLoaderContext;
+import org.apache.flink.util.WrappingProxy;
+import org.apache.flink.util.WrappingProxyUtil;
 
 import org.apache.flink.shaded.guava33.com.google.common.base.Splitter;
 import org.apache.flink.shaded.guava33.com.google.common.collect.ImmutableMultimap;
@@ -297,6 +299,21 @@ private static void initializeWithoutPlugins(Configuration config)
         initialize(config, null);
     }
 
+    /**
+     * Returns a list of registered {@link FileSystemFactory FS factories}.
+     *
+     * @return a snapshot of the currently registered file system factories
+     */
+    @Internal
+    public static List<FileSystemFactory> getRegisteredFileSystemFactories() {
+        LOCK.lock();
+        try {
+            return new ArrayList<>(FS_FACTORIES.values());
+        } finally {
+            LOCK.unlock();
+        }
+    }
+
     /**
      * Initializes the shared file system settings.
      *
@@ -338,14 +355,48 @@ public static void initialize(Configuration config, @Nullable PluginManager plug
             final List<FileSystemFactory> fileSystemFactories =
                     loadFileSystemFactories(factorySuppliers);
 
+            // Track registered priorities for factory selection
+            final Map<String, Integer> registeredPriorities = new HashMap<>();
+
             // configure all file system factories
             for (FileSystemFactory factory : fileSystemFactories) {
                 factory.configure(config);
-                String scheme = factory.getScheme();
+                final String scheme = factory.getScheme();
 
                 FileSystemFactory fsf =
                         ConnectionLimitingFactory.decorateIfLimited(factory, scheme, config);
-                FS_FACTORIES.put(scheme, fsf);
+
+                final String className = resolveFactoryClassName(factory);
+                final int registeredPriority =
+                        registeredPriorities.getOrDefault(scheme, Integer.MIN_VALUE);
+                final int newPriority =
+                        config.getOptional(CoreOptions.fileSystemFactoryPriority(scheme, className))
+                                .orElse(factory.getPriority());
+
+                LOG.info(
+                        "{} filesystem factory {} for scheme '{}' "
+                                + "with priority {} (highest registered priority: {})",
+                        newPriority >= registeredPriority ? "Registering" : "Skipping",
+                        className,
+                        scheme,
+                        newPriority,
+                        registeredPriority);
+                if (newPriority >= registeredPriority) {
+                    FS_FACTORIES.put(scheme, fsf);
+                    registeredPriorities.put(scheme, newPriority);
+                    if (newPriority == registeredPriority) {
+                        LOG.warn(
+                                "Filesystem factory {} overrides a previously registered factory "
+                                        + "for scheme '{}' at the same priority {}. "
+                                        + "The winner depends on classloading order. "
+                                        + "Set fs.{}.priority.<factoryClassName> to assign "
+                                        + "explicit priorities.",
+                                className,
+                                scheme,
+                                newPriority,
+                                scheme);
+                    }
+                }
             }
 
             // configure the default (fallback) factory
@@ -946,6 +997,20 @@ private static void addAllFactoriesToList(
         }
     }
 
+    /**
+     * Resolves the real class name of a {@link FileSystemFactory}, unwrapping {@link
+     * PluginFileSystemFactory}.
+     */
+    private static String resolveFactoryClassName(FileSystemFactory factory) {
+        if (factory instanceof WrappingProxy) {
+            @SuppressWarnings("unchecked")
+            FileSystemFactory unwrapped =
+                    WrappingProxyUtil.stripProxy((WrappingProxy<FileSystemFactory>) factory);
+            return unwrapped.getClass().getName();
+        }
+        return factory.getClass().getName();
+    }
+
     /**
      * Utility loader for the Hadoop file system factory. We treat the Hadoop FS factory in a
      * special way, because we use it as a catch all for file systems schemes not supported directly

flink-core/src/main/java/org/apache/flink/core/fs/FileSystemFactory.java

@@ -47,4 +47,13 @@ public interface FileSystemFactory extends Plugin {
      * @throws IOException Thrown if the file system could not be instantiated.
      */
     FileSystem create(URI fsUri) throws IOException;
+
+    /**
+     * Returns the default priority of this factory when multiple factories compete for the same
+     * scheme. Higher priority wins. Can be overridden via config key {@code
+     * fs.<scheme>.priority.<factoryClassName>}.
+     */
+    default int getPriority() {
+        return 0;
+    }
 }

flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java

@@ -29,7 +29,8 @@
  * A wrapper around {@link FileSystemFactory} that ensures the plugin classloader is used for all
  * {@link FileSystem} operations.
  */
-public class PluginFileSystemFactory implements FileSystemFactory {
+public class PluginFileSystemFactory
+        implements FileSystemFactory, WrappingProxy<FileSystemFactory> {
     private final FileSystemFactory inner;
     private final ClassLoader loader;
 
@@ -64,6 +65,16 @@ public FileSystem create(final URI fsUri) throws IOException {
         }
     }
 
+    @Override
+    public int getPriority() {
+        return inner.getPriority();
+    }
+
+    @Override
+    public FileSystemFactory getWrappedDelegate() {
+        return inner;
+    }
+
     @Override
     public String toString() {
         return String.format("Plugin %s", inner.getClass().getName());