static List<Factory> discoverFactories(ClassLoader classLoader) {
        final Iterator<Factory> serviceLoaderIterator =
                ServiceLoader.load(Factory.class, classLoader).iterator();

        final List<Factory> loadResults = new ArrayList<>();
        while (true) {
            try {
                // error handling should also be applied to the hasNext() call because service
                // loading might cause problems here as well
                if (!serviceLoaderIterator.hasNext()) {
                    break;
                }

                loadResults.add(serviceLoaderIterator.next());
            } catch (Throwable t) {
                if (t instanceof NoClassDefFoundError) {
                    LOG.debug(
                            "NoClassDefFoundError when loading a "
                                    + Factory.class.getCanonicalName()
                                    + ". This is expected when trying to load a format dependency but no flink-connector-files is loaded.",
                            t);
                } else {
                    throw new TableException(
                            "Unexpected error when trying to load service provider.", t);
                }
            }
        }

        return loadResults;
    }
@Test
    void testDiscoverFactoryBadClass(@TempDir Path tempDir) throws IOException {
        // Let's prepare the classloader with a factory interface and 2 classes, one implements our
        // sub-interface of SerializationFormatFactory and the other implements only
        // SerializationFormatFactory.
        final String subInterfaceName = "MyFancySerializationSchemaFormat";
        final String subInterfaceImplementationName = "MyFancySerializationSchemaFormatImpl";
        final String serializationSchemaImplementationName = "AnotherSerializationSchema";

        final URLClassLoader classLoaderIncludingTheInterface =
                ClassLoaderUtils.withRoot(tempDir.toFile())
                        .addClass(
                                subInterfaceName,
                                "public interface "
                                        + subInterfaceName
                                        + " extends "
                                        + SerializationFormatFactory.class.getName()
                                        + " {}")
                        .addClass(
                                subInterfaceImplementationName,
                                "import org.apache.flink.api.common.serialization.SerializationSchema;"
                                        + "import org.apache.flink.configuration.ConfigOption;"
                                        + "import org.apache.flink.configuration.ReadableConfig;"
                                        + "import org.apache.flink.table.connector.format.EncodingFormat;"
                                        + "import org.apache.flink.table.data.RowData;"
                                        + "import org.apache.flink.table.factories.DynamicTableFactory;"
                                        + "import org.apache.flink.table.factories.SerializationFormatFactory;"
                                        + "import java.util.Set;"
                                        + "public class "
                                        + subInterfaceImplementationName
                                        + " implements "
                                        + subInterfaceName
                                        + " {"
                                        + "@Override public String factoryIdentifier() { return null; }"
                                        + "@Override public Set<ConfigOption<?>> requiredOptions() { return null; }"
                                        + "@Override public Set<ConfigOption<?>> optionalOptions() { return null; }"
                                        + "@Override public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) { return null; }"
                                        + "}")
                        .addClass(
                                serializationSchemaImplementationName,
                                "import org.apache.flink.api.common.serialization.SerializationSchema;"
                                        + "import org.apache.flink.configuration.ConfigOption;"
                                        + "import org.apache.flink.configuration.ReadableConfig;"
                                        + "import org.apache.flink.table.connector.format.EncodingFormat;"
                                        + "import org.apache.flink.table.data.RowData;"
                                        + "import org.apache.flink.table.factories.DynamicTableFactory;"
                                        + "import org.apache.flink.table.factories.SerializationFormatFactory;"
                                        + "import java.util.Set;"
                                        + "public class "
                                        + serializationSchemaImplementationName
                                        + " implements "
                                        + SerializationFormatFactory.class.getName()
                                        + " {"
                                        + "@Override public String factoryIdentifier() { return null; }"
                                        + "@Override public Set<ConfigOption<?>> requiredOptions() { return null; }"
                                        + "@Override public Set<ConfigOption<?>> optionalOptions() { return null; }"
                                        + "@Override public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) { return null; }"
                                        + "}")
                        .addService(Factory.class.getName(), subInterfaceImplementationName)
                        .addService(Factory.class.getName(), serializationSchemaImplementationName)
                        .build();

        // Delete the sub interface now, so it can't be loaded
        Files.delete(tempDir.resolve(subInterfaceName + ".class"));

        assertThat(FactoryUtil.discoverFactories(classLoaderIncludingTheInterface))
                .map(f -> f.getClass().getName())
                .doesNotContain(subInterfaceImplementationName)
                .contains(serializationSchemaImplementationName);
    }