[FLINK-40005][cdc-cli] CliExecutor application mode reads pipeline definition from local file path

In application mode, CliExecutor.main received args[0] as the pipeline
definition file PATH but passed it to
YamlPipelineDefinitionParser.parse(String, Configuration), which treats the
String as YAML content. The path string itself was parsed as YAML and failed
validation with: Missing required field "source" in top-level configuration.

Read the file content with the local JVM file API (java.nio Files.readAllBytes)
instead of Flink's FileSystem — the cluster default FileSystem is often S3
(checkpoints/HA on S3), which cannot resolve the local file shipped next to the
JobManager — then parse the content.

Regression introduced by #3643 (FLINK-35360, Support Yarn application mode for
yaml job).

Author: Jiyong Wang

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java

@@ -35,6 +35,9 @@
 
 import org.apache.commons.cli.CommandLine;
 
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.List;
 
 /** Executor for doing the composing and submitting logic for {@link CliFrontend}. */
@@ -108,7 +111,18 @@ public PipelineExecution.ExecutionInfo deployWithNoOpComposer() throws Exception
     // The main class for running application mode
     public static void main(String[] args) throws Exception {
         PipelineDefinitionParser pipelineDefinitionParser = new YamlPipelineDefinitionParser();
-        PipelineDef pipelineDef = pipelineDefinitionParser.parse(args[0], new Configuration());
+        // args[0] is the pipeline definition FILE PATH. In application mode the deployment
+        // executors set ApplicationConfiguration.APPLICATION_ARGS = commandLine.getArgList(), i.e.
+        // the pipeline definition file path (shipped into the JobManager container). Read its
+        // content with the local JVM file API instead of Flink's FileSystem: the cluster default
+        // FileSystem may be S3, in which case FileSystem.get(localPath) would not find the local
+        // file. Then parse the content via the String overload. Without this, the String overload
+        // would treat the path itself as YAML content and fail with:
+        // Missing required field "source" in top-level configuration.
+        String pipelineDefContent =
+                new String(Files.readAllBytes(Paths.get(args[0])), StandardCharsets.UTF_8);
+        PipelineDef pipelineDef =
+                pipelineDefinitionParser.parse(pipelineDefContent, new Configuration());
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         FlinkPipelineComposer flinkPipelineComposer =
                 FlinkPipelineComposer.ofApplicationCluster(env);

flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/CliExecutorTest.java

@@ -0,0 +1,72 @@
+/*
+ * 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.cli;
+
+import org.apache.flink.cdc.cli.parser.YamlPipelineDefinitionParser;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.composer.definition.PipelineDef;
+import org.apache.flink.core.fs.Path;
+
+import org.apache.flink.shaded.guava31.com.google.common.io.Resources;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.URL;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests covering how {@link CliExecutor#main(String[])} (the application-mode entrypoint) loads the
+ * pipeline definition.
+ *
+ * <p>In application mode the deployment executors set {@code
+ * ApplicationConfiguration.APPLICATION_ARGS = commandLine.getArgList()}, so {@code args[0]} is the
+ * pipeline definition FILE PATH. It must be parsed via the {@link
+ * YamlPipelineDefinitionParser#parse(Path, Configuration)} overload (read the file). Using the
+ * {@link YamlPipelineDefinitionParser#parse(String, Configuration)} overload would treat the path
+ * string itself as YAML content and fail with {@code Missing required field "source"}.
+ */
+class CliExecutorTest {
+
+    /** The fixed behavior: parse the pipeline definition from a file path (Path overload). */
+    @Test
+    void testParsePipelineDefinitionFromFilePath() throws Exception {
+        URL resource = Resources.getResource("definitions/pipeline-definition-minimized.yaml");
+        Path pipelineDefPath = new Path(resource.toURI());
+        PipelineDef pipelineDef =
+                new YamlPipelineDefinitionParser().parse(pipelineDefPath, new Configuration());
+        assertThat(pipelineDef).isNotNull();
+        assertThat(pipelineDef.getSource()).isNotNull();
+    }
+
+    /**
+     * Reproduces the application-mode bug: passing the path STRING to the content overload makes
+     * the parser treat the path as YAML content, yielding a scalar node without a {@code source}.
+     */
+    @Test
+    void testParsingFilePathAsYamlContentFails() {
+        String pipelineDefPath = "/opt/flink/config/pipeline.yaml";
+        assertThatThrownBy(
+                        () ->
+                                new YamlPipelineDefinitionParser()
+                                        .parse(pipelineDefPath, new Configuration()))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("Missing required field \"source\"");
+    }
+}