[Bugfix] Abort transaction if flush failed (#188)

Signed-off-by: PengFei Li <lpengfei2016@gmail.com>
StarRocks · Mar 6, 2023 · 61e115e · 61e115e
1 parent a273ed7
commit 61e115e
Showing 1 changed file with 11 additions and 4 deletions.
diff --git a/src/main/java/com/starrocks/connector/flink/table/sink/StarRocksDynamicSinkFunctionV2.java b/src/main/java/com/starrocks/connector/flink/table/sink/StarRocksDynamicSinkFunctionV2.java
@@ -177,10 +177,16 @@ public void finish() {
 
     @Override
     public void close() {
-        sinkManager.flush();
-        StreamLoadSnapshot snapshot = sinkManager.snapshot();
-        sinkManager.abort(snapshot);
-        sinkManager.close();
+        try {
+            sinkManager.flush();
+        } catch (Exception e) {
+            log.error("Failed to flush when closing", e);
+            throw e;
+        } finally {
+            StreamLoadSnapshot snapshot = sinkManager.snapshot();
+            sinkManager.abort(snapshot);
+            sinkManager.close();
+        }
     }
 
     @Override
@@ -198,6 +204,7 @@ public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throw
             snapshotStates.clear();
             snapshotStates.add(StarrocksSnapshotState.of(snapshotMap));
         } else {
+            sinkManager.abort(snapshot);
             throw new RuntimeException("Snapshot state failed by prepare");
         }