(iotdb) branch master updated: Avoid empty pipe metadata sync on retained subscription unsubscribe (#19055)

[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 45a78ca18e1 Avoid empty pipe metadata sync on retained subscription 
unsubscribe (#19055)
45a78ca18e1 is described below

commit 45a78ca18e1198779ef4fe2b212aa84ce3dba624
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Fri Oct 30 10:28:08 2026 +0800

    Avoid empty pipe metadata sync on retained subscription unsubscribe (#19055)
    
    * Fix subscription unsubscribe pipe metadata sync
    
    * Strengthen unsubscribe pipe metadata regression test
---
 .../subscription/DropSubscriptionProcedure.java    | 32 ++++++++++++----------
 .../DropSubscriptionProcedureTest.java             | 19 +++++++++++++
 2 files changed, 36 insertions(+), 15 deletions(-)

diff --git 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
index a2b2b443331..f2ff6b7301b 100644
--- 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
+++ 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java
@@ -188,21 +188,23 @@ public class DropSubscriptionProcedure extends 
AbstractOperateSubscriptionAndPip
       throws SubscriptionException, IOException {
     
LOGGER.info(ProcedureMessages.DROPSUBSCRIPTIONPROCEDURE_EXECUTEFROMOPERATEONDATANODES);
 
-    // Push pipe meta to data nodes
-    final String exceptionMessage =
-        AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe(
-            null, dropMultiPipeOnDataNodes(dropPipeProcedures, env));
-    if (!exceptionMessage.isEmpty()) {
-      // throw exception instead of logging warn, do not rely on metadata 
synchronization
-      throw new SubscriptionException(
-          String.format(
-              ProcedureMessages
-                  
.FAILED_TO_DROP_PIPES_WHEN_DROPPING_SUBSCRIPTION_WITH_REQUEST_BECAUSE,
-              dropPipeProcedures.stream()
-                  .map(DropPipeProcedureV2::getPipeName)
-                  .collect(Collectors.toList()),
-              unsubscribeReq,
-              exceptionMessage));
+    if (!dropPipeProcedures.isEmpty()) {
+      // An empty Pipe metadata batch triggers full reconciliation and drops 
every local Pipe.
+      final String exceptionMessage =
+          AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe(
+              null, dropMultiPipeOnDataNodes(dropPipeProcedures, env));
+      if (!exceptionMessage.isEmpty()) {
+        // throw exception instead of logging warn, do not rely on metadata 
synchronization
+        throw new SubscriptionException(
+            String.format(
+                ProcedureMessages
+                    
.FAILED_TO_DROP_PIPES_WHEN_DROPPING_SUBSCRIPTION_WITH_REQUEST_BECAUSE,
+                dropPipeProcedures.stream()
+                    .map(DropPipeProcedureV2::getPipeName)
+                    .collect(Collectors.toList()),
+                unsubscribeReq,
+                exceptionMessage));
+      }
     }
 
     // Push consumer group meta to data nodes
diff --git 
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
 
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
index c00a7ed5f46..3dea42d1e82 100644
--- 
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
+++ 
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java
@@ -130,6 +130,25 @@ public class DropSubscriptionProcedureTest {
     }
   }
 
+  @Test
+  public void executeFromOperateOnDataNodesShouldSkipEmptyPipeMetaSync() 
throws Exception {
+    final DropSubscriptionProcedure procedure =
+        new DropSubscriptionProcedure(
+            new TUnsubscribeReq("old_consumer", "test_consumer_group", 
Collections.emptySet()));
+    final AlterConsumerGroupProcedure alterConsumerGroupProcedure =
+        Mockito.mock(AlterConsumerGroupProcedure.class);
+    procedure.setAlterConsumerGroupProcedure(alterConsumerGroupProcedure);
+    procedure.setDropPipeProcedures(Collections.emptyList());
+
+    final ConfigNodeProcedureEnv env = 
Mockito.mock(ConfigNodeProcedureEnv.class);
+
+    procedure.executeFromOperateOnDataNodes(env);
+
+    Mockito.verify(env, 
Mockito.never()).pushMultiPipeMetaToDataNodes(Mockito.anyList());
+    Mockito.verify(env, 
Mockito.never()).dropMultiPipeOnDataNodes(Mockito.anyList());
+    
Mockito.verify(alterConsumerGroupProcedure).executeFromOperateOnDataNodes(env);
+  }
+
   @Test
   public void executeFromValidateShouldResetDropPipeProceduresOnRetry() throws 
Exception {
     final Map<String, String> consumerAttributes = new HashMap<>();