[fix-33351][core] 修复同一个任务写入多张kudu结果表，当任务结束时，kudu结果表数据不一致问题。

Author: FlechazoW

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -40,10 +40,11 @@
 import java.security.PrivilegedAction;
 import java.sql.Timestamp;
 import java.util.Date;
+import java.util.Objects;
 
 /**
- *  @author  gituser
- *  @modify  xiuzhu
+ * @author gituser
+ * @modify xiuzhu
  */
 public class KuduOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
 
@@ -74,6 +75,8 @@ public enum WriteMode {
 
     private KuduTable table;
 
+    private AsyncKuduSession session;
+
     private Integer workerCount;
 
     private Integer defaultOperationTimeoutMs;
@@ -117,17 +120,12 @@ private void establishConnection() throws IOException {
 
         if (enableKrb) {
             UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(
-                principal,
-                keytab,
-                krb5conf
+                    principal,
+                    keytab,
+                    krb5conf
             );
             client = ugi.doAs(
-                new PrivilegedAction<AsyncKuduClient>() {
-                    @Override
-                    public AsyncKuduClient run() {
-                        return asyncKuduClientBuilder.build();
-                    }
-                });
+                    (PrivilegedAction<AsyncKuduClient>) asyncKuduClientBuilder::build);
         } else {
             client = asyncKuduClientBuilder.build();
         }
@@ -136,6 +134,7 @@ public AsyncKuduClient run() {
         if (syncClient.tableExists(tableName)) {
             table = syncClient.openTable(tableName);
         }
+        session = client.newSession();
     }
 
     @Override
@@ -147,26 +146,24 @@ public void writeRecord(Tuple2 record) throws IOException {
         }
         Row row = tupleTrans.getField(1);
         if (row.getArity() != fieldNames.length) {
-            if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0) {
+            if (outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0) {
                 LOG.error("record insert failed ..{}", row.toString());
                 LOG.error("cause by row.getArity() != fieldNames.length");
             }
             outDirtyRecords.inc();
             return;
         }
         Operation operation = toOperation(writeMode, row);
-        AsyncKuduSession session = client.newSession();
 
         try {
             if (outRecords.getCount() % ROW_PRINT_FREQUENCY == 0) {
                 LOG.info("Receive data : {}", row);
             }
 
             session.apply(operation);
-            session.close();
             outRecords.inc();
         } catch (KuduException e) {
-            if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
+            if (outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0) {
                 LOG.error("record insert failed, total dirty record:{} current row:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
@@ -176,9 +173,15 @@ public void writeRecord(Tuple2 record) throws IOException {
 
     @Override
     public void close() {
+        if (Objects.nonNull(session)) {
+            // 先把未执行完的操作执行掉，防止操作不一致
+            session.flush();
+            session.close();
+        }
+
         if (null != client) {
             try {
-                client.close();
+                client.shutdown();
             } catch (Exception e) {
                 throw new IllegalArgumentException("[closeKudu]:" + e.getMessage());
             }

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/table/KuduSinkParser.java

@@ -94,6 +94,7 @@ public Class dbTypeConvertToJavaType(String fieldType) {
             case "date":
                 return Date.class;
             case "unixtime_micros":
+            case "timestamp":
                 return Timestamp.class;
             case "decimal":
                 return BigDecimal.class;