kingbase

src/main/java/com/dsk/flink/dsc/common/function/AsyncDbDataTransferFunction.java

@@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory;
 import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -166,15 +167,15 @@ public class AsyncDbDataTransferFunction extends RichAsyncFunction<JSONObject,St
         List<String> valueList = new ArrayList<>();
         List<String> updateList = new ArrayList<>();
         for (String s : columnSet) {
-            colBuilder.append(s).append(",");
+            colBuilder.append("\"").append(s).append("\",");
             valueList.add(getValueStringKingbase(dataObj, s, mysqlType.getString(s)));
-            updateList.add(s + " = EXCLUDED." + s);
+            updateList.add("\"" + s + "\" = EXCLUDED.\"" + s + "\"");
         }
         colBuilder.setLength(colBuilder.length() - 1);
         String valueString = String.join(",", valueList);
         String updateString = String.join(",", updateList);
         List<String> pkList = new ArrayList<>(pkNameSet);
-        String pkString = String.join(",", pkList);
+        String pkString = pkList.stream().map(pk -> "\"" + pk + "\"").collect(Collectors.joining(","));
         if (pkList.isEmpty()) {
             return String.format("INSERT INTO %s (%s) values (%s)", table, colBuilder.toString(), valueString);
         }

src/main/java/com/dsk/flink/dsc/common/function/AsyncDbDataTransferFunctionNew.java

@@ -16,6 +16,7 @@ import org.slf4j.LoggerFactory;
 import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -183,15 +184,15 @@ public class AsyncDbDataTransferFunctionNew extends RichAsyncFunction<JSONObject
         List<String> valueList = new ArrayList<>();
         List<String> updateList = new ArrayList<>();
         for (String s : columnSet) {
-            colBuilder.append(s).append(",");
+            colBuilder.append("\"").append(s).append("\",");
             valueList.add(getValueStringKingbase(dataObj, s, mysqlType.getString(s)));
-            updateList.add(s + " = EXCLUDED." + s);
+            updateList.add("\"" + s + "\" = EXCLUDED.\"" + s + "\"");
         }
         colBuilder.setLength(colBuilder.length() - 1);
         String valueString = String.join(",", valueList);
         String updateString = String.join(",", updateList);
         List<String> pkList = new ArrayList<>(pkNameSet);
-        String pkString = String.join(",", pkList);
+        String pkString = pkList.stream().map(pk -> "\"" + pk + "\"").collect(Collectors.joining(","));
         if (pkList.isEmpty()) {
             return String.format("INSERT INTO %s (%s) values (%s)", table, colBuilder.toString(), valueString);
         }

src/main/java/com/dsk/flink/dsc/common/function/DbDataTransferFunction.java

@@ -17,6 +17,7 @@ import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
+import java.util.stream.Collectors;
 /**
  * 重构代码
  * @author lww
@@ -151,16 +152,16 @@ public class DbDataTransferFunction extends ProcessFunction<JSONObject, Tuple3<S
         List<String> valueList = new ArrayList<>();
         List<String> updateList = new ArrayList<>();
         for (String s : columnSet) {
-            colBuilder.append(s).append(",");
+            colBuilder.append("\"").append(s).append("\",");
             String val = getValueStringKingbase(dataObj, s, mysqlType.getString(s));
             valueList.add(val);
-            updateList.add(s + " = EXCLUDED." + s);
+            updateList.add("\"" + s + "\" = EXCLUDED.\"" + s + "\"");
         }
         colBuilder.setLength(colBuilder.length() - 1);
         String valueString = String.join(",", valueList);
         String updateString = String.join(",", updateList);
         List<String> pkList = new ArrayList<>(pkNameSet);
-        String pkString = String.join(",", pkList);
+        String pkString = pkList.stream().map(pk -> "\"" + pk + "\"").collect(Collectors.joining(","));
         // 若没有主键，退化为普通 INSERT（重复会报错，由下游表约束保证）
         if (pkList.isEmpty()) {
             return String.format("INSERT INTO %s (%s) values (%s)", table, colBuilder.toString(), valueString);

src/main/java/com/dsk/flink/dsc/common/sink/DbDataSlideSink.java

@@ -44,11 +44,11 @@ public class DbDataSlideSink extends RichSinkFunction<Tuple6<String,String,Strin
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         //System.out.println(configTidbUrl);
         dataSource = new DruidDataSource();
-        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 MySQL
+        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 Kingbase
         String driverClass = envProps.getDb_driver();
         if (StrUtil.isBlank(driverClass)) {
-            driverClass = "kingbase".equalsIgnoreCase(envProps.getDb_dialect())
-                    ? "com.kingbase8.Driver" : "com.mysql.cj.jdbc.Driver";
+            driverClass = "mysql".equalsIgnoreCase(envProps.getDb_dialect())
+                    ? "com.mysql.cj.jdbc.Driver" : "com.kingbase8.Driver";
         }
         dataSource.setDriverClassName(driverClass);
         dataSource.setUsername(envProps.getDb_username());
@@ -63,9 +63,20 @@ public class DbDataSlideSink extends RichSinkFunction<Tuple6<String,String,Strin
 
     @Override
     public void close() throws Exception {
+        if (executorService != null) {
             executorService.shutdown();
+            try {
+                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
+                    executorService.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                executorService.shutdownNow();
+            }
+        }
+        if (dataSource != null && !dataSource.isClosed()) {
             dataSource.close();
         }
+    }
 
     @Override
     public void invoke(Tuple6<String,String,String,String,String,Long> value, Context context) throws Exception {

src/main/java/com/dsk/flink/dsc/common/sink/DbDataTransferSink.java

@@ -42,11 +42,11 @@ public class DbDataTransferSink extends RichSinkFunction<String> {
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         //System.out.println(configTidbUrl);
         dataSource = new DruidDataSource();
-        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 MySQL
+        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 Kingbase
         String driverClass = envProps.getDb_driver();
         if (StrUtil.isBlank(driverClass)) {
-            driverClass = "kingbase".equalsIgnoreCase(envProps.getDb_dialect())
-                    ? "com.kingbase8.Driver" : "com.mysql.cj.jdbc.Driver";
+            driverClass = "mysql".equalsIgnoreCase(envProps.getDb_dialect())
+                    ? "com.mysql.cj.jdbc.Driver" : "com.kingbase8.Driver";
         }
         dataSource.setDriverClassName(driverClass);
         dataSource.setUsername(envProps.getDb_username());
@@ -61,9 +61,20 @@ public class DbDataTransferSink extends RichSinkFunction<String> {
 
     @Override
     public void close() throws Exception {
+        if (executorService != null) {
             executorService.shutdown();
+            try {
+                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
+                    executorService.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                executorService.shutdownNow();
+            }
+        }
+        if (dataSource != null && !dataSource.isClosed()) {
             dataSource.close();
         }
+    }
 
     @Override
     public void invoke(String value, Context context) throws Exception {

src/main/java/com/dsk/flink/dsc/common/sink/DbDataTransferSinkBatch.java

@@ -52,11 +52,11 @@ public class DbDataTransferSinkBatch extends RichSinkFunction<String> {
         // 初始化获取配置
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         dataSource = new DruidDataSource();
-        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 MySQL
+        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 Kingbase
         String driverClass = envProps.getDb_driver();
         if (StrUtil.isBlank(driverClass)) {
-            driverClass = "kingbase".equalsIgnoreCase(envProps.getDb_dialect())
-                    ? "com.kingbase8.Driver" : "com.mysql.cj.jdbc.Driver";
+            driverClass = "mysql".equalsIgnoreCase(envProps.getDb_dialect())
+                    ? "com.mysql.cj.jdbc.Driver" : "com.kingbase8.Driver";
         }
         dataSource.setDriverClassName(driverClass);
         dataSource.setUsername(envProps.getDb_username());
@@ -78,10 +78,28 @@ public class DbDataTransferSinkBatch extends RichSinkFunction<String> {
 
     @Override
     public void close() throws Exception {
-        executorService.shutdown();
+        if (scheduledExecutorService != null) {
             scheduledExecutorService.shutdown();
+            try {
+                scheduledExecutorService.awaitTermination(5, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                scheduledExecutorService.shutdownNow();
+            }
+        }
+        if (executorService != null) {
+            executorService.shutdown();
+            try {
+                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
+                    executorService.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                executorService.shutdownNow();
+            }
+        }
+        if (dataSource != null && !dataSource.isClosed()) {
             dataSource.close();
         }
+    }
 
     @Override
     public void invoke(String value, Context context) throws Exception {

src/main/java/com/dsk/flink/dsc/utils/EnvProperties.java

@@ -124,7 +124,7 @@ public class EnvProperties extends Properties {
 	String log_enable;
 
 	public String getLog_enable() {
-		return logical_delete == null ? this.getProperty("logical_delete") : logical_delete;
+		return log_enable == null ? this.getProperty("log_enable") : log_enable;
 	}
 
 	public void setLog_enable(String log_enable) {

src/main/java/com/dsk/flink/dsc/utils/EtlUtils.java

@@ -15,7 +15,7 @@ public class EtlUtils {
         properties.setProperty("auto.offset.reset", "earliest");
         properties.setProperty("sasl.jaas.config", getSaslJaasConfig(username,password));
         properties.setProperty("security.protocol", "SASL_PLAINTEXT");
-        properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
+        properties.setProperty("sasl.mechanism", "SCRAM-SHA-256");
         properties.setProperty("fetch.max.bytes", "20971520"); //20M
         properties.setProperty("flink.consumer.max.fetch.size", "20971520");//20M
         properties.setProperty("session.timeout.ms", "60000");