upgrade cdc to 3.2.0 (DataLinkDC#3878)

Co-authored-by: gaoyan1998 <[email protected]>
aiwenmo · Oct 18, 2024 · c081c3d · c081c3d
1 parent 51ee0bb
commit c081c3d
Show file tree

Hide file tree

Showing 11 changed files with 26 additions and 423 deletions.
diff --git a/dinky-core/src/main/java/org/dinky/trans/pipeline/DinkyFlinkPipelineComposer.java b/dinky-core/src/main/java/org/dinky/trans/pipeline/DinkyFlinkPipelineComposer.java
diff --git a/dinky-core/src/main/java/org/dinky/trans/pipeline/DinkyFlinkPipelineExecution.java b/dinky-core/src/main/java/org/dinky/trans/pipeline/DinkyFlinkPipelineExecution.java
diff --git a/dinky-core/src/main/java/org/dinky/trans/pipeline/FlinkCDCPipelineOperation.java b/dinky-core/src/main/java/org/dinky/trans/pipeline/FlinkCDCPipelineOperation.java
@@ -23,12 +23,16 @@
 import org.dinky.trans.AbstractOperation;
 import org.dinky.trans.Operation;
 
+import org.apache.flink.cdc.cli.parser.YamlPipelineDefinitionParser;
 import org.apache.flink.cdc.common.configuration.Configuration;
 import org.apache.flink.cdc.composer.PipelineComposer;
 import org.apache.flink.cdc.composer.definition.PipelineDef;
+import org.apache.flink.cdc.composer.flink.FlinkPipelineComposer;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.api.internal.TableResultImpl;
 
+import java.lang.reflect.Constructor;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -86,11 +90,11 @@ public Operation create(String statement) {
     public TableResult execute(Executor executor) {
         String yamlText = getPipelineConfigure(statement);
         Configuration globalPipelineConfig = Configuration.fromMap(executor.getSetConfig());
-        // Parse pipeline definition file
-        YamlTextPipelineDefinitionParser pipelineDefinitionParser = new YamlTextPipelineDefinitionParser();
-        // Create composer
-        PipelineComposer composer = createComposer(executor);
         try {
+            // Parse pipeline definition file
+            YamlPipelineDefinitionParser pipelineDefinitionParser = new YamlPipelineDefinitionParser();
+            // Create composer
+            PipelineComposer composer = createComposer(executor);
             PipelineDef pipelineDef = pipelineDefinitionParser.parse(yamlText, globalPipelineConfig);
             // Compose pipeline
             composer.compose(pipelineDef);
@@ -110,8 +114,16 @@ public String getPipelineConfigure(String statement) {
         return "";
     }
 
-    public DinkyFlinkPipelineComposer createComposer(Executor executor) {
-
-        return DinkyFlinkPipelineComposer.of(executor);
+    public PipelineComposer createComposer(Executor executor) {
+        try {
+            Class<FlinkPipelineComposer> clazz = (Class<FlinkPipelineComposer>)
+                    Class.forName("org.apache.flink.cdc.composer.flink.FlinkPipelineComposer");
+            Constructor<FlinkPipelineComposer> constructor =
+                    clazz.getDeclaredConstructor(StreamExecutionEnvironment.class, boolean.class);
+            constructor.setAccessible(true);
+            return constructor.newInstance(executor.getStreamExecutionEnvironment(), false);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
     }
 }