Zhile opened a new issue, #4449:
URL: https://github.com/apache/flink-cdc/issues/4449
## Flink CDC version
- Flink CDC 3.6.0 (`flink-cdc-dist-3.6.0-1.20.jar`)
- Flink 1.20.x
- Deployment: Application mode via Flink Kubernetes Operator
(FlinkDeployment), checkpoint/HA on S3
## What happened
Submitting a YAML pipeline in **application mode** fails to start. The
FlinkDeployment job is configured as:
```yaml
job:
jarURI: local:///opt/flink/lib/flink-cdc-dist-3.6.0-1.20.jar
entryClass: org.apache.flink.cdc.cli.CliExecutor
args:
- /flink-cdc-resolved/pipeline.yaml # local path to the pipeline
definition file
```
JobManager fails during job graph construction with:
```
Missing required field "source" in top-level configuration.
```
The job never runs. The same `pipeline.yaml` submitted via session mode
(`flink-cdc.sh` / `CliFrontend`) works fine.
## Root cause
`CliExecutor.main(String[] args)` (the application-mode entry point) does:
```java
PipelineDef pipelineDef = pipelineDefinitionParser.parse(args[0], new
Configuration());
```
`args[0]` is the pipeline definition **file path** (the deployment executors
set
`ApplicationConfiguration.APPLICATION_ARGS = commandLine.getArgList()`, i.e.
the file path shipped
into the JobManager container). But
`YamlPipelineDefinitionParser.parse(String, Configuration)`
treats its `String` argument as the YAML **content**, not a path. So the
path string itself
(`/flink-cdc-resolved/pipeline.yaml`) is parsed as YAML, producing a
document with no `source`/`sink`
keys, which then fails validation with *Missing required field "source"*.
Session mode works because `CliFrontend` reads the file into content before
parsing.
## Expected behavior
In application mode `CliExecutor` should read the pipeline definition file
referenced by `args[0]`
and parse its **content**.
## Suggested fix
Read the file content with the **local JVM file API** (not Flink's
`FileSystem`) and pass the content
to the `String` (content) overload. Using Flink `FileSystem.get(localPath)`
is unreliable here because
the cluster default FileSystem is often S3 (checkpoints/HA on S3), so it
would not resolve the local
file shipped next to the JM.
```java
// flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java
(main)
String pipelineDefContent =
new String(Files.readAllBytes(Paths.get(args[0])),
StandardCharsets.UTF_8);
PipelineDef pipelineDef =
pipelineDefinitionParser.parse(pipelineDefContent, new
Configuration());
```
## How to reproduce
1. Build a Flink 1.20 image with `flink-cdc-dist-3.6.0-1.20.jar` in
`/opt/flink/lib`.
2. Deploy a FlinkDeployment (application mode) with
`entryClass=org.apache.flink.cdc.cli.CliExecutor`
and `args=[/path/to/pipeline.yaml]`, the pipeline file mounted/shipped
into the JM container.
3. JM fails with `Missing required field "source" in top-level
configuration`.
## Affected component
`flink-cdc-cli` — `org.apache.flink.cdc.cli.CliExecutor#main`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
