This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/topic-consumer-group-response-statuses in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b7d3010d70c012238fc203ee9e104b8cd8e975ee Author: Caideyipi <[email protected]> AuthorDate: Thu Oct 29 03:25:35 2026 +0000 Fix consumer group response handling in topic procedures --- .../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());
