[FLINK-35633] Add more YAML structure validations & provide better error messages (apache#4172)

Author: yuxiqian

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

@@ -42,6 +42,9 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -55,6 +58,8 @@
 /** Parser for converting YAML formatted pipeline definition to {@link PipelineDef}. */
 public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
 
+    private static final String TOP_LEVEL_NAME = "top-level";
+
     // Parent node keys
     private static final String SOURCE_KEY = "source";
     private static final String SINK_KEY = "sink";
@@ -118,6 +123,11 @@ public PipelineDef parse(String pipelineDefText, Configuration globalPipelineCon
 
     private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipelineConfig)
             throws Exception {
+        validateJsonNodeKeys(
+                TOP_LEVEL_NAME,
+                pipelineDefJsonNode,
+                Arrays.asList(SOURCE_KEY, SINK_KEY),
+                Arrays.asList(ROUTE_KEY, TRANSFORM_KEY, PIPELINE_KEY));
 
         // UDFs are optional. We parse UDF first and remove it from the pipelineDefJsonNode since
         // it's not of plain data types and must be removed before calling toPipelineConfig.
@@ -126,10 +136,12 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         if (pipelineDefJsonNode.get(PIPELINE_KEY) != null) {
             Optional.ofNullable(
                             ((ObjectNode) pipelineDefJsonNode.get(PIPELINE_KEY)).remove(UDF_KEY))
+                    .map(node -> validateArray("UDF", node))
                     .ifPresent(node -> node.forEach(udf -> udfDefs.add(toUdfDef(udf))));
 
             Optional.ofNullable(
                             ((ObjectNode) pipelineDefJsonNode.get(PIPELINE_KEY)).remove(MODEL_KEY))
+                    .map(node -> validateArray("model", node))
                     .ifPresent(node -> modelDefs.addAll(parseModels(node)));
         }
 
@@ -159,6 +171,7 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         // Transforms are optional
         List<TransformDef> transformDefs = new ArrayList<>();
         Optional.ofNullable(pipelineDefJsonNode.get(TRANSFORM_KEY))
+                .map(node -> validateArray("transform", node))
                 .ifPresent(
                         node ->
                                 node.forEach(
@@ -167,6 +180,7 @@ private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipe
         // Routes are optional
         List<RouteDef> routeDefs = new ArrayList<>();
         Optional.ofNullable(pipelineDefJsonNode.get(ROUTE_KEY))
+                .map(node -> validateArray("route", node))
                 .ifPresent(node -> node.forEach(route -> routeDefs.add(toRouteDef(route))));
 
         // Merge user config into global config
@@ -247,6 +261,12 @@ private SinkDef toSinkDef(JsonNode sinkNode, SchemaChangeBehavior schemaChangeBe
     }
 
     private RouteDef toRouteDef(JsonNode routeNode) {
+        validateJsonNodeKeys(
+                "route",
+                routeNode,
+                Arrays.asList(ROUTE_SOURCE_TABLE_KEY, ROUTE_SINK_TABLE_KEY),
+                Arrays.asList(ROUTE_REPLACE_SYMBOL, ROUTE_DESCRIPTION_KEY));
+
         String sourceTable =
                 checkNotNull(
                                 routeNode.get(ROUTE_SOURCE_TABLE_KEY),
@@ -271,6 +291,12 @@ private RouteDef toRouteDef(JsonNode routeNode) {
     }
 
     private UdfDef toUdfDef(JsonNode udfNode) {
+        validateJsonNodeKeys(
+                "UDF",
+                udfNode,
+                Arrays.asList(UDF_FUNCTION_NAME_KEY, UDF_CLASSPATH_KEY),
+                Collections.emptyList());
+
         String functionName =
                 checkNotNull(
                                 udfNode.get(UDF_FUNCTION_NAME_KEY),
@@ -288,6 +314,19 @@ private UdfDef toUdfDef(JsonNode udfNode) {
     }
 
     private TransformDef toTransformDef(JsonNode transformNode) {
+        validateJsonNodeKeys(
+                "transform",
+                transformNode,
+                Collections.singletonList(TRANSFORM_SOURCE_TABLE_KEY),
+                Arrays.asList(
+                        TRANSFORM_PROJECTION_KEY,
+                        TRANSFORM_FILTER_KEY,
+                        TRANSFORM_PRIMARY_KEY_KEY,
+                        TRANSFORM_PARTITION_KEY_KEY,
+                        TRANSFORM_TABLE_OPTION_KEY,
+                        TRANSFORM_DESCRIPTION_KEY,
+                        TRANSFORM_CONVERTER_AFTER_TRANSFORM_KEY));
+
         String sourceTable =
                 checkNotNull(
                                 transformNode.get(TRANSFORM_SOURCE_TABLE_KEY),
@@ -377,4 +416,57 @@ private ModelDef convertJsonNodeToModelDef(JsonNode modelNode) {
         Map<String, String> properties = mapper.convertValue(modelNode, Map.class);
         return new ModelDef(name, model, properties);
     }
+
+    private void validateJsonNodeKeys(
+            String contextName,
+            JsonNode jsonNode,
+            List<String> requiredKeys,
+            List<String> optionalKeys)
+            throws IllegalArgumentException {
+        List<String> validKeys = new ArrayList<>(requiredKeys);
+        Set<String> presentedKeys = new HashSet<>();
+        validKeys.addAll(optionalKeys);
+
+        for (Iterator<String> it = jsonNode.fieldNames(); it.hasNext(); ) {
+            String key = it.next();
+            presentedKeys.add(key);
+            if (!validKeys.contains(key)) {
+                if (TOP_LEVEL_NAME.equals(contextName)) {
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "Unexpected key `%s` in YAML top-level block.\n"
+                                            + "Allowed keys in this context are: %s\n"
+                                            + "Note: Flink configurations should be defined in \"Runtime Configurations\" instead of YAML scripts.",
+                                    key, validKeys));
+                } else {
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "Unexpected key `%s` in YAML %s block.\n"
+                                            + "Allowed keys in this context are: %s\n"
+                                            + "Note: option %s: %s is unexpected. It was silently ignored in previous versions, and probably should be removed.",
+                                    key, contextName, validKeys, key, jsonNode.get(key)));
+                }
+            }
+        }
+
+        for (String key : requiredKeys) {
+            if (!presentedKeys.contains(key)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Missing required field \"%s\" in %s configuration",
+                                key, contextName));
+            }
+        }
+    }
+
+    private JsonNode validateArray(String contextName, JsonNode jsonNode) {
+        if (jsonNode.isArray()) {
+            return jsonNode;
+        } else {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "YAML %s block is expecting an array children, but got an %s (%s). Perhaps you missed a dash prefix `-`?",
+                            contextName, jsonNode.getNodeType(), jsonNode));
+        }
+    }
 }

flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml

@@ -60,8 +60,8 @@ pipeline:
   schema-operator.rpc-timeout: 1 h
   execution.runtime-mode: STREAMING
   model:
-    model-name: GET_EMBEDDING
-    class-name: OpenAIEmbeddingModel
-    openai.model: text-embedding-3-small
-    openai.host: https://xxxx
-    openai.apikey: abcd1234
+    - model-name: GET_EMBEDDING
+      class-name: OpenAIEmbeddingModel
+      openai.model: text-embedding-3-small
+      openai.host: https://xxxx
+      openai.apikey: abcd1234

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java

@@ -183,7 +183,7 @@ void testMultipleTransformRule(boolean batchMode) throws Exception {
                                 + "\n"
                                 + "sink:\n"
                                 + "  type: values\n"
-                                + "route:\n"
+                                + "\n"
                                 + "transform:\n"
                                 + "  - source-table: %s.\\.*\n"
                                 + "    projection: ID, VERSION, 'Type-A' AS CATEGORY\n"

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/migration/YamlJobMigrationITCase.java

@@ -217,7 +217,7 @@ void testStartingJobFromSavepointWithSchemaChange(TarballFetcher.CdcVersion migr
                                 + "\n"
                                 + "pipeline:\n"
                                 + "  parallelism: %d\n"
-                                + "use.legacy.json.format: true\n",
+                                + "  use.legacy.json.format: true\n",
                         INTER_CONTAINER_MYSQL_ALIAS,
                         MySqlContainer.MYSQL_PORT,
                         MYSQL_TEST_USER,

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/CheckStep.java

@@ -0,0 +1,31 @@
+/*
+ * 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.pipeline.tests.specs;
+
+import java.util.List;
+
+/** A {@link SpecStep} to check expected output results. */
+class CheckStep implements SpecStep {
+    public List<String> jmLogs;
+    public List<String> tmLogs;
+
+    public CheckStep(List<String> jmLogs, List<String> tmLogs) {
+        this.jmLogs = jmLogs;
+        this.tmLogs = tmLogs;
+    }
+}

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/specs/ExecStep.java

@@ -0,0 +1,27 @@
+/*
+ * 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.pipeline.tests.specs;
+
+/** A {@link SpecStep} to execute SQL commands. */
+class ExecStep implements SpecStep {
+    public String sql;
+
+    public ExecStep(String sql) {
+        this.sql = sql;
+    }
+}