[INLONG-9384][Sort] Fix pulsar audit data loss when restarting (#9386)

apache · Dec 1, 2023 · 65f9214 · 65f9214
1 parent 912dce3
commit 65f9214
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 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,6 +298,16 @@ 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/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java b/...ectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java
@@ -886,6 +886,9 @@ public void snapshotState(FunctionSnapshotContext context) throws Exception {
         if (!running) {
             log.debug("snapshotState() called on closed source");
         } else {
+
+            flushAudit();
+
             unionOffsetStates.clear();
 
             PulsarFetcher<T> fetcher = this.pulsarFetcher;
@@ -925,6 +928,13 @@ public void snapshotState(FunctionSnapshotContext context) throws Exception {
         }
     }
 
+    // flush audit data first to avoid audit data loss
+    private void flushAudit() {
+        if (sourceMetricData != null) {
+            sourceMetricData.flushAuditData();
+        }
+    }
+
     @Override
     public void notifyCheckpointComplete(long checkpointId) throws Exception {
         if (!running) {