Merge branch 'DataLinkDC:dev' into dev

Zzm0809 · Dec 26, 2023 · 90f1188 · 90f1188
2 parents 37b5d3b + 06ff57b
commit 90f1188
Show file tree

Hide file tree

Showing 3 changed files with 62 additions and 8 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java
@@ -24,6 +24,7 @@
 import org.dinky.cluster.FlinkCluster;
 import org.dinky.cluster.FlinkClusterInfo;
 import org.dinky.data.dto.ClusterInstanceDTO;
+import org.dinky.data.exception.DinkyException;
 import org.dinky.data.model.ClusterConfiguration;
 import org.dinky.data.model.ClusterInstance;
 import org.dinky.gateway.config.GatewayConfig;
@@ -203,13 +204,17 @@ public ClusterInstance deploySessionCluster(Integer id) {
                 GatewayConfig.build(FlinkClusterConfig.create(clusterCfg.getType(), clusterCfg.getConfigJson()));
         gatewayConfig.setType(gatewayConfig.getType().getSessionType());
         GatewayResult gatewayResult = JobManager.deploySessionCluster(gatewayConfig);
-        return registersCluster(ClusterInstanceDTO.autoRegistersClusterDTO(
-                gatewayResult.getWebURL().replace("http://", ""),
-                gatewayResult.getId(),
-                clusterCfg.getName() + "_" + LocalDateTime.now(),
-                clusterCfg.getName() + LocalDateTime.now(),
-                id,
-                null));
+        if (gatewayResult.isSuccess()) {
+            Asserts.checkNullString(gatewayResult.getWebURL(), "Unable to obtain Web URL.");
+            return registersCluster(ClusterInstanceDTO.autoRegistersClusterDTO(
+                    gatewayResult.getWebURL().replace("http://", ""),
+                    gatewayResult.getId(),
+                    clusterCfg.getName() + "_" + LocalDateTime.now(),
+                    clusterCfg.getName() + LocalDateTime.now(),
+                    id,
+                    null));
+        }
+        throw new DinkyException("Deploy session cluster error: " + gatewayResult.getError());
     }
 
     /**

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
@@ -30,6 +30,10 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.jetbrains.annotations.Nullable;
 
 import com.ververica.cdc.composer.PipelineComposer;
 import com.ververica.cdc.composer.definition.PipelineDef;
@@ -84,7 +88,7 @@ public Operation create(String statement) {
 
     @Override
     public TableResult execute(Executor executor) {
-        String yamlText = statement.substring(statement.indexOf("(") + 1, statement.lastIndexOf(")"));
+        String yamlText = getPipelineConfigure(statement);
         com.ververica.cdc.common.configuration.Configuration globalPipelineConfig =
                 com.ververica.cdc.common.configuration.Configuration.fromMap(executor.getSetConfig());
         // Parse pipeline definition file
@@ -111,6 +115,16 @@ public TableResult execute(Executor executor) {
         return null;
     }
 
+    @Nullable
+    public String getPipelineConfigure(String statement) {
+        Pattern patternYaml = Pattern.compile("(?is)^EXECUTE\\s+PIPELINE\\s+WITHYAML\\s+\\((.+)\\)");
+        Matcher matcherYaml = patternYaml.matcher(statement);
+        if (matcherYaml.find()) {
+            return matcherYaml.group(1);
+        }
+        return "";
+    }
+
     public DinkyFlinkPipelineComposer createComposer(
             boolean useMiniCluster, Configuration flinkConfig, List<Path> additionalJars) {
         if (useMiniCluster) {

diff --git a/dinky-core/src/test/java/org/dinky/trans/pipeline/FlinkCDCPipelineOperationTest.java b/dinky-core/src/test/java/org/dinky/trans/pipeline/FlinkCDCPipelineOperationTest.java
@@ -0,0 +1,35 @@
+/*
+ *
+ *  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.dinky.trans.pipeline;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.Test;
+
+class FlinkCDCPipelineOperationTest {
+
+    @Test
+    void getPipelineConfigure() {
+        FlinkCDCPipelineOperation operation = new FlinkCDCPipelineOperation();
+        String configure = "EXECUTE PIPELINE withYaml (a.b=1\n a.b.c=2)";
+        String result = operation.getPipelineConfigure(configure);
+        assertEquals("a.b=1\n a.b.c=2", result);
+    }
+}