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

Code before:
            config.set(CoreOptions.ALLOWED_FALLBACK_FILESYSTEMS, "s3;wasb");
            FileSystem.initialize(config, null);

            /*
            exception should be:
            org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file
            system implementation for scheme 's3'. File system schemes are supported by Flink through the following
            plugin(s): flink-s3-fs-hadoop, flink-s3-fs-presto. No file system to support this scheme could be loaded.
            Please ensure that each plugin is configured properly and resides within its own subfolder in the plugins directory.
            See https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/filesystems/plugins/ for more information. */
            assertThatThrownBy(() -> getFileSystemWithoutSafetyNet("s3://authority/"))
                    .isInstanceOf(UnsupportedFileSystemSchemeException.class)
                    .hasMessageContaining("File system schemes are supported")
                    .hasMessageContaining("flink-s3-fs-hadoop")
                    .hasMessageContaining("flink-s3-fs-presto")
                    .hasMessageContaining("Please ensure that each plugin is configured properly");
        } finally {
            FileSystem.initialize(new Configuration(), null);
        }
    }

    private static final String PRIORITY_TEST_SCHEME = "priority-test";

    static Stream<Arguments> prioritySelectionCases() {
        return Stream.of(
                Arguments.of(0, 0, DummyFsB.class),
                Arguments.of(0, 100, DummyFsB.class),
                Arguments.of(100, 0, DummyFsA.class),
                Arguments.of(-10, 0, DummyFsB.class),
                Arguments.of(50, 50, DummyFsB.class));
    }

    @ParameterizedTest
    @MethodSource("prioritySelectionCases")
    void shouldSelectFactoryByPriority(
            final int firstPriority,
            final int secondPriority,
            final Class<? extends FileSystem> expectedFsClass)
            throws Exception {
        final Configuration config = new Configuration();
        config.set(
                CoreOptions.fileSystemFactoryPriority(
                        PRIORITY_TEST_SCHEME, FactoryA.class.getName()),
                firstPriority);
        config.set(
                CoreOptions.fileSystemFactoryPriority(
                        PRIORITY_TEST_SCHEME, FactoryB.class.getName()),
                secondPriority);

        try {
            FileSystem.initialize(config, pluginManager(new FactoryA(), new FactoryB()));

Review comment (bug):
I see a silent bug here, not sure if it holds true or not . 

The (0, 0, DummyFsB) and (50, 50, DummyFsB) cases assert that DummyFsB wins on a tie. This relies on FactoryB being the last in iteration order (since the pluginManager is called with new FactoryA(), new FactoryB()). If the iteration order ever changes, these tests would break silently.



Code after:
            config.set(CoreOptions.ALLOWED_FALLBACK_FILESYSTEMS, "s3;wasb");
            FileSystem.initialize(config, null);

            /*
            exception should be:
            org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file
            system implementation for scheme 's3'. File system schemes are supported by Flink through the following
            plugin(s): flink-s3-fs-hadoop, flink-s3-fs-presto. No file system to support this scheme could be loaded.
            Please ensure that each plugin is configured properly and resides within its own subfolder in the plugins directory.
            See https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/filesystems/plugins/ for more information. */
            assertThatThrownBy(() -> getFileSystemWithoutSafetyNet("s3://authority/"))
                    .isInstanceOf(UnsupportedFileSystemSchemeException.class)
                    .hasMessageContaining("File system schemes are supported")
                    .hasMessageContaining("flink-s3-fs-hadoop")
                    .hasMessageContaining("flink-s3-fs-presto")
                    .hasMessageContaining("Please ensure that each plugin is configured properly");
        } finally {
            FileSystem.initialize(new Configuration(), null);
        }
    }

    private static final String PRIORITY_TEST_SCHEME = "priority-test";

    private static final FileSystemFactory[] ORDER_AB = {new FactoryA(), new FactoryB()};
    private static final FileSystemFactory[] ORDER_BA = {new FactoryB(), new FactoryA()};

    static Stream<Arguments> prioritySelectionCases() {
        // Each case: (priorityA, priorityB, expectedFsClass, factories).
        // Higher priority always wins. Equal priority: last-loaded factory wins.
        return Stream.of(
                Arguments.of(0, 0, DummyFsB.class, ORDER_AB),
                Arguments.of(0, 100, DummyFsB.class, ORDER_AB),
                Arguments.of(100, 0, DummyFsA.class, ORDER_AB),
                Arguments.of(-10, 0, DummyFsB.class, ORDER_AB),
                Arguments.of(50, 50, DummyFsB.class, ORDER_AB),
                // Reversed loading order — equal-priority cases flip to A
                Arguments.of(0, 0, DummyFsA.class, ORDER_BA),
                Arguments.of(0, 100, DummyFsB.class, ORDER_BA),
                Arguments.of(100, 0, DummyFsA.class, ORDER_BA),
                Arguments.of(-10, 0, DummyFsB.class, ORDER_BA),
                Arguments.of(50, 50, DummyFsA.class, ORDER_BA));
    }

    @ParameterizedTest
    @MethodSource("prioritySelectionCases")
    void shouldSelectFactoryByPriority(
            final int priorityA,
            final int priorityB,
            final Class<? extends FileSystem> expectedFsClass,
            final FileSystemFactory[] factories)
            throws Exception {