[Fix-714][client] Catch translateToPlan exception in SQLSinkBuilder (#722)

Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>

.gitignore

@@ -28,6 +28,7 @@ target/*
 *.lst
 */target/*
 */*/target/*
+*/*/*/target/*
 dlink-web/node_modules/*
 dlink-web/node/*
 dlink-web/dist/*

dlink-catalog/dlink-catalog-mysql/dlink-catalog-mysql-1.13/target/classes/META-INF.services/org.apache.flink.table.factories.Factory

-# 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.
-
-com.dlink.flink.catalog.factory.DlinkMysqlCatalogFactory

dlink-catalog/dlink-catalog-mysql/dlink-catalog-mysql-1.15/target/classes/META-INF.services/org.apache.flink.table.factories.Factory

-# 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.
-
-com.dlink.flink.catalog.factory.DlinkMysqlCatalogFactory

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -139,11 +139,16 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
         logger.info(cdcSqlInsert);
         List<Operation> operations = customTableEnvironment.getParser().parse(cdcSqlInsert);
         logger.info("Create " + sinkTableName + " FlinkSQL insert into successful...");
-        if (operations.size() > 0) {
-            Operation operation = operations.get(0);
-            if (operation instanceof ModifyOperation) {
-                modifyOperations.add((ModifyOperation) operation);
+        try {
+            if (operations.size() > 0) {
+                Operation operation = operations.get(0);
+                if (operation instanceof ModifyOperation) {
+                    modifyOperations.add((ModifyOperation) operation);
+                }
             }
+        }catch (Exception e) {
+            logger.error("Translate to plan occur exception: {}", e);
+            throw e;
         }
     }
 

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -139,11 +139,16 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
         logger.info(cdcSqlInsert);
         List<Operation> operations = customTableEnvironment.getParser().parse(cdcSqlInsert);
         logger.info("Create " + sinkTableName + " FlinkSQL insert into successful...");
-        if (operations.size() > 0) {
-            Operation operation = operations.get(0);
-            if (operation instanceof ModifyOperation) {
-                modifyOperations.add((ModifyOperation) operation);
+        try {
+            if (operations.size() > 0) {
+                Operation operation = operations.get(0);
+                if (operation instanceof ModifyOperation) {
+                    modifyOperations.add((ModifyOperation) operation);
+                }
             }
+        }catch (Exception e) {
+            logger.error("Translate to plan occur exception: {}", e);
+            throw e;
         }
     }
 

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -140,11 +140,16 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
         logger.info(cdcSqlInsert);
         List<Operation> operations = customTableEnvironment.getParser().parse(cdcSqlInsert);
         logger.info("Create " + sinkTableName + " FlinkSQL insert into successful...");
-        if (operations.size() > 0) {
-            Operation operation = operations.get(0);
-            if (operation instanceof ModifyOperation) {
-                modifyOperations.add((ModifyOperation) operation);
+        try {
+            if (operations.size() > 0) {
+                Operation operation = operations.get(0);
+                if (operation instanceof ModifyOperation) {
+                    modifyOperations.add((ModifyOperation) operation);
+                }
             }
+        }catch (Exception e) {
+            logger.error("Translate to plan occur exception: {}", e);
+            throw e;
         }
     }
 

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -139,11 +139,16 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
         logger.info(cdcSqlInsert);
         List<Operation> operations = customTableEnvironment.getParser().parse(cdcSqlInsert);
         logger.info("Create " + sinkTableName + " FlinkSQL insert into successful...");
-        if (operations.size() > 0) {
-            Operation operation = operations.get(0);
-            if (operation instanceof ModifyOperation) {
-                modifyOperations.add((ModifyOperation) operation);
+        try {
+            if (operations.size() > 0) {
+                Operation operation = operations.get(0);
+                if (operation instanceof ModifyOperation) {
+                    modifyOperations.add((ModifyOperation) operation);
+                }
             }
+        }catch (Exception e) {
+            logger.error("Translate to plan occur exception: {}", e);
+            throw e;
         }
     }
 

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/sql/SQLSinkBuilder.java

@@ -139,11 +139,16 @@ public class SQLSinkBuilder extends AbstractSinkBuilder implements SinkBuilder,
         logger.info(cdcSqlInsert);
         List<Operation> operations = customTableEnvironment.getParser().parse(cdcSqlInsert);
         logger.info("Create " + sinkTableName + " FlinkSQL insert into successful...");
-        if (operations.size() > 0) {
-            Operation operation = operations.get(0);
-            if (operation instanceof ModifyOperation) {
-                modifyOperations.add((ModifyOperation) operation);
+        try {
+            if (operations.size() > 0) {
+                Operation operation = operations.get(0);
+                if (operation instanceof ModifyOperation) {
+                    modifyOperations.add((ModifyOperation) operation);
+                }
             }
+        }catch (Exception e) {
+            logger.error("Translate to plan occur exception: {}", e);
+            throw e;
         }
     }