(iotdb) branch master updated: Fix consumer group response handling in topic procedures (#19047)

[jackietien]

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

JackieTien97 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 2d047dc1009 Fix consumer group response handling in topic procedures 
(#19047)
2d047dc1009 is described below

commit 2d047dc1009a457f807c01b2ccae52f0538f8c63
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Thu Oct 29 14:19:20 2026 +0800

    Fix consumer group response handling in topic procedures (#19047)
---
 .../procedure/impl/subscription/topic/AlterTopicProcedure.java   | 9 +++++++--
 .../procedure/impl/subscription/topic/DropTopicProcedure.java    | 9 +++++++--
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
index 63052a951ac..58f1c66ccee 100644
--- 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
+++ 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
@@ -42,6 +42,7 @@ import 
org.apache.iotdb.confignode.procedure.impl.subscription.SubscriptionOpera
 import 
org.apache.iotdb.confignode.procedure.impl.subscription.subscription.AbstractOperateSubscriptionAndPipeProcedure;
 import org.apache.iotdb.confignode.procedure.store.ProcedureType;
 import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
@@ -279,7 +280,9 @@ public class AlterTopicProcedure extends 
AbstractOperateSubscriptionAndPipeProce
 
     for (final ConsumerGroupMeta consumerGroupMeta : 
getUpdatedRetainedConsumerGroupMetas()) {
       final List<TSStatus> consumerGroupStatuses =
-          env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize());
+          
env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()).values().stream()
+              .map(TPushConsumerGroupMetaResp::getStatus)
+              .toList();
       if (RpcUtils.squashResponseStatusList(consumerGroupStatuses).getCode()
           != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
         throw new SubscriptionException(consumerGroupStatuses.toString());
@@ -353,7 +356,9 @@ public class AlterTopicProcedure extends 
AbstractOperateSubscriptionAndPipeProce
 
     for (final ConsumerGroupMeta consumerGroupMeta : 
retainedConsumerGroupMetas) {
       final List<TSStatus> consumerGroupStatuses =
-          env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize());
+          
env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()).values().stream()
+              .map(TPushConsumerGroupMetaResp::getStatus)
+              .toList();
       if (RpcUtils.squashResponseStatusList(consumerGroupStatuses).getCode()
           != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
         throw new SubscriptionException(consumerGroupStatuses.toString());
diff --git 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
index 5d69675f76a..c5e8a6c5fa6 100644
--- 
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
+++ 
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java
@@ -42,6 +42,7 @@ import 
org.apache.iotdb.confignode.procedure.impl.subscription.SubscriptionOpera
 import 
org.apache.iotdb.confignode.procedure.impl.subscription.subscription.AbstractOperateSubscriptionAndPipeProcedure;
 import org.apache.iotdb.confignode.procedure.store.ProcedureType;
 import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
@@ -217,7 +218,9 @@ public class DropTopicProcedure extends 
AbstractOperateSubscriptionAndPipeProced
     }
     for (final ConsumerGroupMeta consumerGroupMeta : 
getUpdatedRetainedConsumerGroupMetas()) {
       final List<TSStatus> consumerGroupStatuses =
-          env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize());
+          
env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()).values().stream()
+              .map(TPushConsumerGroupMetaResp::getStatus)
+              .toList();
       if (RpcUtils.squashResponseStatusList(consumerGroupStatuses).getCode()
           != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
         throw new SubscriptionException(consumerGroupStatuses.toString());
@@ -284,7 +287,9 @@ public class DropTopicProcedure extends 
AbstractOperateSubscriptionAndPipeProced
     if (Objects.nonNull(existedTopicMeta)) {
       for (final ConsumerGroupMeta consumerGroupMeta : 
retainedConsumerGroupMetas) {
         final List<TSStatus> statuses =
-            
env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize());
+            
env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()).values().stream()
+                .map(TPushConsumerGroupMetaResp::getStatus)
+                .toList();
         if (RpcUtils.squashResponseStatusList(statuses).getCode()
             != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
           throw new SubscriptionException(statuses.toString());