You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Caideyipi (via GitHub)" <gi...@apache.org> on 2026/09/24 05:16:57 UTC
[PR] Support retaining subscription progress after unsubscribe [iotdb]
Caideyipi opened a new pull request, #18718:
URL: https://github.com/apache/iotdb/pull/18718
## Description
Add the incremental topic attribute `retain.progress.after.unsubscribe` to preserve a consumer group's committed progress and the WAL files required to resume from that progress after the group unsubscribes from a topic.
### Configuration and validation
- The attribute defaults to `false`, preserving the existing unsubscribe behavior.
- Attribute names are matched case-insensitively. Values are trimmed and must be `true` or `false`, also case-insensitively.
- The attribute is supported only by incremental topics, together with the existing retention attributes.
- The attribute can be updated online. Changing it from `true` to `false` removes retained state for groups that are no longer subscribed. The existing `retention.bytes` and `retention.ms` attributes remain immutable after topic creation.
### Progress and WAL lifecycle
- When the final consumer in a group unsubscribes, ConfigNode retains that group's committed topic progress when the attribute is enabled.
- DataNode atomically replaces the active prefetching queue with detached WAL retention. Detached retention does not receive live subscription requests and only prevents deletion of WAL files needed by the committed progress.
- Resubscribing atomically replaces detached retention with an active queue and resumes from the retained progress.
- During DataNode recovery, IoTConsensus installs a conservative retention guard before the recovered peer starts. Subscription metadata recovery then replaces that guard with the active queue, the final detached retention policy, or removes it when the retained state is no longer valid.
- Dropping a topic, consumer group, or region always removes its progress and WAL retention state. Setup rollback restores the startup retention guard.
### Tests
- `TopicConfigTest`: 9 tests
- `CommitProgressKeeperTest`: 14 tests
- `SubscriptionQueueRegistryTest`: 4 tests
- Targeted DataNode subscription tests: 32 tests
- Targeted ConfigNode subscription tests: 43 tests
- `mvn test-compile -DskipTests`: 53-module reactor, `BUILD SUCCESS`
- `mvn test-compile -DskipTests -P with-zh-locale`: 53-module reactor, `BUILD SUCCESS`
<hr>
This PR has:
- [x] been self-reviewed.
- [x] concurrent read
- [x] concurrent write
- [x] concurrent read and write
- [x] added comments explaining the "why" and the intent of the code wherever it would not be obvious to an unfamiliar reader.
- [x] added or modified unit tests to cover the new code paths.
<hr>
##### Key changed/added classes (or packages if there are too many classes) in this PR
- `TopicConfig` and `TopicConstant`: expose the new topic attribute and its default.
- `SubscriptionInfo` and `CommitProgressKeeper`: validate the attribute and retain or clean ConfigNode progress.
- `ConsensusSubscriptionSetupHandler` and `ConsensusSubscriptionWalRetention`: switch between active queues and detached WAL retention, including recovery and rollback.
- `IoTConsensusServerImpl` and `SubscriptionQueueRegistry`: manage WAL retention policies and their atomic lifecycle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support retaining subscription progress after unsubscribe [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #18718:
URL: https://github.com/apache/iotdb/pull/18718
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support retaining subscription progress after unsubscribe [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #18718:
URL: https://github.com/apache/iotdb/pull/18718#discussion_r4128847398
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetention.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.subscription.broker.consensus;
+
+import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
+import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
+import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Objects;
+
+final class ConsensusSubscriptionWalRetention {
+
+ private ConsensusSubscriptionWalRetention() {}
+
+ static String generateRetentionId(
+ final String consumerGroupId, final String topicName, final ConsensusGroupId regionId) {
+ return consumerGroupId + "\u0000" + topicName + "\u0000" + regionId;
+ }
+
+ static long computeCommittedRetainedMinVersionId(
+ final ConsensusReqReader consensusReqReader,
+ final ConsensusGroupId consensusGroupId,
+ final RegionProgress committedRegionProgress) {
+ if (!(consensusReqReader instanceof WALNode)) {
+ return 0L;
+ }
+
+ final WALNode walNode = (WALNode) consensusReqReader;
+ final long currentWalVersion = walNode.getCurrentWALFileVersion();
+ final File[] walFiles = WALFileUtils.listAllWALFiles(walNode.getLogDirectory());
Review Comment:
May use org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode#getSortedWalFiles
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java:
##########
@@ -722,6 +722,46 @@ public void removeAllStatesForTopic(final String consumerGroupId, final String t
}
}
+ public void removeAllStatesForConsumerGroup(final String consumerGroupId) {
+ recoverAllTopicStatesIfNeeded();
+ final Set<String> topicNames = new java.util.HashSet<>();
Review Comment:
Mind the import
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetention.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.subscription.broker.consensus;
+
+import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
+import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
+import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Objects;
+
+final class ConsensusSubscriptionWalRetention {
+
+ private ConsensusSubscriptionWalRetention() {}
+
+ static String generateRetentionId(
+ final String consumerGroupId, final String topicName, final ConsensusGroupId regionId) {
+ return consumerGroupId + "\u0000" + topicName + "\u0000" + regionId;
+ }
+
+ static long computeCommittedRetainedMinVersionId(
+ final ConsensusReqReader consensusReqReader,
+ final ConsensusGroupId consensusGroupId,
+ final RegionProgress committedRegionProgress) {
+ if (!(consensusReqReader instanceof WALNode)) {
+ return 0L;
+ }
+
+ final WALNode walNode = (WALNode) consensusReqReader;
+ final long currentWalVersion = walNode.getCurrentWALFileVersion();
+ final File[] walFiles = WALFileUtils.listAllWALFiles(walNode.getLogDirectory());
+ if (Objects.isNull(walFiles) || walFiles.length == 0) {
+ return Math.max(0L, currentWalVersion);
+ }
+
+ WALFileUtils.ascSortByVersionId(walFiles);
+ for (final File walFile : walFiles) {
+ final long versionId = WALFileUtils.parseVersionId(walFile.getName());
+ if (versionId >= currentWalVersion) {
+ return Math.max(0L, currentWalVersion);
+ }
+ if (ProgressWALIterator.isHeaderOnlyWalFile(walFile)) {
+ continue;
+ }
+
+ try (final ProgressWALReader reader = new ProgressWALReader(walFile)) {
+ if (!ConsensusPrefetchingQueue.WalFileCommitRequirement.fromMetadata(
+ consensusGroupId.toString(), reader.getMetaData())
+ .isCoveredBy(committedRegionProgress)) {
+ return versionId;
+ }
+ } catch (final IOException e) {
+ return versionId;
+ }
+ }
+ return Math.max(0L, currentWalVersion);
+ }
Review Comment:
Since the progress is monotonic before any seek, may cache the last covered file version to reduce metadata read.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java:
##########
@@ -116,6 +116,21 @@ public synchronized void removeTopicProgress(
&& isValidRegionAndDataNodeProgressKey(key, legacyTopicKeyPrefix)));
}
+ public synchronized void removeConsumerGroupProgress(final String consumerGroupId) {
+ final String versionedConsumerGroupKeyPrefix =
+ VERSIONED_KEY_PREFIX + encodeKeyComponent(consumerGroupId) + KEY_COMPONENT_SEPARATOR;
+ final String legacyConsumerGroupKeyPrefix = String.valueOf(consumerGroupId) + KEY_SEPARATOR;
+ final boolean legacyConsumerGroupKeyIsUnambiguous =
+ !String.valueOf(consumerGroupId).contains(KEY_SEPARATOR);
+ regionProgressMap
+ .keySet()
+ .removeIf(
+ key ->
+ key.startsWith(versionedConsumerGroupKeyPrefix)
+ || (legacyConsumerGroupKeyIsUnambiguous
+ && key.startsWith(legacyConsumerGroupKeyPrefix)));
+ }
Review Comment:
May skip initializing legacyConsumerGroupKeyPrefix when legacyConsumerGroupKeyIsUnambiguous is false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org