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

Code before:
            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 (suggestion):
Consider adding a reverse-order test to verify that FactoryA wins the tie when it's last, confirming the behaviour is truly order-dependent.

Code after:
            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 {
        final Configuration config = new Configuration();