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 89d4fbbcd9c Add consensus subscription progress observability (#18521)
89d4fbbcd9c is described below
commit 89d4fbbcd9c0af99b1a12cd8721276e8c24985b5
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 26 10:44:46 2026 +0800
Add consensus subscription progress observability (#18521)
---
.../org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4 | 2 +-
.../iotdb/confignode/manager/ConfigManager.java | 111 +++++++-
.../impl/DataNodeInternalRPCServiceImpl.java | 8 +-
.../common/header/DatasetHeaderFactory.java | 4 +
.../config/executor/ClusterConfigTaskExecutor.java | 19 +-
.../sys/subscription/ShowSubscriptionsTask.java | 50 ++++
.../db/queryengine/plan/parser/ASTVisitor.java | 1 +
.../plan/relational/sql/ast/ShowSubscriptions.java | 16 +-
.../plan/relational/sql/parser/AstBuilder.java | 2 +-
.../relational/sql/util/DataNodeSqlFormatter.java | 8 +-
.../subscription/ShowSubscriptionsStatement.java | 9 +
.../agent/SubscriptionBrokerAgent.java | 16 ++
.../broker/ConsensusSubscriptionBroker.java | 16 ++
.../subscription/broker/ISubscriptionBroker.java | 5 +
.../consensus/ConsensusPrefetchingQueue.java | 112 +++++++++
...nsensusSubscriptionPrefetchingQueueMetrics.java | 98 ++++++++
.../subscription/ShowSubscriptionsTaskTest.java | 80 ++++++
.../parser/ShowSubscriptionsStatementTest.java | 48 ++++
.../plan/relational/sql/ShowSubscriptionsTest.java | 64 +++++
...susSubscriptionPrefetchingQueueMetricsTest.java | 20 ++
.../schema/column/ColumnHeaderConstant.java | 36 +++
.../iotdb/commons/service/metric/enums/Metric.java | 11 +
.../consumer/SubscriptionProgressSnapshot.java | 278 +++++++++++++++++++++
.../consumer/SubscriptionProgressSnapshotTest.java | 78 ++++++
.../db/relational/grammar/sql/RelationalSql.g4 | 2 +-
.../src/main/thrift/confignode.thrift | 24 ++
.../src/main/thrift/datanode.thrift | 1 +
27 files changed, 1101 insertions(+), 18 deletions(-)
diff --git
a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
index 7bf79721e8f..5a3198d1aab 100644
---
a/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
+++
b/iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4
@@ -742,7 +742,7 @@ showTopics
;
showSubscriptions
- : SHOW SUBSCRIPTIONS (ON topicName=identifier)?
+ : SHOW SUBSCRIPTIONS (DETAILS)? (ON topicName=identifier)?
;
dropSubscription
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 7ff21974540..f86681f2bbe 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
@@ -68,6 +68,7 @@ import
org.apache.iotdb.commons.schema.tree.AlterTimeSeriesOperationType;
import org.apache.iotdb.commons.schema.ttl.TTLCache;
import org.apache.iotdb.commons.service.metric.MetricService;
import
org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper;
+import
org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot;
import org.apache.iotdb.commons.utils.AuthUtils;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.commons.utils.StatusUtils;
@@ -256,6 +257,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp;
import org.apache.iotdb.confignode.rpc.thrift.TStartPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TStopPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TSubscriptionProgressInfo;
import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq;
@@ -264,6 +266,7 @@ import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.schemaengine.template.TemplateAlterOperationType;
import
org.apache.iotdb.db.schemaengine.template.alter.TemplateAlterOperationUtil;
+import org.apache.iotdb.mpp.rpc.thrift.TPullCommitProgressResp;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
@@ -2676,9 +2679,111 @@ public class ConfigManager implements IManager {
@Override
public TShowSubscriptionResp showSubscription(TShowSubscriptionReq req) {
TSStatus status = confirmLeader();
- return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
- ?
subscriptionManager.getSubscriptionCoordinator().showSubscription(req)
- : new TShowSubscriptionResp().setStatus(status);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return new TShowSubscriptionResp().setStatus(status);
+ }
+ final TShowSubscriptionResp response =
+ subscriptionManager.getSubscriptionCoordinator().showSubscription(req);
+ if (!req.isSetDetails()
+ || !req.isDetails()
+ || response.getStatus().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return response;
+ }
+
+ final Set<String> visibleSubscriptions = new HashSet<>();
+ final Set<String> subscriptionsWithProgress = new HashSet<>();
+ if (response.isSetSubscriptionInfoList()) {
+ response
+ .getSubscriptionInfoList()
+ .forEach(
+ info ->
+ visibleSubscriptions.add(
+ info.getTopicName() + "\u0000" +
info.getConsumerGroupId()));
+ }
+ final List<TSubscriptionProgressInfo> progressInfoList = new ArrayList<>();
+ final Map<Integer, TPullCommitProgressResp> dataNodeResponses =
+
getProcedureManager().getEnv().pullCommitProgressFromDataNodesBestEffort();
+ for (final Map.Entry<Integer, TPullCommitProgressResp> dataNodeEntry :
+ dataNodeResponses.entrySet()) {
+ final TPullCommitProgressResp dataNodeResponse =
dataNodeEntry.getValue();
+ if (!dataNodeResponse.isSetSubscriptionProgress()) {
+ continue;
+ }
+ for (final ByteBuffer serializedSnapshot :
+ dataNodeResponse.getSubscriptionProgress().values()) {
+ final SubscriptionProgressSnapshot snapshot;
+ try {
+ snapshot =
SubscriptionProgressSnapshot.deserialize(serializedSnapshot);
+ } catch (final RuntimeException ignored) {
+ // A rolling upgrade may return a snapshot encoded by a different
software version.
+ continue;
+ }
+ if (!visibleSubscriptions.contains(
+ snapshot.getTopicName() + "\u0000" +
snapshot.getConsumerGroupId())) {
+ continue;
+ }
+ subscriptionsWithProgress.add(
+ snapshot.getTopicName() + "\u0000" +
snapshot.getConsumerGroupId());
+ progressInfoList.add(
+ new TSubscriptionProgressInfo(
+ snapshot.getTopicName(),
+ snapshot.getConsumerGroupId(),
+ snapshot.getRegionId(),
+ dataNodeEntry.getKey(),
+ snapshot.isActive(),
+ snapshot.isInitialized(),
+ snapshot.getStatus(),
+ snapshot.getRemainingEventCount(),
+ snapshot.getRawWalGap(),
+ snapshot.getApproximateLag(),
+ snapshot.getInFlightEventCount(),
+ snapshot.getPrefetchedEventCount(),
+ snapshot.getPendingEventCount(),
+ snapshot.getCurrentWalSearchIndex(),
+ snapshot.getNextReadSearchIndex(),
+ snapshot.getLastProgressTimeMs(),
+ snapshot.getLastPollTimeMs(),
+ snapshot.getLastConsumerId(),
+ snapshot.getSeekGeneration()));
+ }
+ }
+ if (response.isSetSubscriptionInfoList()) {
+ response
+ .getSubscriptionInfoList()
+ .forEach(
+ info -> {
+ final String key = info.getTopicName() + "\u0000" +
info.getConsumerGroupId();
+ if (!subscriptionsWithProgress.contains(key)) {
+ progressInfoList.add(
+ new TSubscriptionProgressInfo(
+ info.getTopicName(),
+ info.getConsumerGroupId(),
+ "",
+ -1,
+ false,
+ false,
+ SubscriptionProgressSnapshot.STATUS_NO_QUEUE,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ "",
+ 0L));
+ }
+ });
+ }
+ progressInfoList.sort(
+ Comparator.comparing(TSubscriptionProgressInfo::getTopicName)
+ .thenComparing(TSubscriptionProgressInfo::getConsumerGroupId)
+ .thenComparingInt(TSubscriptionProgressInfo::getDataNodeId)
+ .thenComparing(TSubscriptionProgressInfo::getRegionId));
+ return response.setSubscriptionProgressList(progressInfoList);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 8685df8f4be..34ef951296c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -1637,15 +1637,19 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
public TPullCommitProgressResp pullCommitProgress(TPullCommitProgressReq
req) {
if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
return new
TPullCommitProgressResp(RpcUtils.getStatus(TSStatusCode.UNSUPPORTED_OPERATION))
- .setCommitRegionProgress(Collections.emptyMap());
+ .setCommitRegionProgress(Collections.emptyMap())
+ .setSubscriptionProgress(Collections.emptyMap());
}
try {
final int dataNodeId =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
final Map<String, ByteBuffer> regionProgress =
SubscriptionAgent.broker().collectAllRegionCommitProgress(dataNodeId);
+ final Map<String, ByteBuffer> subscriptionProgress =
+ SubscriptionAgent.broker().collectAllProgressSnapshots();
return new TPullCommitProgressResp(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))
- .setCommitRegionProgress(regionProgress);
+ .setCommitRegionProgress(regionProgress)
+ .setSubscriptionProgress(subscriptionProgress);
} catch (Exception e) {
LOGGER.warn(
DataNodeMiscMessages.MISC_LOG_ERROR_OCCURRED_WHEN_PULLING_COMMIT_PROGRESS_48C12E4B,
e);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
index e674199df52..1c83be23650 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
@@ -182,6 +182,10 @@ public class DatasetHeaderFactory {
return new
DatasetHeader(ColumnHeaderConstant.showSubscriptionColumnHeaders, true);
}
+ public static DatasetHeader getShowSubscriptionDetailsHeader() {
+ return new
DatasetHeader(ColumnHeaderConstant.showSubscriptionDetailsColumnHeaders, true);
+ }
+
public static DatasetHeader getGetRegionIdHeader() {
return new DatasetHeader(ColumnHeaderConstant.getRegionIdColumnHeaders,
true);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 65d4b3f102d..3ddd2531200 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -3085,6 +3085,7 @@ public class ClusterConfigTaskExecutor implements
IConfigTaskExecutor {
showSubscriptionReq.setTopicName(showSubscriptionsStatement.getTopicName());
}
showSubscriptionReq.setIsTableModel(showSubscriptionsStatement.isTableModel());
+ showSubscriptionReq.setDetails(showSubscriptionsStatement.isDetails());
final TShowSubscriptionResp showSubscriptionResp =
configNodeClient.showSubscription(showSubscriptionReq);
@@ -3097,11 +3098,19 @@ public class ClusterConfigTaskExecutor implements
IConfigTaskExecutor {
return future;
}
- ShowSubscriptionsTask.buildTSBlock(
- showSubscriptionResp.isSetSubscriptionInfoList()
- ? showSubscriptionResp.getSubscriptionInfoList()
- : Collections.emptyList(),
- future);
+ if (showSubscriptionsStatement.isDetails()) {
+ ShowSubscriptionsTask.buildDetailsTSBlock(
+ showSubscriptionResp.isSetSubscriptionProgressList()
+ ? showSubscriptionResp.getSubscriptionProgressList()
+ : Collections.emptyList(),
+ future);
+ } else {
+ ShowSubscriptionsTask.buildTSBlock(
+ showSubscriptionResp.isSetSubscriptionInfoList()
+ ? showSubscriptionResp.getSubscriptionInfoList()
+ : Collections.emptyList(),
+ future);
+ }
} catch (final Exception e) {
future.setException(e);
}
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 87e07063629..711e85bcf19 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
@@ -22,6 +22,7 @@ package
org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription;
import org.apache.iotdb.commons.schema.column.ColumnHeader;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TSubscriptionProgressInfo;
import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
@@ -50,6 +51,7 @@ public class ShowSubscriptionsTask implements IConfigTask {
public ShowSubscriptionsTask(final ShowSubscriptions showSubscriptions) {
this.showSubscriptionsStatement = new ShowSubscriptionsStatement();
this.showSubscriptionsStatement.setTopicName(showSubscriptions.getTopicName());
+ this.showSubscriptionsStatement.setDetails(showSubscriptions.isDetails());
this.showSubscriptionsStatement.setTableModel(true);
}
@@ -100,4 +102,52 @@ public class ShowSubscriptionsTask implements IConfigTask {
builder.build(),
DatasetHeaderFactory.getShowSubscriptionHeader()));
}
+
+ public static void buildDetailsTSBlock(
+ final List<TSubscriptionProgressInfo> progressInfoList,
+ final SettableFuture<ConfigTaskResult> future) {
+ final TsBlockBuilder builder =
+ new TsBlockBuilder(
+ ColumnHeaderConstant.showSubscriptionDetailsColumnHeaders.stream()
+ .map(ColumnHeader::getColumnType)
+ .collect(Collectors.toList()));
+
+ for (final TSubscriptionProgressInfo progressInfo : progressInfoList) {
+ builder.getTimeColumnBuilder().writeLong(0L);
+ writeText(builder, 0, progressInfo.getTopicName() + "_" +
progressInfo.getConsumerGroupId());
+ writeText(builder, 1, progressInfo.getTopicName());
+ writeText(builder, 2, progressInfo.getConsumerGroupId());
+ builder.getColumnBuilder(3).writeInt(progressInfo.getDataNodeId());
+ writeText(builder, 4, progressInfo.getRegionId());
+ writeText(builder, 5, progressInfo.getStatus());
+ builder.getColumnBuilder(6).writeBoolean(progressInfo.isActive());
+ builder.getColumnBuilder(7).writeBoolean(progressInfo.isInitialized());
+
builder.getColumnBuilder(8).writeLong(progressInfo.getRemainingEventCount());
+ builder.getColumnBuilder(9).writeLong(progressInfo.getRawWalGap());
+ builder.getColumnBuilder(10).writeLong(progressInfo.getApproximateLag());
+
builder.getColumnBuilder(11).writeLong(progressInfo.getInFlightEventCount());
+
builder.getColumnBuilder(12).writeLong(progressInfo.getPrefetchedEventCount());
+
builder.getColumnBuilder(13).writeLong(progressInfo.getPendingEventCount());
+
builder.getColumnBuilder(14).writeLong(progressInfo.getCurrentWalSearchIndex());
+
builder.getColumnBuilder(15).writeLong(progressInfo.getNextReadSearchIndex());
+
builder.getColumnBuilder(16).writeLong(progressInfo.getLastProgressTimeMs());
+ builder.getColumnBuilder(17).writeLong(progressInfo.getLastPollTimeMs());
+ writeText(builder, 18, progressInfo.getLastConsumerId());
+ builder.getColumnBuilder(19).writeLong(progressInfo.getSeekGeneration());
+ builder.declarePosition();
+ }
+
+ future.set(
+ new ConfigTaskResult(
+ TSStatusCode.SUCCESS_STATUS,
+ builder.build(),
+ DatasetHeaderFactory.getShowSubscriptionDetailsHeader()));
+ }
+
+ private static void writeText(
+ final TsBlockBuilder builder, final int columnIndex, final String value)
{
+ builder
+ .getColumnBuilder(columnIndex)
+ .writeBinary(new Binary(value == null ? "" : value,
TSFileConfig.STRING_CHARSET));
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
index dc78f695902..ce3c9f6e485 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
@@ -4543,6 +4543,7 @@ public class ASTVisitor extends
IoTDBSqlParserBaseVisitor<Statement> {
if (ctx.topicName != null) {
showSubscriptionsStatement.setTopicName(parseIdentifier(ctx.topicName.getText()));
}
+ showSubscriptionsStatement.setDetails(ctx.DETAILS() != null);
return showSubscriptionsStatement;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowSubscriptions.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowSubscriptions.java
index 81af08652d9..7a2501f502a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowSubscriptions.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/ShowSubscriptions.java
@@ -34,15 +34,25 @@ public class ShowSubscriptions extends
SubscriptionStatement {
RamUsageEstimator.shallowSizeOfInstance(ShowSubscriptions.class);
private final String topicName;
+ private final boolean details;
public ShowSubscriptions(final String topicName) {
+ this(topicName, false);
+ }
+
+ public ShowSubscriptions(final String topicName, final boolean details) {
this.topicName = topicName;
+ this.details = details;
}
public String getTopicName() {
return topicName;
}
+ public boolean isDetails() {
+ return details;
+ }
+
@Override
public <R, C> R accept(final IAstVisitor<R, C> visitor, final C context) {
return ((AstVisitor<R, C>) visitor).visitShowSubscriptions(this, context);
@@ -50,7 +60,7 @@ public class ShowSubscriptions extends SubscriptionStatement {
@Override
public int hashCode() {
- return Objects.hash(topicName);
+ return Objects.hash(topicName, details);
}
@Override
@@ -62,12 +72,12 @@ public class ShowSubscriptions extends
SubscriptionStatement {
return false;
}
final ShowSubscriptions that = (ShowSubscriptions) obj;
- return Objects.equals(this.topicName, that.topicName);
+ return Objects.equals(this.topicName, that.topicName) && this.details ==
that.details;
}
@Override
public String toString() {
- return toStringHelper(this).add("topicName", topicName).toString();
+ return toStringHelper(this).add("topicName", topicName).add("details",
details).toString();
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index 87e1b417e44..f7a99ab3485 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -1455,7 +1455,7 @@ public class AstBuilder extends
RelationalSqlBaseVisitor<Node> {
RelationalSqlParser.ShowSubscriptionsStatementContext ctx) {
final String topicName =
getIdentifierIfPresent(ctx.identifier()).map(Identifier::getValue).orElse(null);
- return new ShowSubscriptions(topicName);
+ return new ShowSubscriptions(topicName, ctx.DETAILS() != null);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
index 28a8dc47138..dec3cc58b96 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
@@ -850,10 +850,14 @@ public final class DataNodeSqlFormatter extends
CommonQuerySqlFormatter
@Override
public Void visitShowSubscriptions(ShowSubscriptions node, Integer context) {
+ builder.append("SHOW SUBSCRIPTIONS");
+ if (node.isDetails()) {
+ builder.append(" DETAILS");
+ }
if (Objects.isNull(node.getTopicName())) {
- builder.append("SHOW SUBSCRIPTIONS");
+ return null;
} else {
- builder.append("SHOW SUBSCRIPTIONS ON ").append(node.getTopicName());
+ builder.append(" ON
").append(ShowCreateTableTask.getIdentifier(node.getTopicName()));
}
return null;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/ShowSubscriptionsStatement.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/ShowSubscriptionsStatement.java
index aa2abbfc188..c86d7254d46 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/ShowSubscriptionsStatement.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/ShowSubscriptionsStatement.java
@@ -33,6 +33,7 @@ public class ShowSubscriptionsStatement extends Statement
implements IConfigStat
private String topicName;
private boolean isTableModel;
+ private boolean details;
public ShowSubscriptionsStatement() {
super();
@@ -47,6 +48,10 @@ public class ShowSubscriptionsStatement extends Statement
implements IConfigStat
return isTableModel;
}
+ public boolean isDetails() {
+ return details;
+ }
+
public void setTopicName(final String topicName) {
this.topicName = topicName;
}
@@ -55,6 +60,10 @@ public class ShowSubscriptionsStatement extends Statement
implements IConfigStat
this.isTableModel = tableModel;
}
+ public void setDetails(final boolean details) {
+ this.details = details;
+ }
+
@Override
public <R, C> R accept(final StatementVisitor<R, C> visitor, final C
context) {
return visitor.visitShowSubscriptions(this, context);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
index 83e816e9cf3..e65f1d88041 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.subscription.agent;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.schema.table.TsTable;
+import
org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.iot.IoTConsensus;
@@ -988,6 +989,21 @@ public class SubscriptionBrokerAgent {
return
ConsensusSubscriptionCommitManager.getInstance().collectAllRegionProgress(dataNodeId);
}
+ public Map<String, ByteBuffer> collectAllProgressSnapshots() {
+ final Map<String, ByteBuffer> result = new ConcurrentHashMap<>();
+ for (final Map.Entry<String, List<ISubscriptionBroker>> entry :
+ consumerGroupIdToBrokers.entrySet()) {
+ for (final ISubscriptionBroker broker : entry.getValue()) {
+ for (final Map.Entry<String, SubscriptionProgressSnapshot>
snapshotEntry :
+ broker.getProgressSnapshotMap().entrySet()) {
+ result.put(
+ entry.getKey() + "/" + snapshotEntry.getKey(),
snapshotEntry.getValue().serialize());
+ }
+ }
+ }
+ return result;
+ }
+
/**
* Receives a committed progress broadcast from another DataNode (Leader →
Follower). Delegates to
* CommitManager to update local progress state.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
index 441f6ac0daf..b911478c2b2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.subscription.broker;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import
org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot;
import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
@@ -496,6 +497,21 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
return lagMap;
}
+ @Override
+ public Map<String, SubscriptionProgressSnapshot> getProgressSnapshotMap() {
+ final Map<String, SubscriptionProgressSnapshot> progressMap = new
ConcurrentHashMap<>();
+ for (final Map.Entry<String, List<ConsensusPrefetchingQueue>> entry :
+ topicNameToConsensusPrefetchingQueues.entrySet()) {
+ for (final ConsensusPrefetchingQueue queue : entry.getValue()) {
+ if (!queue.isClosed()) {
+ progressMap.put(
+ entry.getKey() + "/" + queue.getConsensusGroupId(),
queue.getProgressSnapshot());
+ }
+ }
+ }
+ return progressMap;
+ }
+
private TopicOwnershipSnapshot refreshAndGetTopicOwnership(
final String topicName,
final List<ConsensusPrefetchingQueue> queues,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ISubscriptionBroker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ISubscriptionBroker.java
index 547ffc6b763..90eb947292e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ISubscriptionBroker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ISubscriptionBroker.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.subscription.broker;
+import
org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
import org.apache.iotdb.rpc.subscription.payload.poll.TopicProgress;
@@ -86,6 +87,10 @@ public interface ISubscriptionBroker {
return Collections.emptyMap();
}
+ default Map<String, SubscriptionProgressSnapshot> getProgressSnapshotMap() {
+ return Collections.emptyMap();
+ }
+
void unbind(String topicName);
void removeQueue(String topicName);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index 3a47be814e5..c6fd52301a4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import
org.apache.iotdb.commons.subscription.meta.consumer.SubscriptionProgressSnapshot;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
@@ -263,6 +264,13 @@ public class ConsensusPrefetchingQueue {
private volatile long lastStatsLogTimeMs = System.currentTimeMillis();
+ /** Wall-clock time of the most recent event poll and committed progress
update. */
+ private volatile long lastPollTimeMs = 0L;
+
+ private volatile long lastProgressTimeMs = 0L;
+
+ private volatile String lastConsumerId = "";
+
private volatile long lastPendingAcceptedEntries = 0L;
private volatile long lastWalAcceptedEntries = 0L;
@@ -1285,6 +1293,8 @@ public class ConsensusPrefetchingQueue {
// Mark as polled before updating inFlightEvents
event.recordLastPolledTimestamp();
+ lastPollTimeMs = System.currentTimeMillis();
+ lastConsumerId = consumerId;
inFlightEvents.put(new InFlightEventKey(consumerId,
event.getCommitContext()), event);
event.recordLastPolledConsumerId(consumerId);
return event;
@@ -3282,6 +3292,7 @@ public class ConsensusPrefetchingQueue {
commitManager.commit(
consumerGroupId, topicName, consensusGroupId, writerId,
writerProgress);
if (committed) {
+ lastProgressTimeMs = System.currentTimeMillis();
refreshCommittedWalRetentionBoundAndNotify();
}
return committed;
@@ -3293,6 +3304,7 @@ public class ConsensusPrefetchingQueue {
commitManager.commitWithoutOutstanding(
consumerGroupId, topicName, consensusGroupId, writerId,
writerProgress);
if (committed) {
+ lastProgressTimeMs = System.currentTimeMillis();
refreshCommittedWalRetentionBoundAndNotify();
}
return committed;
@@ -3918,6 +3930,98 @@ public class ConsensusPrefetchingQueue {
return prefetchInitialized ? 1L : 0L;
}
+ public long getRawWalGap() {
+ final long currentSearchIndex = consensusReqReader.getCurrentSearchIndex();
+ final long nextSearchIndex = nextExpectedSearchIndex.get();
+ return currentSearchIndex >= nextSearchIndex && currentSearchIndex > 0
+ ? currentSearchIndex - nextSearchIndex + 1
+ : 0L;
+ }
+
+ public long getRemainingEventCount() {
+ return 0L
+ + prefetchingQueue.size()
+ + inFlightEvents.size()
+ + pendingEntries.size()
+ + getRealtimeBufferedEntryCount()
+ + lingerBatch.getEntryCount();
+ }
+
+ public long getLastPollTimeMs() {
+ return lastPollTimeMs;
+ }
+
+ public long getLastProgressTimeMs() {
+ return lastProgressTimeMs;
+ }
+
+ /** Returns 0=uninitialized, 1=inactive, 2=caught up, 3=catching up,
4=stalled. */
+ public long getProgressStatus() {
+ switch (getProgressStatusName()) {
+ case SubscriptionProgressSnapshot.STATUS_INACTIVE:
+ return 1L;
+ case SubscriptionProgressSnapshot.STATUS_CAUGHT_UP:
+ return 2L;
+ case SubscriptionProgressSnapshot.STATUS_CATCHING_UP:
+ return 3L;
+ case SubscriptionProgressSnapshot.STATUS_STALLED:
+ return 4L;
+ default:
+ return 0L;
+ }
+ }
+
+ public String getProgressStatusName() {
+ if (!prefetchInitialized) {
+ return SubscriptionProgressSnapshot.STATUS_UNINITIALIZED;
+ }
+ if (!isActive) {
+ return SubscriptionProgressSnapshot.STATUS_INACTIVE;
+ }
+ if (getLag() <= 0L) {
+ return SubscriptionProgressSnapshot.STATUS_CAUGHT_UP;
+ }
+ final long stalledTimeoutMs =
+
SubscriptionConfig.getInstance().getSubscriptionConsensusConsumerEvictionTimeoutMs();
+ final long progressReferenceTimeMs =
+ lastProgressTimeMs > 0L ? lastProgressTimeMs : lastPollTimeMs;
+ if (progressReferenceTimeMs > 0L
+ && stalledTimeoutMs > 0L
+ && System.currentTimeMillis() - progressReferenceTimeMs >=
stalledTimeoutMs) {
+ return SubscriptionProgressSnapshot.STATUS_STALLED;
+ }
+ return SubscriptionProgressSnapshot.STATUS_CATCHING_UP;
+ }
+
+ public SubscriptionProgressSnapshot getProgressSnapshot() {
+ final long currentWalSearchIndex =
consensusReqReader.getCurrentSearchIndex();
+ final long nextReadSearchIndex = nextExpectedSearchIndex.get();
+ return new SubscriptionProgressSnapshot(
+ IoTDBDescriptor.getInstance().getConfig().getDataNodeId(),
+ consumerGroupId,
+ topicName,
+ consensusGroupId.toString(),
+ isActive,
+ prefetchInitialized,
+ currentWalSearchIndex,
+ nextReadSearchIndex,
+ getRawWalGap(),
+ getLag(),
+ prefetchingQueue.size(),
+ inFlightEvents.size(),
+ pendingEntries.size(),
+ getRealtimeBufferedEntryCount(),
+ lingerBatch.getEntryCount(),
+ lastPollTimeMs,
+ lastProgressTimeMs,
+ lastConsumerId,
+ seekGeneration.get(),
+ walGapSkippedEntries.get(),
+ runtimeVersionChangeCount.get(),
+ maxObservedTimestamp,
+ getProgressStatusName());
+ }
+
public void setActiveWriterNodeIds(final Set<Integer> activeWriterNodeIds) {
this.runtimeActiveWriterNodeIds =
Collections.unmodifiableSet(
@@ -4119,10 +4223,18 @@ public class ConsensusPrefetchingQueue {
return prefetchingQueue.size();
}
+ public int getPendingEventCount() {
+ return pendingEntries.size();
+ }
+
public long getCurrentReadSearchIndex() {
return nextExpectedSearchIndex.get();
}
+ public long getCurrentWalSearchIndex() {
+ return consensusReqReader.getCurrentSearchIndex();
+ }
+
public long getPendingPathAcceptedEntries() {
return pendingPathAcceptedEntries.get();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
index 50ea8402118..b7331749b8c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
@@ -145,6 +145,66 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
queue,
ConsensusPrefetchingQueue::getInitializedStatus,
key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_REMAINING_EVENT_COUNT.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getRemainingEventCount,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_RAW_WAL_GAP.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getRawWalGap,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_CURRENT_WAL_SEARCH_INDEX.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getCurrentWalSearchIndex,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_NEXT_READ_SEARCH_INDEX.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getCurrentReadSearchIndex,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_IN_FLIGHT_EVENT_COUNT.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getSubscriptionUncommittedEventCount,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_PREFETCHED_EVENT_COUNT.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getPrefetchedEventCount,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_PENDING_EVENT_COUNT.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getPendingEventCount,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getLastPollTimeMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getLastProgressTimeMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getProgressStatus,
+ key.getTags());
}
private void createRate(final QueueMetricsKey key) {
@@ -217,6 +277,44 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_ACTIVE.toString(), key.getTags());
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_INITIALIZED.toString(), key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_REMAINING_EVENT_COUNT.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_RAW_WAL_GAP.toString(), key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_CURRENT_WAL_SEARCH_INDEX.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_NEXT_READ_SEARCH_INDEX.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_IN_FLIGHT_EVENT_COUNT.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_PREFETCHED_EVENT_COUNT.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_PENDING_EVENT_COUNT.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(),
+ key.getTags());
}
private void removeRate(final QueueMetricsKey key) {
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
new file mode 100644
index 00000000000..b8191dca629
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/ShowSubscriptionsTaskTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.db.queryengine.plan.execution.config.sys.subscription;
+
+import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
+import org.apache.iotdb.confignode.rpc.thrift.TSubscriptionProgressInfo;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ShowSubscriptionsTaskTest {
+
+ @Test
+ public void testBuildDetailsTSBlock() throws Exception {
+ final TSubscriptionProgressInfo progressInfo =
+ new TSubscriptionProgressInfo(
+ "topic",
+ "group",
+ "DataRegion[1]",
+ 3,
+ true,
+ true,
+ "CATCHING_UP",
+ 5L,
+ 20L,
+ 6L,
+ 2L,
+ 1L,
+ 2L,
+ 100L,
+ 81L,
+ 123L,
+ 120L,
+ "consumer",
+ 4L);
+ final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+
+
ShowSubscriptionsTask.buildDetailsTSBlock(Collections.singletonList(progressInfo),
future);
+
+ final ConfigTaskResult result = future.get();
+ final TsBlock resultSet = result.getResultSet();
+ assertEquals(TSStatusCode.SUCCESS_STATUS, result.getStatusCode());
+ assertEquals(
+ ColumnHeaderConstant.SUBSCRIPTION_ID,
result.getResultSetHeader().getRespColumns().get(0));
+ assertEquals(
+ ColumnHeaderConstant.SEEK_GENERATION,
result.getResultSetHeader().getRespColumns().get(19));
+ assertEquals(1, resultSet.getPositionCount());
+ assertEquals("topic_group",
resultSet.getColumn(0).getBinary(0).toString());
+ assertEquals("CATCHING_UP",
resultSet.getColumn(5).getBinary(0).toString());
+ assertTrue(resultSet.getColumn(6).getBoolean(0));
+ assertEquals(5L, resultSet.getColumn(8).getLong(0));
+ assertEquals(20L, resultSet.getColumn(9).getLong(0));
+ assertEquals(4L, resultSet.getColumn(19).getLong(0));
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/ShowSubscriptionsStatementTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/ShowSubscriptionsStatementTest.java
new file mode 100644
index 00000000000..b4d05f71b33
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/ShowSubscriptionsStatementTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.db.queryengine.plan.parser;
+
+import org.apache.iotdb.db.queryengine.plan.statement.Statement;
+import
org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.ShowSubscriptionsStatement;
+
+import org.junit.Test;
+
+import java.time.ZoneId;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ShowSubscriptionsStatementTest {
+
+ @Test
+ public void testParseDetails() {
+ assertDetails("SHOW SUBSCRIPTIONS DETAILS", null);
+ assertDetails("SHOW SUBSCRIPTIONS DETAILS ON `topic-1`", "topic-1");
+ }
+
+ private void assertDetails(final String sql, final String topicName) {
+ final Statement statement = StatementGenerator.createStatement(sql,
ZoneId.systemDefault());
+
+ assertTrue(statement instanceof ShowSubscriptionsStatement);
+ final ShowSubscriptionsStatement showSubscriptions =
(ShowSubscriptionsStatement) statement;
+ assertTrue(showSubscriptions.isDetails());
+ assertEquals(topicName, showSubscriptions.getTopicName());
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ShowSubscriptionsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ShowSubscriptionsTest.java
new file mode 100644
index 00000000000..389fcaf0c25
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ShowSubscriptionsTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.db.queryengine.plan.relational.sql;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.protocol.session.InternalClientSession;
+import
org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowSubscriptions;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.parser.SqlParser;
+import
org.apache.iotdb.db.queryengine.plan.relational.sql.util.DataNodeSqlFormatter;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.time.ZoneId;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ShowSubscriptionsTest {
+
+ private SqlParser sqlParser;
+ private IClientSession clientSession;
+
+ @Before
+ public void setUp() {
+ sqlParser = new SqlParser();
+ clientSession = new InternalClientSession("testClient");
+ }
+
+ @Test
+ public void testDetailsRoundTrip() {
+ assertDetailsRoundTrip("SHOW SUBSCRIPTIONS DETAILS", null);
+ assertDetailsRoundTrip("SHOW SUBSCRIPTIONS DETAILS ON \"topic-1\"",
"topic-1");
+ }
+
+ private void assertDetailsRoundTrip(final String sql, final String
topicName) {
+ final Statement statement =
+ sqlParser.createStatement(sql, ZoneId.systemDefault(), clientSession);
+
+ assertTrue(statement instanceof ShowSubscriptions);
+ final ShowSubscriptions showSubscriptions = (ShowSubscriptions) statement;
+ assertTrue(showSubscriptions.isDetails());
+ assertEquals(topicName, showSubscriptions.getTopicName());
+ assertEquals(sql, DataNodeSqlFormatter.formatDataNodeSql(statement));
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
index f6bae43f238..08880f4fcb3 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
@@ -112,6 +112,26 @@ public class
ConsensusSubscriptionPrefetchingQueueMetricsTest {
eq(queueId),
eq(Tag.REGION.toString()),
eq(firstRegionId.toString()));
+ verify(metricService)
+ .createAutoGauge(
+
eq(Metric.SUBSCRIPTION_CONSENSUS_REMAINING_EVENT_COUNT.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(firstQueue),
+ any(),
+ eq(Tag.NAME.toString()),
+ eq(queueId),
+ eq(Tag.REGION.toString()),
+ eq(firstRegionId.toString()));
+ verify(metricService)
+ .createAutoGauge(
+ eq(Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(firstQueue),
+ any(),
+ eq(Tag.NAME.toString()),
+ eq(queueId),
+ eq(Tag.REGION.toString()),
+ eq(firstRegionId.toString()));
verify(metricService)
.createAutoGauge(
eq(Metric.SUBSCRIPTION_CONSENSUS_LAG.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 06c361eba0a..d66b5a8a301 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
@@ -140,6 +140,19 @@ public class ColumnHeaderConstant {
public static final String CONSUMER_GROUP_NAME = "ConsumerGroupName";
public static final String SUBSCRIBED_CONSUMERS = "SubscribedConsumers";
public static final String SUBSCRIPTION_ID = "SubscriptionID";
+ public static final String SUBSCRIPTION_ACTIVE = "Active";
+ public static final String SUBSCRIPTION_INITIALIZED = "Initialized";
+ public static final String RAW_WAL_GAP = "RawWalGap";
+ public static final String APPROXIMATE_LAG = "ApproximateLag";
+ public static final String IN_FLIGHT_EVENT_COUNT = "InFlightEventCount";
+ public static final String PREFETCHED_EVENT_COUNT = "PrefetchedEventCount";
+ public static final String PENDING_EVENT_COUNT = "PendingEventCount";
+ public static final String CURRENT_WAL_SEARCH_INDEX =
"CurrentWalSearchIndex";
+ public static final String NEXT_READ_SEARCH_INDEX = "NextReadSearchIndex";
+ public static final String LAST_PROGRESS_TIME = "LastProgressTime";
+ public static final String LAST_POLL_TIME = "LastPollTime";
+ public static final String LAST_CONSUMER_ID = "LastConsumerId";
+ public static final String SEEK_GENERATION = "SeekGeneration";
// show cluster status
public static final String NODE_TYPE_CONFIG_NODE = "ConfigNode";
@@ -647,6 +660,29 @@ public class ColumnHeaderConstant {
new ColumnHeader(CONSUMER_GROUP_NAME, TSDataType.TEXT),
new ColumnHeader(SUBSCRIBED_CONSUMERS, TSDataType.TEXT));
+ public static final List<ColumnHeader> showSubscriptionDetailsColumnHeaders =
+ ImmutableList.of(
+ new ColumnHeader(SUBSCRIPTION_ID, TSDataType.TEXT),
+ new ColumnHeader(TOPIC_NAME, TSDataType.TEXT),
+ new ColumnHeader(CONSUMER_GROUP_NAME, TSDataType.TEXT),
+ new ColumnHeader(DATA_NODE_ID, TSDataType.INT32),
+ new ColumnHeader(REGION_ID, TSDataType.TEXT),
+ new ColumnHeader(STATUS, TSDataType.TEXT),
+ new ColumnHeader(SUBSCRIPTION_ACTIVE, TSDataType.BOOLEAN),
+ new ColumnHeader(SUBSCRIPTION_INITIALIZED, TSDataType.BOOLEAN),
+ new ColumnHeader(REMAINING_EVENT_COUNT, TSDataType.INT64),
+ new ColumnHeader(RAW_WAL_GAP, TSDataType.INT64),
+ new ColumnHeader(APPROXIMATE_LAG, TSDataType.INT64),
+ new ColumnHeader(IN_FLIGHT_EVENT_COUNT, TSDataType.INT64),
+ new ColumnHeader(PREFETCHED_EVENT_COUNT, TSDataType.INT64),
+ new ColumnHeader(PENDING_EVENT_COUNT, TSDataType.INT64),
+ new ColumnHeader(CURRENT_WAL_SEARCH_INDEX, TSDataType.INT64),
+ new ColumnHeader(NEXT_READ_SEARCH_INDEX, TSDataType.INT64),
+ 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));
+
public static final List<ColumnHeader> selectIntoColumnHeaders =
ImmutableList.of(
new ColumnHeader(SOURCE_COLUMN, TSDataType.TEXT),
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index 70c4976df28..b41d8fb53ac 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -217,6 +217,17 @@ public enum Metric {
SUBSCRIPTION_CONSENSUS_SEEK_GENERATION("subscription_consensus_seek_generation"),
SUBSCRIPTION_CONSENSUS_ACTIVE("subscription_consensus_active"),
SUBSCRIPTION_CONSENSUS_INITIALIZED("subscription_consensus_initialized"),
+
SUBSCRIPTION_CONSENSUS_REMAINING_EVENT_COUNT("subscription_consensus_remaining_event_count"),
+ SUBSCRIPTION_CONSENSUS_RAW_WAL_GAP("subscription_consensus_raw_wal_gap"),
+ SUBSCRIPTION_CONSENSUS_CURRENT_WAL_SEARCH_INDEX(
+ "subscription_consensus_current_wal_search_index"),
+
SUBSCRIPTION_CONSENSUS_NEXT_READ_SEARCH_INDEX("subscription_consensus_next_read_search_index"),
+
SUBSCRIPTION_CONSENSUS_IN_FLIGHT_EVENT_COUNT("subscription_consensus_in_flight_event_count"),
+
SUBSCRIPTION_CONSENSUS_PREFETCHED_EVENT_COUNT("subscription_consensus_prefetched_event_count"),
+
SUBSCRIPTION_CONSENSUS_PENDING_EVENT_COUNT("subscription_consensus_pending_event_count"),
+
SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME("subscription_consensus_last_poll_time"),
+
SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME("subscription_consensus_last_progress_time"),
+
SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS("subscription_consensus_progress_status"),
// load related
ACTIVE_LOADING_FILES_NUMBER("active_loading_files_number"),
ACTIVE_LOADING_FILES_SIZE("active_loading_files_size"),
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/SubscriptionProgressSnapshot.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/SubscriptionProgressSnapshot.java
new file mode 100644
index 00000000000..75ff3776923
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/SubscriptionProgressSnapshot.java
@@ -0,0 +1,278 @@
+/*
+ * 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.commons.subscription.meta.consumer;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * A side-effect-free point-in-time view of a consensus subscription queue.
+ *
+ * <p>The in-memory counts are exact for the queue lifecycle stages. {@link
#rawWalGap} is a
+ * search-index distance and therefore is deliberately named as such: it is
not the number of
+ * topic-filtered events remaining in the WAL.
+ */
+public final class SubscriptionProgressSnapshot {
+
+ public static final String STATUS_UNINITIALIZED = "UNINITIALIZED";
+ public static final String STATUS_INACTIVE = "INACTIVE";
+ public static final String STATUS_CAUGHT_UP = "CAUGHT_UP";
+ public static final String STATUS_CATCHING_UP = "CATCHING_UP";
+ public static final String STATUS_STALLED = "STALLED";
+ public static final String STATUS_NO_QUEUE = "NO_QUEUE";
+
+ private final int dataNodeId;
+ private final String consumerGroupId;
+ private final String topicName;
+ private final String regionId;
+ private final boolean active;
+ private final boolean initialized;
+ private final long currentWalSearchIndex;
+ private final long nextReadSearchIndex;
+ private final long rawWalGap;
+ private final long approximateLag;
+ private final long prefetchedEventCount;
+ private final long inFlightEventCount;
+ private final long pendingEventCount;
+ private final long realtimeBufferedEventCount;
+ private final long lingerEventCount;
+ private final long lastPollTimeMs;
+ private final long lastProgressTimeMs;
+ private final String lastConsumerId;
+ private final long seekGeneration;
+ private final long walGapSkippedEntries;
+ private final long routingEpochChangeCount;
+ private final long maxObservedTimestamp;
+ private final String status;
+
+ public SubscriptionProgressSnapshot(
+ final int dataNodeId,
+ final String consumerGroupId,
+ final String topicName,
+ final String regionId,
+ final boolean active,
+ final boolean initialized,
+ final long currentWalSearchIndex,
+ final long nextReadSearchIndex,
+ final long rawWalGap,
+ final long approximateLag,
+ final long prefetchedEventCount,
+ final long inFlightEventCount,
+ final long pendingEventCount,
+ final long realtimeBufferedEventCount,
+ final long lingerEventCount,
+ final long lastPollTimeMs,
+ final long lastProgressTimeMs,
+ final String lastConsumerId,
+ final long seekGeneration,
+ final long walGapSkippedEntries,
+ final long routingEpochChangeCount,
+ final long maxObservedTimestamp,
+ final String status) {
+ this.dataNodeId = dataNodeId;
+ this.consumerGroupId = consumerGroupId;
+ this.topicName = topicName;
+ this.regionId = regionId;
+ this.active = active;
+ this.initialized = initialized;
+ this.currentWalSearchIndex = currentWalSearchIndex;
+ this.nextReadSearchIndex = nextReadSearchIndex;
+ this.rawWalGap = rawWalGap;
+ this.approximateLag = approximateLag;
+ this.prefetchedEventCount = prefetchedEventCount;
+ this.inFlightEventCount = inFlightEventCount;
+ this.pendingEventCount = pendingEventCount;
+ this.realtimeBufferedEventCount = realtimeBufferedEventCount;
+ this.lingerEventCount = lingerEventCount;
+ this.lastPollTimeMs = lastPollTimeMs;
+ this.lastProgressTimeMs = lastProgressTimeMs;
+ this.lastConsumerId = lastConsumerId;
+ this.seekGeneration = seekGeneration;
+ this.walGapSkippedEntries = walGapSkippedEntries;
+ this.routingEpochChangeCount = routingEpochChangeCount;
+ this.maxObservedTimestamp = maxObservedTimestamp;
+ this.status = status;
+ }
+
+ public int getDataNodeId() {
+ return dataNodeId;
+ }
+
+ public String getConsumerGroupId() {
+ return consumerGroupId;
+ }
+
+ public String getTopicName() {
+ return topicName;
+ }
+
+ public String getRegionId() {
+ return regionId;
+ }
+
+ public boolean isActive() {
+ return active;
+ }
+
+ public boolean isInitialized() {
+ return initialized;
+ }
+
+ public long getCurrentWalSearchIndex() {
+ return currentWalSearchIndex;
+ }
+
+ public long getNextReadSearchIndex() {
+ return nextReadSearchIndex;
+ }
+
+ public long getRawWalGap() {
+ return rawWalGap;
+ }
+
+ public long getApproximateLag() {
+ return approximateLag;
+ }
+
+ public long getPrefetchedEventCount() {
+ return prefetchedEventCount;
+ }
+
+ public long getInFlightEventCount() {
+ return inFlightEventCount;
+ }
+
+ public long getPendingEventCount() {
+ return pendingEventCount;
+ }
+
+ public long getRealtimeBufferedEventCount() {
+ return realtimeBufferedEventCount;
+ }
+
+ public long getLingerEventCount() {
+ return lingerEventCount;
+ }
+
+ public long getRemainingEventCount() {
+ return prefetchedEventCount
+ + inFlightEventCount
+ + pendingEventCount
+ + realtimeBufferedEventCount
+ + lingerEventCount;
+ }
+
+ public long getLastPollTimeMs() {
+ return lastPollTimeMs;
+ }
+
+ public long getLastProgressTimeMs() {
+ return lastProgressTimeMs;
+ }
+
+ public String getLastConsumerId() {
+ return lastConsumerId;
+ }
+
+ public long getSeekGeneration() {
+ return seekGeneration;
+ }
+
+ public long getWalGapSkippedEntries() {
+ return walGapSkippedEntries;
+ }
+
+ public long getRoutingEpochChangeCount() {
+ return routingEpochChangeCount;
+ }
+
+ public long getMaxObservedTimestamp() {
+ return maxObservedTimestamp;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public ByteBuffer serialize() {
+ try (final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ final DataOutputStream stream = new DataOutputStream(output)) {
+ ReadWriteIOUtils.write(dataNodeId, stream);
+ ReadWriteIOUtils.write(consumerGroupId, stream);
+ ReadWriteIOUtils.write(topicName, stream);
+ ReadWriteIOUtils.write(regionId, stream);
+ ReadWriteIOUtils.write(active, stream);
+ ReadWriteIOUtils.write(initialized, stream);
+ ReadWriteIOUtils.write(currentWalSearchIndex, stream);
+ ReadWriteIOUtils.write(nextReadSearchIndex, stream);
+ ReadWriteIOUtils.write(rawWalGap, stream);
+ ReadWriteIOUtils.write(approximateLag, stream);
+ ReadWriteIOUtils.write(prefetchedEventCount, stream);
+ ReadWriteIOUtils.write(inFlightEventCount, stream);
+ ReadWriteIOUtils.write(pendingEventCount, stream);
+ ReadWriteIOUtils.write(realtimeBufferedEventCount, stream);
+ ReadWriteIOUtils.write(lingerEventCount, stream);
+ ReadWriteIOUtils.write(lastPollTimeMs, stream);
+ ReadWriteIOUtils.write(lastProgressTimeMs, stream);
+ ReadWriteIOUtils.write(lastConsumerId, stream);
+ ReadWriteIOUtils.write(seekGeneration, stream);
+ ReadWriteIOUtils.write(walGapSkippedEntries, stream);
+ ReadWriteIOUtils.write(routingEpochChangeCount, stream);
+ ReadWriteIOUtils.write(maxObservedTimestamp, stream);
+ ReadWriteIOUtils.write(status, stream);
+ stream.flush();
+ return ByteBuffer.wrap(output.toByteArray());
+ } catch (final IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ public static SubscriptionProgressSnapshot deserialize(final ByteBuffer
input) {
+ final ByteBuffer buffer = input.asReadOnlyBuffer();
+ return new SubscriptionProgressSnapshot(
+ ReadWriteIOUtils.readInt(buffer),
+ ReadWriteIOUtils.readString(buffer),
+ ReadWriteIOUtils.readString(buffer),
+ ReadWriteIOUtils.readString(buffer),
+ ReadWriteIOUtils.readBool(buffer),
+ ReadWriteIOUtils.readBool(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readString(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readLong(buffer),
+ ReadWriteIOUtils.readString(buffer));
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/SubscriptionProgressSnapshotTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/SubscriptionProgressSnapshotTest.java
new file mode 100644
index 00000000000..e3bbb2c4728
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/SubscriptionProgressSnapshotTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.commons.subscription.meta.consumer;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class SubscriptionProgressSnapshotTest {
+
+ @Test
+ public void testSerializationRoundTrip() {
+ final SubscriptionProgressSnapshot snapshot =
+ new SubscriptionProgressSnapshot(
+ 3,
+ "group",
+ "topic",
+ "3_1",
+ true,
+ true,
+ 100L,
+ 90L,
+ 11L,
+ 12L,
+ 1L,
+ 2L,
+ 3L,
+ 4L,
+ 5L,
+ 6L,
+ 7L,
+ "consumer",
+ 8L,
+ 9L,
+ 10L,
+ 11L,
+ SubscriptionProgressSnapshot.STATUS_STALLED);
+
+ final SubscriptionProgressSnapshot restored =
+ SubscriptionProgressSnapshot.deserialize(snapshot.serialize());
+
+ assertEquals(snapshot.getDataNodeId(), restored.getDataNodeId());
+ assertEquals(snapshot.getConsumerGroupId(), restored.getConsumerGroupId());
+ assertEquals(snapshot.getTopicName(), restored.getTopicName());
+ assertEquals(snapshot.getRegionId(), restored.getRegionId());
+ assertTrue(restored.isActive());
+ assertTrue(restored.isInitialized());
+ assertEquals(snapshot.getCurrentWalSearchIndex(),
restored.getCurrentWalSearchIndex());
+ assertEquals(snapshot.getNextReadSearchIndex(),
restored.getNextReadSearchIndex());
+ assertEquals(snapshot.getRawWalGap(), restored.getRawWalGap());
+ assertEquals(snapshot.getApproximateLag(), restored.getApproximateLag());
+ assertEquals(snapshot.getRemainingEventCount(),
restored.getRemainingEventCount());
+ assertEquals(snapshot.getLastPollTimeMs(), restored.getLastPollTimeMs());
+ assertEquals(snapshot.getLastProgressTimeMs(),
restored.getLastProgressTimeMs());
+ assertEquals(snapshot.getLastConsumerId(), restored.getLastConsumerId());
+ assertEquals(snapshot.getSeekGeneration(), restored.getSeekGeneration());
+ assertEquals(snapshot.getStatus(), restored.getStatus());
+ assertTrue(restored.serialize().remaining() > 0);
+ }
+}
diff --git
a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
index 6ab05044dfb..037778597d4 100644
---
a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
+++
b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
@@ -574,7 +574,7 @@ showCreateTopicStatement
;
showSubscriptionsStatement
- : SHOW SUBSCRIPTIONS (ON topicName=identifier)?
+ : SHOW SUBSCRIPTIONS (DETAILS)? (ON topicName=identifier)?
;
dropSubscriptionStatement
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index f0f2292027a..512e2e05f05 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -1068,11 +1068,13 @@ struct TUnsubscribeReq {
struct TShowSubscriptionReq {
1: optional string topicName
2: optional bool isTableModel
+ 3: optional bool details
}
struct TShowSubscriptionResp {
1: required common.TSStatus status
2: optional list<TShowSubscriptionInfo> subscriptionInfoList
+ 3: optional list<TSubscriptionProgressInfo> subscriptionProgressList
}
struct TShowSubscriptionInfo {
@@ -1082,6 +1084,28 @@ struct TShowSubscriptionInfo {
4: optional i64 creationTime
}
+struct TSubscriptionProgressInfo {
+ 1: required string topicName
+ 2: required string consumerGroupId
+ 3: required string regionId
+ 4: required i32 dataNodeId
+ 5: required bool active
+ 6: required bool initialized
+ 7: required string status
+ 8: required i64 remainingEventCount
+ 9: required i64 rawWalGap
+ 10: required i64 approximateLag
+ 11: required i64 inFlightEventCount
+ 12: required i64 prefetchedEventCount
+ 13: required i64 pendingEventCount
+ 14: required i64 currentWalSearchIndex
+ 15: required i64 nextReadSearchIndex
+ 16: required i64 lastProgressTimeMs
+ 17: required i64 lastPollTimeMs
+ 18: required string lastConsumerId
+ 19: required i64 seekGeneration
+}
+
struct TDropSubscriptionReq {
1: required string subsciptionId
2: optional bool ifExistsCondition
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 1c3fd78470d..bcef41032aa 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -630,6 +630,7 @@ struct TPullCommitProgressReq {
struct TPullCommitProgressResp {
1: required common.TSStatus status
2: optional map<string, binary> commitRegionProgress
+ 3: optional map<string, binary> subscriptionProgress
}
struct TSyncSubscriptionProgressReq {
