Repository: apache/flink
File: flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java (Java)

Code before:
            final HashMap<String, Integer> registeredPriorities = new HashMap<>();

            // configure all file system factories
            for (FileSystemFactory factory : fileSystemFactories) {
                factory.configure(config);
                final String scheme = factory.getScheme();

                FileSystemFactory fsf =
                        ConnectionLimitingFactory.decorateIfLimited(factory, scheme, config);

                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
            FALLBACK_FACTORY.configure(config);

            // also read the default file system scheme
            final String stringifiedUri = config.get(CoreOptions.DEFAULT_FILESYSTEM_SCHEME, null);
            if (stringifiedUri == null) {
                defaultScheme = null;
            } else {

Review comment (suggestion):

There is no write up for this. The Flip and Jira do not mention priority. The example in PR says "For new experimental fs factories, suggested default is -1.". So I assume that this will not be picked up unless -1 is specified in config

Please could you write up more about the motivation for this change 
- how to associate a file system with a priority
- are we likely to have more that 2 matching schemes in you use case.
- why it is be useful to have multiple priorities to justify the if `newPriority >= registeredPriority `. Rather than just targeting the required priority
- is the concept of priority different from a version? Would it not be better to just target a scheme version.
- without the config; how do we determine which file system is the highest priority - I assume this could be different for different users; this implies that one file system scheme is inherently higher priority than another.



 

Code after:
            final Map<String, Integer> registeredPriorities = new HashMap<>();

            // configure all file system factories
            for (FileSystemFactory factory : fileSystemFactories) {
                factory.configure(config);
                final String scheme = factory.getScheme();

                FileSystemFactory fsf =
                        ConnectionLimitingFactory.decorateIfLimited(factory, scheme, config);

                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
            FALLBACK_FACTORY.configure(config);

            // also read the default file system scheme
            final String stringifiedUri = config.get(CoreOptions.DEFAULT_FILESYSTEM_SCHEME, null);
            if (stringifiedUri == null) {
                defaultScheme = null;
            } else {