(iotdb) branch master updated: Fix async subscription commits after close (#18757)

[jiangtian]

This is an automated email from the ASF dual-hosted git repository.

jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 864321f86a7 Fix async subscription commits after close (#18757)
864321f86a7 is described below

commit 864321f86a7d71f5daef53f64d3329c30cd5347e
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Tue Sep 29 15:05:34 2026 +0800

    Fix async subscription commits after close (#18757)
---
 .../base/AbstractSubscriptionConsumer.java         | 10 ++----
 .../base/SubscriptionConsumerLifecycleTest.java    | 41 ++++++++++++++++++++++
 2 files changed, 43 insertions(+), 8 deletions(-)

diff --git 
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
 
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
index d9e7943ad80..4eecc4d028c 100644
--- 
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
+++ 
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -1752,10 +1752,7 @@ abstract class AbstractSubscriptionConsumer implements 
AutoCloseable {
     @Override
     public void run() {
       try {
-        checkIfFenced();
-        if (isClosed()) {
-          return;
-        }
+        checkIfOpened();
         ack(messages);
         callback.onComplete();
       } catch (final Exception e) {
@@ -1769,10 +1766,7 @@ abstract class AbstractSubscriptionConsumer implements 
AutoCloseable {
     SubscriptionExecutorServiceManager.submitAsyncCommitWorker(
         () -> {
           try {
-            checkIfFenced();
-            if (isClosed()) {
-              return;
-            }
+            checkIfOpened();
             ack(messages);
             future.complete(null);
           } catch (final Throwable e) {
diff --git 
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
 
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
index 8f734b24629..75f1ed78650 100644
--- 
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
+++ 
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
@@ -282,6 +282,47 @@ public class SubscriptionConsumerLifecycleTest {
     }
   }
 
+  @Test
+  public void testClosedAsyncCommitCompletesExceptionally() throws Exception {
+    final TestPullConsumer consumer = new TestPullConsumer();
+    try {
+      consumer.open();
+      consumer.close();
+
+      final CountDownLatch callbackCompleted = new CountDownLatch(1);
+      final AtomicReference<Throwable> callbackFailure = new 
AtomicReference<>();
+      consumer.commitAsync(
+          Collections.emptyList(),
+          new AsyncCommitCallback() {
+            @Override
+            public void onComplete() {
+              callbackCompleted.countDown();
+            }
+
+            @Override
+            public void onFailure(final Throwable e) {
+              callbackFailure.set(e);
+              callbackCompleted.countDown();
+            }
+          });
+
+      Assert.assertTrue(callbackCompleted.await(5, TimeUnit.SECONDS));
+      Assert.assertTrue(callbackFailure.get() instanceof 
SubscriptionException);
+
+      final CompletableFuture<Void> future = 
consumer.commitAsync(Collections.emptyList());
+      try {
+        future.get(5, TimeUnit.SECONDS);
+        Assert.fail("An async commit on a closed consumer must fail");
+      } catch (final ExecutionException expected) {
+        Assert.assertTrue(expected.getCause() instanceof 
SubscriptionException);
+      }
+
+      Assert.assertEquals(0, consumer.commitRequestCount);
+    } finally {
+      consumer.close();
+    }
+  }
+
   @Test
   public void testSyncCommitClearsAcceptedContextsFromAutoCommitBuffer() 
throws Exception {
     final TestPullConsumer consumer = new TestPullConsumer(true);