[FLINK-36504][table-common]Remove all deprecated methods ModuleFactory (

#25963)

Co-authored-by: shiwei10 <shiwei10@staff.weibo.com>

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

@@ -53,7 +53,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -427,41 +426,32 @@ public static Module createModule(
                             MODULE_TYPE.key(), options.get(MODULE_TYPE.key())));
         }
 
+        final DefaultModuleContext discoveryContext =
+                new DefaultModuleContext(options, configuration, classLoader);
         try {
-            final Map<String, String> optionsWithType = new HashMap<>(options);
-            optionsWithType.put(MODULE_TYPE.key(), moduleName);
-
-            final ModuleFactory legacyFactory =
-                    TableFactoryService.find(ModuleFactory.class, optionsWithType, classLoader);
-            return legacyFactory.createModule(optionsWithType);
-        } catch (NoMatchingTableFactoryException e) {
-            final DefaultModuleContext discoveryContext =
+            final ModuleFactory factory =
+                    discoverFactory(
+                            ((ModuleFactory.Context) discoveryContext).getClassLoader(),
+                            ModuleFactory.class,
+                            moduleName);
+
+            final DefaultModuleContext context =
                     new DefaultModuleContext(options, configuration, classLoader);
-            try {
-                final ModuleFactory factory =
-                        discoverFactory(
-                                ((ModuleFactory.Context) discoveryContext).getClassLoader(),
-                                ModuleFactory.class,
-                                moduleName);
-
-                final DefaultModuleContext context =
-                        new DefaultModuleContext(options, configuration, classLoader);
-                return factory.createModule(context);
-            } catch (Throwable t) {
-                throw new ValidationException(
-                        String.format(
-                                "Unable to create module '%s'.%n%nModule options are:%n%s",
-                                moduleName,
-                                options.entrySet().stream()
-                                        .map(
-                                                optionEntry ->
-                                                        stringifyOption(
-                                                                optionEntry.getKey(),
-                                                                optionEntry.getValue()))
-                                        .sorted()
-                                        .collect(Collectors.joining("\n"))),
-                        t);
-            }
+            return factory.createModule(context);
+        } catch (Throwable t) {
+            throw new ValidationException(
+                    String.format(
+                            "Unable to create module '%s'.%n%nModule options are:%n%s",
+                            moduleName,
+                            options.entrySet().stream()
+                                    .map(
+                                            optionEntry ->
+                                                    stringifyOption(
+                                                            optionEntry.getKey(),
+                                                            optionEntry.getValue()))
+                                    .sorted()
+                                    .collect(Collectors.joining("\n"))),
+                    t);
         }
     }
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/ModuleFactory.java

@@ -25,7 +25,6 @@
 import org.apache.flink.table.module.Module;
 import org.apache.flink.table.module.ModuleException;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -38,20 +37,7 @@
  * instead.
  */
 @PublicEvolving
-public interface ModuleFactory extends TableFactory, Factory {
-
-    /**
-     * Creates and configures a {@link Module} using the given properties.
-     *
-     * @param properties normalized properties describing a module.
-     * @return the configured module.
-     * @deprecated Use {@link #createModule(Context)} instead and implement {@link Factory} instead
-     *     of {@link TableFactory}.
-     */
-    @Deprecated
-    default Module createModule(Map<String, String> properties) {
-        throw new ModuleException("Module factories must implement createModule().");
-    }
+public interface ModuleFactory extends Factory {
 
     /** Creates and configures a {@link Module}. */
     default Module createModule(Context context) {
@@ -79,49 +65,9 @@ interface Context {
         ClassLoader getClassLoader();
     }
 
-    default String factoryIdentifier() {
-        if (requiredContext() == null || supportedProperties() == null) {
-            throw new ModuleException("Module factories must implement factoryIdentifier()");
-        }
-
-        return null;
-    }
-
-    default Set<ConfigOption<?>> requiredOptions() {
-        if (requiredContext() == null || supportedProperties() == null) {
-            throw new ModuleException("Module factories must implement requiredOptions()");
-        }
-
-        return null;
-    }
+    String factoryIdentifier();
 
-    default Set<ConfigOption<?>> optionalOptions() {
-        if (requiredContext() == null || supportedProperties() == null) {
-            throw new ModuleException("Module factories must implement optionalOptions()");
-        }
+    Set<ConfigOption<?>> requiredOptions();
 
-        return null;
-    }
-
-    // --------------------------------------------------------------------------------------------
-    // Default implementations for legacy {@link TableFactory} stack.
-    // --------------------------------------------------------------------------------------------
-
-    /**
-     * @deprecated Implement the {@link Factory} based stack instead.
-     */
-    @Deprecated
-    default Map<String, String> requiredContext() {
-        // Default implementation for modules implementing the new {@link Factory} stack instead.
-        return null;
-    }
-
-    /**
-     * @deprecated Implement the {@link Factory} based stack instead.
-     */
-    @Deprecated
-    default List<String> supportedProperties() {
-        // Default implementation for modules implementing the new {@link Factory} stack instead.
-        return null;
-    }
+    Set<ConfigOption<?>> optionalOptions();
 }

flink-table/flink-table-common/src/test/resources/META-INF/services/org.apache.flink.table.legacy.factories.TableFactory

@@ -14,4 +14,3 @@
 # limitations under the License.
 
 org.apache.flink.table.factories.TestTableSinkFactory
-org.apache.flink.table.factories.module.LegacyDummyModuleFactory

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/TableEnvironmentTest.scala

@@ -1631,12 +1631,6 @@ class TableEnvironmentTest {
     validateShowModules(("core", false))
   }
 
-  @Test
-  def testLegacyModule(): Unit = {
-    tableEnv.executeSql("LOAD MODULE LegacyModule")
-    validateShowModules(("core", true), ("LegacyModule", true))
-  }
-
   @Test
   def testExecuteSqlWithCreateDropView(): Unit = {
     createTableForTests()