[INLONG-9380][Sort] Audit lost when stop job immediately after checkp…

…oint
vernedeng · Dec 4, 2023 · 9e06aaf · 9e06aaf
1 parent 040569a
commit 9e06aaf
Show file tree

Hide file tree

Showing 2 changed files with 1 addition and 11 deletions.
diff --git a/...rt/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java b/...rt/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java
@@ -298,16 +298,6 @@ public void outputMetrics(long rowCountSize, long rowDataSize, long fetchDelay,
         }
     }
 
-    /**
-     * flush audit data
-     * usually call this method in close method or when checkpointing
-     */
-    public void flushAuditData() {
-        if (auditOperator != null) {
-            auditOperator.send();
-        }
-    }
-
     public void outputMetrics(long rowCountSize, long rowDataSize, long dataTime) {
         outputDefaultMetrics(rowCountSize, rowDataSize);
         if (auditOperator != null) {

diff --git a/...ectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergStreamWriter.java b/...ectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/IcebergStreamWriter.java
@@ -17,10 +17,10 @@
 
 package org.apache.inlong.sort.iceberg.sink;
 
-import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.inlong.sort.base.metric.MetricOption;
 import org.apache.inlong.sort.iceberg.utils.SinkMetadataUtils;
 
+import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.BoundedOneInput;
 import org.apache.flink.streaming.api.operators.ChainingStrategy;