factor out defer logic to avoid further bugs

Signed-off-by: tison <[email protected]>
HQebupt · Sep 27, 2022 · f7f4634 · f7f4634
1 parent 747086f
commit f7f4634
Showing 1 changed file with 8 additions and 4 deletions.
diff --git a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
@@ -240,8 +240,15 @@ public void scheduleNow() {
         @Override
         public void run() {
             phaser.register();
-            if (closed) {
+            try {
+                internalRun();
+            } finally {
                 phaser.arriveAndDeregister();
+            }
+        }
+
+        private void internalRun() {
+            if (closed) {
                 return;
             }
             // if not in flushing state, do flush, else return;
@@ -250,7 +257,6 @@ public void run() {
                     log.debug("Starting flush, queue size: {}", incomingList.size());
                 }
                 if (!swapList.isEmpty()) {
-                    phaser.arriveAndDeregister();
                     throw new IllegalStateException("swapList should be empty since last flush. swapList.size: "
                             + swapList.size());
                 }
@@ -303,7 +309,6 @@ public void run() {
                             connection.rollback();
                         }
                     } catch (Exception ex) {
-                        phaser.arriveAndDeregister();
                         throw new RuntimeException(ex);
                     }
                 }
@@ -323,7 +328,6 @@ public void run() {
                     log.debug("Already in flushing state, will not flush, queue size: {}", incomingList.size());
                 }
             }
-            phaser.arriveAndDeregister();
         }
 
         @Override