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 ca95a046c22 Add ready DataNodes to subscription details (#19030)
ca95a046c22 is described below
commit ca95a046c2266078563f4356f2e60d6cdbcbdaa4
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 29 10:18:14 2026 +0800
Add ready DataNodes to subscription details (#19030)
---
.../iotdb/confignode/manager/ConfigManager.java | 13 +++
.../manager/subscription/SubscriptionManager.java | 6 +
.../SubscriptionReadyDataNodeTracker.java | 128 +++++++++++++++++++++
.../procedure/env/ConfigNodeProcedureEnv.java | 7 +-
.../AbstractOperateSubscriptionProcedure.java | 24 +++-
.../consumer/AlterConsumerGroupProcedure.java | 18 ++-
.../SubscriptionReadyDataNodeTrackerTest.java | 77 +++++++++++++
.../sys/subscription/ShowSubscriptionsTask.java | 7 ++
.../subscription/ShowSubscriptionsTaskTest.java | 3 +
.../schema/column/ColumnHeaderConstant.java | 4 +-
.../src/main/thrift/confignode.thrift | 1 +
11 files changed, 279 insertions(+), 9 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 360a50a24d5..3e407143e46 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -2808,6 +2808,19 @@ public class ConfigManager implements IManager {
}
});
}
+ final boolean tableModel = req.isSetIsTableModel() && req.isIsTableModel();
+ final Set<Integer> registeredDataNodeIds =
+ nodeManager.getRegisteredDataNodeLocations().keySet();
+ for (final TSubscriptionProgressInfo progressInfo : progressInfoList) {
+ progressInfo.setReadyOnDns(
+ subscriptionManager
+ .getSubscriptionReadyDataNodeTracker()
+ .getReadyDataNodeIds(
+ progressInfo.getTopicName(),
+ progressInfo.getConsumerGroupId(),
+ tableModel,
+ registeredDataNodeIds));
+ }
progressInfoList.sort(
Comparator.comparing(TSubscriptionProgressInfo::getTopicName)
.thenComparing(TSubscriptionProgressInfo::getConsumerGroupId)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionManager.java
index ff06e20cf2d..766b1c1b139 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionManager.java
@@ -29,6 +29,8 @@ public class SubscriptionManager {
private final SubscriptionCoordinator subscriptionCoordinator;
private final SubscriptionRuntimeCoordinator subscriptionRuntimeCoordinator;
private final SubscriptionLeaderChangeHandler
subscriptionLeaderChangeHandler;
+ private final SubscriptionReadyDataNodeTracker
subscriptionReadyDataNodeTracker =
+ new SubscriptionReadyDataNodeTracker();
public SubscriptionManager(ConfigManager configManager, SubscriptionInfo
subscriptionInfo) {
this.subscriptionCoordinator = new SubscriptionCoordinator(configManager,
subscriptionInfo);
@@ -48,4 +50,8 @@ public class SubscriptionManager {
public SubscriptionLeaderChangeHandler getSubscriptionLeaderChangeHandler() {
return subscriptionLeaderChangeHandler;
}
+
+ public SubscriptionReadyDataNodeTracker
getSubscriptionReadyDataNodeTracker() {
+ return subscriptionReadyDataNodeTracker;
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionReadyDataNodeTracker.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionReadyDataNodeTracker.java
new file mode 100644
index 00000000000..0dd4f765b93
--- /dev/null
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionReadyDataNodeTracker.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.manager.subscription;
+
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+public class SubscriptionReadyDataNodeTracker {
+
+ private record SubscriptionKey(String topicName, String consumerGroupId,
boolean tableModel) {}
+
+ private final Map<SubscriptionKey, Set<Integer>> readyDataNodeIds = new
HashMap<>();
+
+ public synchronized void replaceFromFullPush(
+ final Collection<ConsumerGroupMeta> consumerGroupMetas,
+ final Map<Integer, TPushConsumerGroupMetaResp> responses) {
+ readyDataNodeIds.clear();
+ for (final ConsumerGroupMeta consumerGroupMeta : consumerGroupMetas) {
+ for (final String topicName :
consumerGroupMeta.getSubscribedTopicNames()) {
+ final SubscriptionKey key =
+ new SubscriptionKey(
+ topicName,
+ consumerGroupMeta.getConsumerGroupId(),
+ consumerGroupMeta.visibleUnder(true));
+ for (final Map.Entry<Integer, TPushConsumerGroupMetaResp> response :
responses.entrySet()) {
+ if (isSuccessful(response.getValue())) {
+ readyDataNodeIds
+ .computeIfAbsent(key, ignored -> new HashSet<>())
+ .add(response.getKey());
+ }
+ }
+ }
+ }
+ }
+
+ public synchronized void recordSinglePush(
+ final ConsumerGroupMeta existingMeta,
+ final ConsumerGroupMeta updatedMeta,
+ final Map<Integer, TPushConsumerGroupMetaResp> responses) {
+ final Set<String> affectedTopics = new HashSet<>();
+ if (existingMeta != null) {
+ affectedTopics.addAll(existingMeta.getSubscribedTopicNames());
+ }
+ if (updatedMeta != null) {
+ affectedTopics.addAll(updatedMeta.getSubscribedTopicNames());
+ }
+ if (affectedTopics.isEmpty()) {
+ return;
+ }
+
+ final String consumerGroupId =
+ updatedMeta == null ? existingMeta.getConsumerGroupId() :
updatedMeta.getConsumerGroupId();
+ final boolean tableModel =
+ updatedMeta == null ? existingMeta.visibleUnder(true) :
updatedMeta.visibleUnder(true);
+ for (final String topicName : affectedTopics) {
+ final SubscriptionKey key = new SubscriptionKey(topicName,
consumerGroupId, tableModel);
+ final Set<Integer> readyIds = readyDataNodeIds.get(key);
+ if (readyIds != null) {
+ readyIds.removeAll(responses.keySet());
+ if (readyIds.isEmpty()) {
+ readyDataNodeIds.remove(key);
+ }
+ }
+ }
+
+ if (updatedMeta == null) {
+ return;
+ }
+ for (final Map.Entry<Integer, TPushConsumerGroupMetaResp> response :
responses.entrySet()) {
+ if (!isSuccessful(response.getValue())) {
+ continue;
+ }
+ for (final String topicName : updatedMeta.getSubscribedTopicNames()) {
+ readyDataNodeIds
+ .computeIfAbsent(
+ new SubscriptionKey(topicName, consumerGroupId, tableModel),
+ ignored -> new HashSet<>())
+ .add(response.getKey());
+ }
+ }
+ }
+
+ public synchronized List<Integer> getReadyDataNodeIds(
+ final String topicName,
+ final String consumerGroupId,
+ final boolean tableModel,
+ final Set<Integer> registeredDataNodeIds) {
+ final Set<Integer> readyIds =
+ new TreeSet<>(
+ readyDataNodeIds.getOrDefault(
+ new SubscriptionKey(topicName, consumerGroupId, tableModel),
Set.of()));
+ readyIds.retainAll(registeredDataNodeIds);
+ return new ArrayList<>(readyIds);
+ }
+
+ private static boolean isSuccessful(final TPushConsumerGroupMetaResp
response) {
+ return response != null
+ && response.isSetStatus()
+ && response.getStatus().getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index d1cfe162467..3d3f689bce1 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -862,7 +862,8 @@ public class ConfigNodeProcedureEnv {
return clientHandler.getResponseMap();
}
- public List<TSStatus> pushSingleConsumerGroupOnDataNode(ByteBuffer
consumerGroupMeta) {
+ public Map<Integer, TPushConsumerGroupMetaResp>
pushSingleConsumerGroupOnDataNode(
+ ByteBuffer consumerGroupMeta) {
final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
configManager.getNodeManager().getRegisteredDataNodeLocations();
final TPushSingleConsumerGroupMetaReq request =
@@ -877,9 +878,7 @@ public class ConfigNodeProcedureEnv {
final long timeoutInMs =
getRequiredSubscriptionMetadataRequestTimeoutInMs();
sendRequiredMetadataRequest(clientHandler, timeoutInMs);
fillMissingConsumerGroupPushMetaResponses(clientHandler, timeoutInMs);
- return clientHandler.getResponseList().stream()
- .map(TPushConsumerGroupMetaResp::getStatus)
- .collect(Collectors.toList());
+ return clientHandler.getResponseMap();
}
public List<TSStatus> dropSingleConsumerGroupOnDataNode(String
consumerGroupNameToDrop) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java
index d6fa0de5d20..eb49cce0be4 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/AbstractOperateSubscriptionProcedure.java
@@ -379,21 +379,41 @@ public abstract class AbstractOperateSubscriptionProcedure
protected Map<Integer, TPushConsumerGroupMetaResp>
pushConsumerGroupMetaToDataNodes(
ConfigNodeProcedureEnv env) throws IOException {
final List<ByteBuffer> consumerGroupMetaBinaryList = new ArrayList<>();
+ final List<ConsumerGroupMeta> consumerGroupMetaList = new ArrayList<>();
for (ConsumerGroupMeta consumerGroupMeta :
subscriptionInfo.get().getAllConsumerGroupMeta()) {
+ consumerGroupMetaList.add(consumerGroupMeta);
consumerGroupMetaBinaryList.add(consumerGroupMeta.serialize());
}
- return
env.pushAllConsumerGroupMetaToDataNodes(consumerGroupMetaBinaryList);
+ final Map<Integer, TPushConsumerGroupMetaResp> responseMap =
+ env.pushAllConsumerGroupMetaToDataNodes(consumerGroupMetaBinaryList);
+ recordFullConsumerGroupMetaPush(env, consumerGroupMetaList, responseMap);
+ return responseMap;
}
protected Map<Integer, TPushConsumerGroupMetaResp>
pushConsumerGroupMetaToDataNodesBestEffort(
ConfigNodeProcedureEnv env) throws IOException {
final List<ByteBuffer> consumerGroupMetaBinaryList = new ArrayList<>();
+ final List<ConsumerGroupMeta> consumerGroupMetaList = new ArrayList<>();
for (ConsumerGroupMeta consumerGroupMeta :
subscriptionInfo.get().getAllConsumerGroupMeta()) {
+ consumerGroupMetaList.add(consumerGroupMeta);
consumerGroupMetaBinaryList.add(consumerGroupMeta.serialize());
}
- return
env.pushAllConsumerGroupMetaToDataNodesBestEffort(consumerGroupMetaBinaryList);
+ final Map<Integer, TPushConsumerGroupMetaResp> responseMap =
+
env.pushAllConsumerGroupMetaToDataNodesBestEffort(consumerGroupMetaBinaryList);
+ recordFullConsumerGroupMetaPush(env, consumerGroupMetaList, responseMap);
+ return responseMap;
+ }
+
+ private void recordFullConsumerGroupMetaPush(
+ final ConfigNodeProcedureEnv env,
+ final List<ConsumerGroupMeta> consumerGroupMetaList,
+ final Map<Integer, TPushConsumerGroupMetaResp> responseMap) {
+ env.getConfigManager()
+ .getSubscriptionManager()
+ .getSubscriptionReadyDataNodeTracker()
+ .replaceFromFullPush(consumerGroupMetaList, responseMap);
}
public static boolean pushConsumerGroupMetaHasException(
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java
index 22a131f5bfb..7222b1963be 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java
@@ -31,6 +31,7 @@ import
org.apache.iotdb.confignode.procedure.impl.subscription.AbstractOperateSu
import
org.apache.iotdb.confignode.procedure.impl.subscription.SubscriptionOperation;
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;
@@ -43,6 +44,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
@@ -129,8 +131,14 @@ public class AlterConsumerGroupProcedure extends
AbstractOperateSubscriptionProc
ProcedureMessages.ALTERCONSUMERGROUPPROCEDURE_EXECUTEFROMOPERATEONDATANODES,
updatedConsumerGroupMeta.getConsumerGroupId());
- final List<TSStatus> statuses =
+ final Map<Integer, TPushConsumerGroupMetaResp> responseMap =
env.pushSingleConsumerGroupOnDataNode(updatedConsumerGroupMeta.serialize());
+ env.getConfigManager()
+ .getSubscriptionManager()
+ .getSubscriptionReadyDataNodeTracker()
+ .recordSinglePush(existingConsumerGroupMeta, updatedConsumerGroupMeta,
responseMap);
+ final List<TSStatus> statuses =
+
responseMap.values().stream().map(TPushConsumerGroupMetaResp::getStatus).toList();
if (RpcUtils.squashResponseStatusList(statuses).getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
// throw exception instead of logging warn, do not rely on metadata
synchronization
@@ -182,8 +190,14 @@ public class AlterConsumerGroupProcedure extends
AbstractOperateSubscriptionProc
throws SubscriptionException, IOException {
LOGGER.info(ProcedureMessages.ALTERCONSUMERGROUPPROCEDURE_ROLLBACKFROMOPERATEONDATANODES);
- final List<TSStatus> statuses =
+ final Map<Integer, TPushConsumerGroupMetaResp> responseMap =
env.pushSingleConsumerGroupOnDataNode(existingConsumerGroupMeta.serialize());
+ env.getConfigManager()
+ .getSubscriptionManager()
+ .getSubscriptionReadyDataNodeTracker()
+ .recordSinglePush(updatedConsumerGroupMeta, existingConsumerGroupMeta,
responseMap);
+ final List<TSStatus> statuses =
+
responseMap.values().stream().map(TPushConsumerGroupMetaResp::getStatus).toList();
if (RpcUtils.squashResponseStatusList(statuses).getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
// throw exception instead of logging warn, do not rely on metadata
synchronization
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionReadyDataNodeTrackerTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionReadyDataNodeTrackerTest.java
new file mode 100644
index 00000000000..3819c490fc8
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionReadyDataNodeTrackerTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.manager.subscription;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
+import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
+import org.apache.iotdb.mpp.rpc.thrift.TPushConsumerGroupMetaResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+
+public class SubscriptionReadyDataNodeTrackerTest {
+
+ @Test
+ public void testTracksSuccessfulFullAndSinglePushes() {
+ final SubscriptionReadyDataNodeTracker tracker = new
SubscriptionReadyDataNodeTracker();
+ final ConsumerGroupMeta group = createGroup("group");
+ group.addSubscription("consumer", Set.of("topic"));
+ final Map<Integer, TPushConsumerGroupMetaResp> fullPushResponses =
+ new HashMap<>(
+ Map.of(3, successfulResponse(), 1, successfulResponse(), 2,
failedResponse()));
+
+ tracker.replaceFromFullPush(Collections.singleton(group),
fullPushResponses);
+
+ assertEquals(
+ Arrays.asList(1, 3),
+ tracker.getReadyDataNodeIds("topic", "group", false, Set.of(1, 2, 3,
4)));
+
+ tracker.recordSinglePush(group, group, Map.of(1, failedResponse(), 2,
successfulResponse()));
+ assertEquals(
+ Arrays.asList(2, 3),
+ tracker.getReadyDataNodeIds("topic", "group", false, Set.of(1, 2, 3,
4)));
+ }
+
+ private static ConsumerGroupMeta createGroup(final String consumerGroupId) {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("consumer_group_id", consumerGroupId);
+ attributes.put("sql_dialect", "tree");
+ return new ConsumerGroupMeta(consumerGroupId, 1L, new
ConsumerMeta("consumer", 1L, attributes));
+ }
+
+ private static TPushConsumerGroupMetaResp successfulResponse() {
+ return new TPushConsumerGroupMetaResp()
+ .setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ }
+
+ private static TPushConsumerGroupMetaResp failedResponse() {
+ return new TPushConsumerGroupMetaResp()
+ .setStatus(new
TSStatus(TSStatusCode.CONSUMER_PUSH_META_ERROR.getStatusCode()));
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTask.java
index 711e85bcf19..15d5f6673f8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTask.java
@@ -37,6 +37,7 @@ import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.utils.Binary;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -134,6 +135,12 @@ public class ShowSubscriptionsTask implements IConfigTask {
builder.getColumnBuilder(17).writeLong(progressInfo.getLastPollTimeMs());
writeText(builder, 18, progressInfo.getLastConsumerId());
builder.getColumnBuilder(19).writeLong(progressInfo.getSeekGeneration());
+ writeText(
+ builder,
+ 20,
+ progressInfo.isSetReadyOnDns()
+ ? progressInfo.getReadyOnDns().toString()
+ : Collections.emptyList().toString());
builder.declarePosition();
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTaskTest.java
index b8191dca629..e3235c77e14 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTaskTest.java
@@ -69,6 +69,8 @@ public class ShowSubscriptionsTaskTest {
ColumnHeaderConstant.SUBSCRIPTION_ID,
result.getResultSetHeader().getRespColumns().get(0));
assertEquals(
ColumnHeaderConstant.SEEK_GENERATION,
result.getResultSetHeader().getRespColumns().get(19));
+ assertEquals(
+ ColumnHeaderConstant.READY_ON_DNS,
result.getResultSetHeader().getRespColumns().get(20));
assertEquals(1, resultSet.getPositionCount());
assertEquals("topic_group",
resultSet.getColumn(0).getBinary(0).toString());
assertEquals("CATCHING_UP",
resultSet.getColumn(5).getBinary(0).toString());
@@ -76,5 +78,6 @@ public class ShowSubscriptionsTaskTest {
assertEquals(5L, resultSet.getColumn(8).getLong(0));
assertEquals(20L, resultSet.getColumn(9).getLong(0));
assertEquals(4L, resultSet.getColumn(19).getLong(0));
+ assertEquals("[]", resultSet.getColumn(20).getBinary(0).toString());
}
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
index d4a46e45105..d439b356094 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
@@ -156,6 +156,7 @@ public class ColumnHeaderConstant {
public static final String LAST_POLL_TIME = "LastPollTime";
public static final String LAST_CONSUMER_ID = "LastConsumerId";
public static final String SEEK_GENERATION = "SeekGeneration";
+ public static final String READY_ON_DNS = "ready_on_dns";
// show cluster status
public static final String NODE_TYPE_CONFIG_NODE = "ConfigNode";
@@ -738,7 +739,8 @@ public class ColumnHeaderConstant {
new ColumnHeader(LAST_PROGRESS_TIME, TSDataType.INT64),
new ColumnHeader(LAST_POLL_TIME, TSDataType.INT64),
new ColumnHeader(LAST_CONSUMER_ID, TSDataType.TEXT),
- new ColumnHeader(SEEK_GENERATION, TSDataType.INT64));
+ new ColumnHeader(SEEK_GENERATION, TSDataType.INT64),
+ new ColumnHeader(READY_ON_DNS, TSDataType.TEXT));
public static final List<ColumnHeader> selectIntoColumnHeaders =
ImmutableList.of(
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index a635bf0db9b..b9a4eb4a49a 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -1113,6 +1113,7 @@ struct TSubscriptionProgressInfo {
17: required i64 lastPollTimeMs
18: required string lastConsumerId
19: required i64 seekGeneration
+ 20: optional list<i32> readyOnDns
}
struct TDropSubscriptionReq {
