lvyanquan commented on code in PR #3093:
URL: https://github.com/apache/flink-cdc/pull/3093#discussion_r1517601229
##########
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/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.
--
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: commits-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
