lvyanquan commented on code in PR #3093:
URL: https://github.com/apache/flink-cdc/pull/3093#discussion_r1517601928
##########
flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliExecutor.java:
##########
@@ -54,22 +62,31 @@ public CliExecutor(
}
public PipelineExecution.ExecutionInfo run() throws Exception {
- // Parse pipeline definition file
- PipelineDefinitionParser pipelineDefinitionParser = new
YamlPipelineDefinitionParser();
- PipelineDef pipelineDef =
- pipelineDefinitionParser.parse(pipelineDefPath,
globalPipelineConfig);
-
- // Create composer
- PipelineComposer composer = getComposer(flinkConfig);
-
- // Compose pipeline
- PipelineExecution execution = composer.compose(pipelineDef);
-
- // Execute the pipeline
- return execution.execute();
+ // Create Submit Executor to deployment flink cdc job Or Run Flink CDC
Job
+ boolean isDeploymentMode =
ConfigurationUtils.isDeploymentMode(commandLine);
+ if (isDeploymentMode) {
+ ComposeExecutorFactory composeExecutorFactory = new
ComposeExecutorFactory();
+ PipelineComposeExecutor composeExecutor =
+
composeExecutorFactory.getFlinkComposeExecutor(commandLine);
Review Comment:
It's better to provided static method like `createFlinkComposeExecutor`
instread of creating Factory.
##########
flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/CliFrontend.java:
##########
@@ -76,13 +74,13 @@ static CliExecutor createExecutor(CommandLine commandLine)
throws Exception {
"Missing pipeline definition file path in arguments. ");
}
- // Take the first unparsed argument as the pipeline definition file
- Path pipelineDefPath = Paths.get(unparsedArgs.get(0));
- if (!Files.exists(pipelineDefPath)) {
- throw new FileNotFoundException(
- String.format("Cannot find pipeline definition file
\"%s\"", pipelineDefPath));
+ Path pipelineDefPath = null;
+ // If Running cdc job,need to set pipeline path
+ if (!ConfigurationUtils.isDeploymentMode(commandLine)) {
+ // Take the first unparsed argument as the pipeline definition file
+ pipelineDefPath = Paths.get(unparsedArgs.get(0));
+ LOG.info("Real Path pipelineDefPath {}", pipelineDefPath);
Review Comment:
We still need to verify `Files.exists(pipelineDefPath)` in `local` and
`remote` mode.
##########
flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/utils/ConfigurationUtils.java:
##########
@@ -48,4 +52,11 @@ public static Configuration loadMapFormattedConfig(Path
configPath) throws Excep
e);
}
}
+
+ public static boolean isDeploymentMode(CommandLine commandLine) {
+ String target = commandLine.getOptionValue(TARGET);
+ return target != null
+ && !target.equalsIgnoreCase("local")
+ && !target.equalsIgnoreCase("remote");
Review Comment:
Can we add a enum to maintain these variables, add provide
`isDeploymentMode` method?
--
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]
