[FLINK-38831] Address review comments.

Author: guoxuanlin

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java

@@ -21,7 +21,6 @@
 import org.apache.flink.cdc.common.event.SchemaChangeEventType;
 import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
-import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.utils.Preconditions;
 import org.apache.flink.cdc.common.utils.StringUtils;
 import org.apache.flink.cdc.composer.definition.ModelDef;
@@ -81,9 +80,6 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
     private static final String ROUTE_REPLACE_SYMBOL = "replace-symbol";
     private static final String ROUTE_DESCRIPTION_KEY = "description";
 
-    // Pipeline keys
-    private static final String ROUTE_MODE_KEY = "route-mode";
-
     // Transform keys
     private static final String TRANSFORM_SOURCE_TABLE_KEY = "source-table";
     private static final String TRANSFORM_PROJECTION_KEY = "projection";
@@ -137,8 +133,6 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         // it's not of plain data types and must be removed before calling toPipelineConfig.
         List<UdfDef> udfDefs = new ArrayList<>();
         final List<ModelDef> modelDefs = new ArrayList<>();
-        // Use default route mode if not specified
-        String routeMode = RouteRule.MatchMode.ALL_MATCH.getConfigValue();
         if (pipelineDefJsonNode.get(PIPELINE_KEY) != null) {
             Optional.ofNullable(
                             ((ObjectNode) pipelineDefJsonNode.get(PIPELINE_KEY)).remove(UDF_KEY))
@@ -149,13 +143,6 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
                             ((ObjectNode) pipelineDefJsonNode.get(PIPELINE_KEY)).remove(MODEL_KEY))
                     .map(node -> validateArray("model", node))
                     .ifPresent(node -> modelDefs.addAll(parseModels(node)));
-
-            // Extract route-mode from pipeline if present
-            JsonNode pipelineNode = pipelineDefJsonNode.get(PIPELINE_KEY);
-            if (pipelineNode.has(ROUTE_MODE_KEY)) {
-                routeMode = pipelineNode.get(ROUTE_MODE_KEY).asText();
-                ((ObjectNode) pipelineNode).remove(ROUTE_MODE_KEY);
-            }
         }
 
         // Pipeline configs are optional
@@ -202,14 +189,7 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         pipelineConfig.addAll(userPipelineConfig);
 
         return new PipelineDef(
-                sourceDef,
-                sinkDef,
-                routeDefs,
-                routeMode,
-                transformDefs,
-                udfDefs,
-                modelDefs,
-                pipelineConfig);
+                sourceDef, sinkDef, routeDefs, transformDefs, udfDefs, modelDefs, pipelineConfig);
     }
 
     private SourceDef toSourceDef(JsonNode sourceNode) {

flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java

@@ -725,13 +725,13 @@ void testParsingFullDefinitionFromString() throws Exception {
                                     "ods_db.ods_<>",
                                     "<>",
                                     "One-to-one mapping for other tables")),
-                    "first-match",
                     Collections.emptyList(),
                     Collections.emptyList(),
                     Collections.emptyList(),
                     Configuration.fromMap(
                             ImmutableMap.<String, String>builder()
                                     .put("name", "mysql_to_doris_with_route_match_mode")
                                     .put("parallelism", "2")
+                                    .put("route-mode", "first-match")
                                     .build()));
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java

@@ -72,6 +72,22 @@ public class PipelineOptions {
                                                             "EXCEPTION: Throw an exception to terminate the sync pipeline.")))
                                     .build());
 
+    public static final ConfigOption<String> PIPELINE_ROUTE_MODE =
+            ConfigOptions.key("route-mode")
+                    .stringType()
+                    .defaultValue(RouteMode.ALL_MATCH.toString())
+                    .withDescription(
+                            Description.builder()
+                                    .text("Match mode for routing rules. ")
+                                    .linebreak()
+                                    .add(
+                                            ListElement.list(
+                                                    text(
+                                                            "all-match: Apply all matching route rules to a table."),
+                                                    text(
+                                                            "first-match: Apply only the first matching route rule and stop evaluation.")))
+                                    .build());
+
     public static final ConfigOption<String> PIPELINE_LOCAL_TIME_ZONE =
             ConfigOptions.key("local-time-zone")
                     .stringType()

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/RouteMode.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.pipeline;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+/** Route mode for routing rules. */
+@PublicEvolving
+public enum RouteMode {
+    /** Match all applicable routing rules. */
+    ALL_MATCH("all-match"),
+    /** Match only the first applicable routing rule. */
+    FIRST_MATCH("first-match");
+
+    private final String value;
+
+    RouteMode(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public String toString() {
+        return value;
+    }
+
+    /**
+     * Parse RouteMode from string. Supports both underscore (ALL_MATCH) and hyphen (all-match)
+     * formats, case-insensitive.
+     *
+     * @param value the string value to parse
+     * @return the corresponding RouteMode
+     * @throws IllegalArgumentException if the value cannot be parsed
+     */
+    public static RouteMode fromString(String value) {
+        if (value == null) {
+            throw new IllegalArgumentException("RouteMode value cannot be null");
+        }
+        for (RouteMode mode : RouteMode.values()) {
+            if (mode.value.equalsIgnoreCase(value)) {
+                return mode;
+            }
+        }
+        throw new IllegalArgumentException(
+                String.format(
+                        "Invalid RouteMode value: '%s'. Expected one of: [all-match, first-match, ALL-MATCH, FIRST-MATCH]",
+                        value));
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/route/RouteRule.java

@@ -24,43 +24,6 @@ public class RouteRule implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
-    /** Match mode for routing rules. */
-    public enum MatchMode {
-        /** Match all applicable routing rules. */
-        ALL_MATCH("all-match"),
-        /** Match only the first applicable routing rule. */
-        FIRST_MATCH("first-match");
-
-        private final String configValue;
-
-        MatchMode(String configValue) {
-            this.configValue = configValue;
-        }
-
-        public String getConfigValue() {
-            return configValue;
-        }
-
-        /**
-         * Parse match mode from configuration value.
-         *
-         * @param value the configuration value
-         * @return the corresponding MatchMode
-         * @throws IllegalArgumentException if the value is invalid
-         */
-        public static MatchMode fromConfigValue(String value) {
-            for (MatchMode mode : values()) {
-                if (mode.configValue.equalsIgnoreCase(value)) {
-                    return mode;
-                }
-            }
-            throw new IllegalArgumentException(
-                    String.format(
-                            "Invalid match-mode value: %s. Allowed values are: %s, %s",
-                            value, FIRST_MATCH.configValue, ALL_MATCH.configValue));
-        }
-    }
-
     public RouteRule(String sourceTable, String sinkTable) {
         this(sourceTable, sinkTable, null);
     }

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/PipelineDef.java

@@ -19,8 +19,8 @@
 
 import org.apache.flink.cdc.common.annotation.VisibleForTesting;
 import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.pipeline.RouteMode;
 import org.apache.flink.cdc.common.pipeline.RuntimeExecutionMode;
-import org.apache.flink.cdc.common.route.RouteRule;
 import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
 import org.apache.flink.cdc.composer.PipelineComposer;
 import org.apache.flink.cdc.composer.PipelineExecution;
@@ -33,6 +33,7 @@
 
 import static org.apache.flink.cdc.common.pipeline.PipelineOptions.PIPELINE_EXECUTION_RUNTIME_MODE;
 import static org.apache.flink.cdc.common.pipeline.PipelineOptions.PIPELINE_LOCAL_TIME_ZONE;
+import static org.apache.flink.cdc.common.pipeline.PipelineOptions.PIPELINE_ROUTE_MODE;
 
 /**
  * Definition of a pipeline.
@@ -57,7 +58,6 @@ public class PipelineDef {
     private final SourceDef source;
     private final SinkDef sink;
     private final List<RouteDef> routes;
-    private final String routeMode;
     private final List<TransformDef> transforms;
     private final List<UdfDef> udfs;
     private final List<ModelDef> models;
@@ -67,15 +67,13 @@ public PipelineDef(
             SourceDef source,
             SinkDef sink,
             List<RouteDef> routes,
-            String routeMode,
             List<TransformDef> transforms,
             List<UdfDef> udfs,
             List<ModelDef> models,
             Configuration config) {
         this.source = source;
         this.sink = sink;
         this.routes = routes;
-        this.routeMode = routeMode;
         this.transforms = transforms;
         this.udfs = udfs;
         this.models = models;
@@ -89,34 +87,7 @@ public PipelineDef(
             List<TransformDef> transforms,
             List<UdfDef> udfs,
             Configuration config) {
-        this(
-                source,
-                sink,
-                routes,
-                RouteRule.MatchMode.ALL_MATCH.getConfigValue(),
-                transforms,
-                udfs,
-                new ArrayList<>(),
-                config);
-    }
-
-    public PipelineDef(
-            SourceDef source,
-            SinkDef sink,
-            List<RouteDef> routes,
-            List<TransformDef> transforms,
-            List<UdfDef> udfs,
-            List<ModelDef> models,
-            Configuration config) {
-        this(
-                source,
-                sink,
-                routes,
-                RouteRule.MatchMode.ALL_MATCH.getConfigValue(),
-                transforms,
-                udfs,
-                models,
-                config);
+        this(source, sink, routes, transforms, udfs, new ArrayList<>(), config);
     }
 
     public SourceDef getSource() {
@@ -131,8 +102,9 @@ public List<RouteDef> getRoute() {
         return routes;
     }
 
-    public String getRouteMode() {
-        return routeMode;
+    public RouteMode getRouteMode() {
+        String routeModeStr = config.get(PIPELINE_ROUTE_MODE);
+        return RouteMode.fromString(routeModeStr);
     }
 
     public List<TransformDef> getTransforms() {
@@ -161,7 +133,7 @@ public String toString() {
                 + ", routes="
                 + routes
                 + ", routeMode="
-                + routeMode
+                + getRouteMode()
                 + ", transforms="
                 + transforms
                 + ", udfs="
@@ -185,7 +157,6 @@ public boolean equals(Object o) {
         return Objects.equals(source, that.source)
                 && Objects.equals(sink, that.sink)
                 && Objects.equals(routes, that.routes)
-                && Objects.equals(routeMode, that.routeMode)
                 && Objects.equals(transforms, that.transforms)
                 && Objects.equals(udfs, that.udfs)
                 && Objects.equals(models, that.models)
@@ -194,7 +165,7 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(source, sink, routes, routeMode, transforms, udfs, models, config);
+        return Objects.hash(source, sink, routes, transforms, udfs, models, config);
     }
 
     // ------------------------------------------------------------------------