update log info

Signed-off-by: zenghua <[email protected]>
lakesoul-io · Aug 9, 2024 · 3fda471 · 3fda471
1 parent 651c21c
commit 3fda471
Show file tree

Hide file tree

Showing 6 changed files with 17 additions and 22 deletions.
diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/LakeSoulMultiTablesSink.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/LakeSoulMultiTablesSink.java
@@ -84,19 +84,18 @@ public Optional<SimpleVersionedSerializer<LakeSoulWriterBucketState>> getWriterS
     // StatefulGlobalTwoPhaseCommittingSinkAdapter
     @Override
     public Optional<Committer<LakeSoulMultiTableSinkCommittable>> createCommitter() throws IOException {
-        return Optional.empty();
-//        return Optional.of(new Committer<LakeSoulMultiTableSinkCommittable>() {
-//            @Override
-//            public List<LakeSoulMultiTableSinkCommittable> commit(List<LakeSoulMultiTableSinkCommittable> committables)
-//                    throws IOException, InterruptedException {
-//                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis()) + " org.apache.flink.api.connector.sink.Committer.commit: " + committables);
-//                return Collections.emptyList();
-//            }
-//
-//            @Override
-//            public void close() throws Exception {
-//            }
-//        });
+        return Optional.of(new Committer<LakeSoulMultiTableSinkCommittable>() {
+            @Override
+            public List<LakeSoulMultiTableSinkCommittable> commit(List<LakeSoulMultiTableSinkCommittable> committables)
+                    throws IOException, InterruptedException {
+                System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis()) + " org.apache.flink.api.connector.sink.Committer.commit: " + committables);
+                return Collections.emptyList();
+            }
+
+            @Override
+            public void close() throws Exception {
+            }
+        });
     }
 
     @Override

diff --git a/...l-flink/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkCommitter.java b/...l-flink/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkCommitter.java
@@ -53,7 +53,7 @@ public List<LakeSoulMultiTableSinkCommittable> commit(List<LakeSoulMultiTableSin
 
         DBManager lakeSoulDBManager = new DBManager();
         for (LakeSoulMultiTableSinkCommittable committable : committables) {
-//            LOG.info("Committing {}", committable);
+            LOG.info("Committing {}", committable);
             for (Map.Entry<String, List<InProgressFileWriter.PendingFileRecoverable>> entry : committable.getPendingFilesMap().entrySet()) {
                 List<InProgressFileWriter.PendingFileRecoverable> pendingFiles = entry.getValue();
 

diff --git a/...k/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkGlobalCommitter.java b/...k/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkGlobalCommitter.java
@@ -139,7 +139,6 @@ public List<LakeSoulMultiTableSinkGlobalCommittable> commit(
             StructType sparkSchema = ArrowUtils.fromArrowSchema(msgSchema);
 
             TableInfo tableInfo = dbManager.getTableInfoByNameAndNamespace(tableName, tableNamespace);
-//            LOG.info("Committing: {}, {}, {}, {} {}", tableNamespace, tableName, isCdc, msgSchema, tableInfo);
             if (tableInfo == null) {
                 String tableId = TABLE_ID_PREFIX + UUID.randomUUID();
                 String partition = DBUtil.formatTableInfoPartitionsField(identity.primaryKeys,
@@ -162,6 +161,7 @@ public List<LakeSoulMultiTableSinkGlobalCommittable> commit(
                 dbManager.createNewTable(tableId, tableNamespace, tableName, identity.tableLocation, msgSchema.toJson(),
                         properties, partition);
             } else {
+                LOG.info("Try to update table: {}, {}, {}, {} {}", tableNamespace, tableName, isCdc, msgSchema, tableInfo);
                 DBUtil.TablePartitionKeys partitionKeys = DBUtil.parseTableInfoPartitions(tableInfo.getPartitions());
                 if (partitionKeys.primaryKeys.size() != identity.primaryKeys.size() ||
                         !new HashSet<>(partitionKeys.primaryKeys).containsAll(identity.primaryKeys)) {

diff --git a/...n/java/org/apache/flink/lakesoul/sink/writer/arrow/LakeSoulArrowMultiTableSinkWriter.java b/...n/java/org/apache/flink/lakesoul/sink/writer/arrow/LakeSoulArrowMultiTableSinkWriter.java
@@ -176,11 +176,7 @@ public List<LakeSoulMultiTableSinkCommittable> prepareCommit(boolean flush) thro
             }
         }
         LOG.info("LakeSoulArrowMultiTableSinkWriter.prepareCommit, subTaskId={}, flush={}, {}", getSubTaskId(), flush, committables);
-
-//        if (flush) {
-//            LOG.info("printStackTrace");
-//            new Exception().printStackTrace(System.out);
-//        }
+
         return committables;
     }
 

diff --git a/.../src/main/java/org/apache/flink/lakesoul/sink/writer/arrow/LakeSoulArrowWriterBucket.java b/.../src/main/java/org/apache/flink/lakesoul/sink/writer/arrow/LakeSoulArrowWriterBucket.java
@@ -263,7 +263,7 @@ private Path assembleNewPartPath() {
     }
 
     private void closePartFile() throws IOException {
-//        LOG.info("ClosePartFile {}", inProgressPartWriter);
+        LOG.info("ClosePartFile {}", inProgressPartWriter);
         if (inProgressPartWriter != null) {
             if (inProgressPartWriter instanceof NativeLakeSoulArrowWrapperWriter) {
                 Map<String, List<InProgressFileWriter.PendingFileRecoverable>> pendingFileRecoverableMap =

diff --git a/...in/java/org/apache/flink/lakesoul/sink/writer/arrow/NativeLakeSoulArrowWrapperWriter.java b/...in/java/org/apache/flink/lakesoul/sink/writer/arrow/NativeLakeSoulArrowWrapperWriter.java
@@ -90,7 +90,7 @@ private void initNativeWriter() throws IOException {
 
         FlinkUtil.setFSConfigs(conf, nativeWriter);
         nativeWriter.initializeWriter();
-//        LOG.info("Initialized NativeLakeSoulArrowWrapperWriter: {}", this);
+        LOG.info("Initialized NativeLakeSoulArrowWrapperWriter: {}", this);
     }
 
     @Override