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 8effe5d6ce4 Support retaining subscription progress after unsubscribe
(#18718)
8effe5d6ce4 is described below
commit 8effe5d6ce475eadfd3160afa3baf0662b684ad8
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 14 17:58:38 2026 +0800
Support retaining subscription progress after unsubscribe (#18718)
* Support retaining subscription progress after unsubscribe
* [Subscription] Address WAL retention review comments
---
.../iotdb/rpc/subscription/config/TopicConfig.java | 8 +
.../rpc/subscription/config/TopicConstant.java | 3 +
.../rpc/subscription/config/TopicConfigTest.java | 12 +
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 6 +
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 6 +
.../persistence/subscription/SubscriptionInfo.java | 109 ++++++--
.../subscription/SubscriptionInfoTest.java | 125 +++++++++
.../SubscriptionInfoTopicValidationTest.java | 44 +++
.../apache/iotdb/consensus/iot/IoTConsensus.java | 15 +-
.../consensus/iot/IoTConsensusServerImpl.java | 38 ++-
.../subscription/SubscriptionQueueRegistry.java | 50 +++-
.../SubscriptionQueueRegistryTest.java | 42 +++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../java/org/apache/iotdb/db/service/DataNode.java | 4 +
.../agent/SubscriptionBrokerAgent.java | 12 +-
.../agent/SubscriptionConsumerAgent.java | 42 +++
.../agent/SubscriptionRuntimeAgent.java | 2 +
.../subscription/agent/SubscriptionTopicAgent.java | 5 +
.../broker/ConsensusSubscriptionBroker.java | 24 +-
.../consensus/ConsensusPrefetchingQueue.java | 113 ++++++--
.../ConsensusSubscriptionCommitManager.java | 76 +++++
.../ConsensusSubscriptionSetupHandler.java | 308 ++++++++++++++++++++-
.../ConsensusSubscriptionWalRetention.java | 163 +++++++++++
.../consensus/ConsensusPrefetchingQueueTest.java | 33 ++-
.../ConsensusSubscriptionCommitStateTest.java | 35 +++
.../ConsensusSubscriptionSetupHandlerTest.java | 47 ++++
.../ConsensusSubscriptionWalRetentionTest.java | 102 +++++++
.../meta/consumer/CommitProgressKeeper.java | 15 +
.../meta/consumer/CommitProgressKeeperTest.java | 19 ++
30 files changed, 1389 insertions(+), 73 deletions(-)
diff --git
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
index 3d107476b85..51b87f910e2 100644
---
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
+++
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
@@ -136,6 +136,14 @@ public class TopicConfig extends PipeParameters {
return TopicConstant.MODE_INCREMENTAL_VALUE.equalsIgnoreCase(getMode());
}
+ public boolean isProgressRetainedAfterUnsubscribe() {
+ return Boolean.parseBoolean(
+ getStringIgnoreCase(
+ TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
+ TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_DEFAULT_VALUE)
+ .trim());
+ }
+
/**
* @deprecated Use {@link #isInitialMode()}.
*/
diff --git
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
index 4d11f2fe8a2..960e3bab524 100644
---
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
+++
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java
@@ -33,6 +33,9 @@ public class TopicConstant {
public static final String COLUMN_FILTER_KEY = "column-filter";
public static final String RETENTION_BYTES_KEY = "retention.bytes";
public static final String RETENTION_MS_KEY = "retention.ms";
+ public static final String RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY =
+ "retain.progress.after.unsubscribe";
+ public static final String RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_DEFAULT_VALUE =
"false";
public static final String DATABASE_DEFAULT_VALUE = ".*";
public static final String TABLE_DEFAULT_VALUE = ".*";
public static final String COLUMN_FILTER_DEFAULT_VALUE = "true";
diff --git
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
index d308ef123cf..3b32d407eba 100644
---
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
+++
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java
@@ -110,6 +110,18 @@ public class TopicConfigTest {
Assert.assertTrue(new TopicConfig(attributes).isColumnFilterTrivial());
}
+ @Test
+ public void testRetainProgressAfterUnsubscribeDefaultsToFalse() {
+ Assert.assertFalse(new TopicConfig().isProgressRetainedAfterUnsubscribe());
+ }
+
+ @Test
+ public void testRetainProgressAfterUnsubscribeIsCaseInsensitiveAndTrimmed() {
+ Assert.assertTrue(
+ new
TopicConfig(Collections.singletonMap("Retain.Progress.After.Unsubscribe", "
TRUE "))
+ .isProgressRetainedAfterUnsubscribe());
+ }
+
private static TopicConfig topicConfigWithMode(final String mode) {
return new TopicConfig(Collections.singletonMap(TopicConstant.MODE_KEY,
mode));
}
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 0e834e0603f..fb73cc93376 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -617,6 +617,9 @@ public final class ConfigNodeMessages {
public static final String EXCEPTION_NO_SUCH_USER_ID_99CA691B = "No such
user id: ";
public static final String
LOG_HANDLING_COMMIT_PROGRESS_META_CHANGES_FA21A080 = "Handling commit progress
meta changes ...";
public static final String
EXCEPTION_FAILED_CREATE_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_1_POSITIVE_A33070FB
= "Failed to create or alter topic, illegal %s=%s, expected -1 or a positive
long value";
+ public static final String
+
EXCEPTION_FAILED_CREATE_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_TRUE_OR_FALSE_855A6926
=
+ "Failed to create or alter topic, illegal %s=%s, expected true or
false";
public static final String
LOG_TRYING_GET_MAX_TTL_UNDER_ONE_DATABASE_USE_LONG_MAX_9D70ACB2 = " when trying
to get max ttl under one database, use Long.MAX_VALUE.";
public static final String
MESSAGE_ENABLE_IOTDB_CLUSTER_S_DATA_SERVICE_PLEASE_REGISTER_ARG_MORE_F48F3890 =
"To enable IoTDB-Cluster's data service, please register %d more
IoTDB-DataNode";
public static final String
MESSAGE_APPLY_NEW_CONFIGNODE_FAILED_BECAUSE_CURRENT_CONFIGNODE_CAN_T_STORE_1BB6A6BF
= "Apply new ConfigNode failed because current ConfigNode can't store
ConfigNode information.";
@@ -675,6 +678,9 @@ public final class ConfigNodeMessages {
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E
=
"Failed to create or alter topic, %s and %s are only supported for
incremental topics";
+ public static final String
+
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ATTRIBUTES_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_55CB0BF2
=
+ "Failed to create or alter topic, attributes %s are only supported
for incremental topics";
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_SUBSCRIBING_ONLY_TO_THE_AUDIT_DATABASE_OR_PATHS_UNDER_IT_IS_NOT_ALLOWED_3E96A6BA
=
"Failed to create or alter topic, subscribing only to the __audit
database or paths under it is not allowed";
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 9f21686d880..948f4a64765 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -638,6 +638,9 @@ public final class ConfigNodeMessages {
public static final String
EXCEPTION_FAILED_CREATE_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_1_POSITIVE_A33070FB
=
"创建或修改 topic 失败，非法参数 %s=%s，期望为 -1 或正 long 值";
+ public static final String
+
EXCEPTION_FAILED_CREATE_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_TRUE_OR_FALSE_855A6926
=
+ "创建或修改 topic 失败，非法参数 %s=%s，期望为 true 或 false";
public static final String
LOG_TRYING_GET_MAX_TTL_UNDER_ONE_DATABASE_USE_LONG_MAX_9D70ACB2 =
" 尝试获取单个数据库下的最大 ttl 时，使用 Long.MAX_VALUE。";
public static final String
@@ -717,6 +720,9 @@ public final class ConfigNodeMessages {
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E
=
"创建或修改 topic 失败，%s 和 %s 仅支持 incremental 模式的 topic";
+ public static final String
+
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ATTRIBUTES_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_55CB0BF2
=
+ "创建或修改 topic 失败，属性 %s 仅支持 incremental 模式的 topic";
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_SUBSCRIBING_ONLY_TO_THE_AUDIT_DATABASE_OR_PATHS_UNDER_IT_IS_NOT_ALLOWED_3E96A6BA
=
"创建或修改 topic 失败，不允许仅订阅 __audit 数据库或其下的路径";
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
index 6ea5152f4df..9fab5d1f536 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java
@@ -108,6 +108,7 @@ public class SubscriptionInfo implements SnapshotProcessor {
TopicConstant.COLUMN_FILTER_KEY,
TopicConstant.RETENTION_BYTES_KEY,
TopicConstant.RETENTION_MS_KEY,
+ TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
TopicConstant.START_TIME_KEY,
TopicConstant.END_TIME_KEY,
TopicConstant.MODE_KEY,
@@ -129,6 +130,7 @@ public class SubscriptionInfo implements SnapshotProcessor {
TopicConstant.COLUMN_FILTER_KEY,
TopicConstant.RETENTION_BYTES_KEY,
TopicConstant.RETENTION_MS_KEY,
+ TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
TopicConstant.MODE_KEY,
TopicConstant.ORDER_MODE_KEY,
TopicConstant.FORMAT_KEY,
@@ -586,8 +588,10 @@ public class SubscriptionInfo implements SnapshotProcessor
{
private void validateIncrementalTopicRetentionConfig(final TopicConfig
topicConfig)
throws SubscriptionException {
- if (!topicConfig.hasAttribute(TopicConstant.RETENTION_BYTES_KEY)
- && !topicConfig.hasAttribute(TopicConstant.RETENTION_MS_KEY)) {
+ if (!hasTopicAttributeIgnoreCase(topicConfig,
TopicConstant.RETENTION_BYTES_KEY)
+ && !hasTopicAttributeIgnoreCase(topicConfig,
TopicConstant.RETENTION_MS_KEY)
+ && !hasTopicAttributeIgnoreCase(
+ topicConfig, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY))
{
return;
}
@@ -595,24 +599,48 @@ public class SubscriptionInfo implements
SnapshotProcessor {
final String exceptionMessage =
String.format(
ConfigNodeMessages
-
.EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E,
- TopicConstant.RETENTION_BYTES_KEY,
- TopicConstant.RETENTION_MS_KEY);
+
.EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ATTRIBUTES_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_55CB0BF2,
+ List.of(
+ TopicConstant.RETENTION_BYTES_KEY,
+ TopicConstant.RETENTION_MS_KEY,
+ TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY));
LOGGER.warn(exceptionMessage);
throw new SubscriptionException(exceptionMessage);
}
validateRetentionValue(topicConfig, TopicConstant.RETENTION_BYTES_KEY);
validateRetentionValue(topicConfig, TopicConstant.RETENTION_MS_KEY);
+ validateRetainProgressAfterUnsubscribe(topicConfig);
+ }
+
+ private void validateRetainProgressAfterUnsubscribe(final TopicConfig
topicConfig)
+ throws SubscriptionException {
+ if (!hasTopicAttributeIgnoreCase(
+ topicConfig, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY)) {
+ return;
+ }
+ final String value =
+ getTopicAttributeIgnoreCase(
+ topicConfig, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY);
+ if (Objects.isNull(value)
+ || (!Boolean.TRUE.toString().equalsIgnoreCase(value.trim())
+ && !Boolean.FALSE.toString().equalsIgnoreCase(value.trim()))) {
+ throw new SubscriptionException(
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_CREATE_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_TRUE_OR_FALSE_855A6926,
+ TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
+ value));
+ }
}
private void validateRetentionValue(final TopicConfig topicConfig, final
String key)
throws SubscriptionException {
- if (!topicConfig.hasAttribute(key)) {
+ if (!hasTopicAttributeIgnoreCase(topicConfig, key)) {
return;
}
- final String rawValue = topicConfig.getAttribute().get(key);
+ final String rawValue = getTopicAttributeIgnoreCase(topicConfig, key);
try {
final long parsedValue = Long.parseLong(rawValue);
if (parsedValue == 0 || parsedValue < -1) {
@@ -636,6 +664,23 @@ public class SubscriptionInfo implements SnapshotProcessor
{
}
}
+ private static boolean hasTopicAttributeIgnoreCase(
+ final TopicConfig topicConfig, final String expectedKey) {
+ return topicConfig.getAttribute().keySet().stream()
+ .filter(Objects::nonNull)
+ .anyMatch(expectedKey::equalsIgnoreCase);
+ }
+
+ private static String getTopicAttributeIgnoreCase(
+ final TopicConfig topicConfig, final String expectedKey) {
+ return topicConfig.getAttribute().entrySet().stream()
+ .filter(entry -> Objects.nonNull(entry.getKey()))
+ .filter(entry -> expectedKey.equalsIgnoreCase(entry.getKey()))
+ .map(Map.Entry::getValue)
+ .findFirst()
+ .orElse(null);
+ }
+
private void validateUnsupportedHotUpdatedTopicConfig(
final String topicName, final TopicConfig existedConfig, final
TopicConfig updatedConfig)
throws SubscriptionException {
@@ -665,8 +710,8 @@ public class SubscriptionInfo implements SnapshotProcessor {
final String... retentionKeys)
throws SubscriptionException {
for (final String retentionKey : retentionKeys) {
- final String existedValue =
existedConfig.getAttribute().get(retentionKey);
- final String updatedValue =
updatedConfig.getAttribute().get(retentionKey);
+ final String existedValue = getTopicAttributeIgnoreCase(existedConfig,
retentionKey);
+ final String updatedValue = getTopicAttributeIgnoreCase(updatedConfig,
retentionKey);
if (!Objects.equals(existedValue, updatedValue)) {
final String exceptionMessage =
String.format(
@@ -837,15 +882,29 @@ public class SubscriptionInfo implements
SnapshotProcessor {
private TSStatus alterTopicInternal(final AlterTopicPlan plan) {
final boolean isTableModel = plan.getTopicMeta().visibleUnderTableModel();
+ final TopicMeta oldTopicMeta =
+ topicMetaKeeper.getTopicMeta(plan.getTopicMeta().getTopicName(),
isTableModel);
try {
- TopicMeta.validateOwnerProgression(
- topicMetaKeeper.getTopicMeta(plan.getTopicMeta().getTopicName(),
isTableModel),
- plan.getTopicMeta());
+ TopicMeta.validateOwnerProgression(oldTopicMeta, plan.getTopicMeta());
} catch (final IllegalArgumentException e) {
return new
TSStatus(TSStatusCode.SUBSCRIPTION_OWNER_EPOCH_CONFLICT.getStatusCode())
.setMessage(e.getMessage());
}
+ if (Objects.nonNull(oldTopicMeta)
+ && oldTopicMeta.getConfig().isProgressRetainedAfterUnsubscribe()
+ &&
!plan.getTopicMeta().getConfig().isProgressRetainedAfterUnsubscribe()) {
+ for (final ConsumerGroupMeta consumerGroupMeta :
+ consumerGroupMetaKeeper.getAllConsumerGroupMeta()) {
+ if (consumerGroupMeta.visibleUnder(isTableModel)
+ && !consumerGroupMeta.isTopicSubscribedByConsumerGroup(
+ plan.getTopicMeta().getTopicName())) {
+ commitProgressKeeper.removeTopicProgress(
+ consumerGroupMeta.getConsumerGroupId(),
plan.getTopicMeta().getTopicName());
+ }
+ }
+ }
+
topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName(),
isTableModel);
topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(),
plan.getTopicMeta());
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
@@ -879,6 +938,13 @@ public class SubscriptionInfo implements SnapshotProcessor
{
public TSStatus dropTopic(DropTopicPlan plan) {
acquireWriteLock();
try {
+ for (final ConsumerGroupMeta consumerGroupMeta :
+ consumerGroupMetaKeeper.getAllConsumerGroupMeta()) {
+ if (!plan.isTableModelSet() ||
consumerGroupMeta.visibleUnder(plan.isTableModel())) {
+ commitProgressKeeper.removeTopicProgress(
+ consumerGroupMeta.getConsumerGroupId(), plan.getTopicName());
+ }
+ }
if (plan.isTableModelSet()) {
topicMetaKeeper.removeTopicMeta(plan.getTopicName(),
plan.isTableModel());
} else {
@@ -1052,10 +1118,21 @@ public class SubscriptionInfo implements
SnapshotProcessor {
? consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId)
: null;
if (Objects.nonNull(currentConsumerGroupMeta)) {
- ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta,
consumerGroupMeta)
- .forEach(
- topicName ->
-
commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName));
+ if (consumerGroupMeta.isEmpty()) {
+ commitProgressKeeper.removeConsumerGroupProgress(consumerGroupId);
+ } else {
+ ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta,
consumerGroupMeta)
+ .forEach(
+ topicName -> {
+ final TopicMeta topicMeta =
+ topicMetaKeeper.getTopicMeta(
+ topicName,
currentConsumerGroupMeta.visibleUnder(true));
+ if (Objects.isNull(topicMeta)
+ ||
!topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) {
+
commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName);
+ }
+ });
+ }
}
consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId);
if (!consumerGroupMeta.isEmpty()) {
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
index 170bbe2d3e1..087cb5868f5 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
@@ -251,6 +251,13 @@ public class SubscriptionInfoTest {
final String retainedTopic = "retained-topic";
final String regionId = "DataRegion[1]";
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(
+ new TopicMeta(
+ removedTopic,
+ 1L,
+ Collections.singletonMap(
+ TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE))));
final ConsumerGroupMeta currentMeta =
new ConsumerGroupMeta(
consumerGroupId, 1, new ConsumerMeta(consumerId, 1,
Collections.emptyMap()));
@@ -277,6 +284,124 @@ public class SubscriptionInfoTest {
Assert.assertNotNull(keeper.getRegionProgress(retainedKey));
}
+ @Test
+ public void
testAlterConsumerGroupRetainsUnsubscribedTopicProgressWhenConfigured() {
+ final String consumerGroupId = "consumer-group";
+ final String consumerId = "consumer";
+ final String topicName = "retained-topic";
+ final String regionId = "DataRegion[1]";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE);
+ topicAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"true");
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(new TopicMeta(topicName, 1L, topicAttributes)));
+
+ final ConsumerGroupMeta currentMeta =
+ new ConsumerGroupMeta(
+ consumerGroupId, 1, new ConsumerMeta(consumerId, 1,
Collections.emptyMap()));
+ currentMeta.addSubscription(consumerId, Collections.singleton(topicName));
+ subscriptionInfo.alterConsumerGroup(new
AlterConsumerGroupPlan(currentMeta));
+
+ final CommitProgressKeeper keeper =
subscriptionInfo.getCommitProgressKeeper();
+ final String progressKey =
+ CommitProgressKeeper.generateKey(consumerGroupId, topicName, regionId,
1);
+ keeper.updateRegionProgress(progressKey, ByteBuffer.wrap(new byte[] {1}));
+
+ final ConsumerGroupMeta updatedMeta = currentMeta.deepCopy();
+ updatedMeta.removeSubscription(consumerId,
Collections.singleton(topicName));
+ subscriptionInfo.alterConsumerGroup(new
AlterConsumerGroupPlan(updatedMeta));
+
+ Assert.assertNotNull(keeper.getRegionProgress(progressKey));
+ }
+
+ @Test
+ public void testAlterTopicDisablingRetentionRemovesUnsubscribedProgress() {
+ final String consumerGroupId = "consumer-group";
+ final String consumerId = "consumer";
+ final String topicName = "retained-topic";
+ final String regionId = "DataRegion[1]";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE);
+ topicAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"true");
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(new TopicMeta(topicName, 1L, topicAttributes)));
+
+ final ConsumerGroupMeta currentMeta =
+ new ConsumerGroupMeta(
+ consumerGroupId, 1, new ConsumerMeta(consumerId, 1,
Collections.emptyMap()));
+ currentMeta.addSubscription(consumerId, Collections.singleton(topicName));
+ subscriptionInfo.alterConsumerGroup(new
AlterConsumerGroupPlan(currentMeta));
+
+ final CommitProgressKeeper keeper =
subscriptionInfo.getCommitProgressKeeper();
+ final String progressKey =
+ CommitProgressKeeper.generateKey(consumerGroupId, topicName, regionId,
1);
+ keeper.updateRegionProgress(progressKey, ByteBuffer.wrap(new byte[] {1}));
+
+ final ConsumerGroupMeta unsubscribedMeta = currentMeta.deepCopy();
+ unsubscribedMeta.removeSubscription(consumerId,
Collections.singleton(topicName));
+ subscriptionInfo.alterConsumerGroup(new
AlterConsumerGroupPlan(unsubscribedMeta));
+ Assert.assertNotNull(keeper.getRegionProgress(progressKey));
+
+ final Map<String, String> updatedAttributes = new
HashMap<>(topicAttributes);
+ updatedAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"false");
+ subscriptionInfo.alterTopic(
+ new AlterTopicPlan(new TopicMeta(topicName, 2L, updatedAttributes)));
+
+ Assert.assertNull(keeper.getRegionProgress(progressKey));
+ }
+
+ @Test
+ public void testDropTopicRemovesRetainedProgress() {
+ final String consumerGroupId = "consumer-group";
+ final String consumerId = "consumer";
+ final String topicName = "retained-topic";
+ final String regionId = "DataRegion[1]";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE);
+ topicAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"true");
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(new TopicMeta(topicName, 1L, topicAttributes)));
+
+ final ConsumerGroupMeta currentMeta =
+ new ConsumerGroupMeta(
+ consumerGroupId, 1, new ConsumerMeta(consumerId, 1,
Collections.emptyMap()));
+ subscriptionInfo.alterConsumerGroup(new
AlterConsumerGroupPlan(currentMeta));
+ final CommitProgressKeeper keeper =
subscriptionInfo.getCommitProgressKeeper();
+ final String progressKey =
+ CommitProgressKeeper.generateKey(consumerGroupId, topicName, regionId,
1);
+ keeper.updateRegionProgress(progressKey, ByteBuffer.wrap(new byte[] {1}));
+
+ subscriptionInfo.dropTopic(new DropTopicPlan(topicName, false));
+
+ Assert.assertNull(keeper.getRegionProgress(progressKey));
+ }
+
+ @Test
+ public void testDropConsumerGroupRemovesAllRetainedProgress() {
+ final String consumerGroupId = "consumer-group";
+ final String consumerId = "consumer";
+ final String topicName = "retained-topic";
+ final String regionId = "DataRegion[1]";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final ConsumerGroupMeta currentMeta =
+ new ConsumerGroupMeta(
+ consumerGroupId, 1, new ConsumerMeta(consumerId, 1,
Collections.emptyMap()));
+ subscriptionInfo.alterConsumerGroup(new
AlterConsumerGroupPlan(currentMeta));
+ final CommitProgressKeeper keeper =
subscriptionInfo.getCommitProgressKeeper();
+ final String progressKey =
+ CommitProgressKeeper.generateKey(consumerGroupId, topicName, regionId,
1);
+ keeper.updateRegionProgress(progressKey, ByteBuffer.wrap(new byte[] {1}));
+
+ final ConsumerGroupMeta emptyMeta = currentMeta.deepCopy();
+ emptyMeta.removeConsumer(consumerId);
+ subscriptionInfo.alterConsumerGroup(new AlterConsumerGroupPlan(emptyMeta));
+
+ Assert.assertNull(keeper.getRegionProgress(progressKey));
+ }
+
private TopicMeta createTopicMeta(
final String topicName, final String ownerId, final long ownerEpoch) {
final Map<String, String> topicAttributes = new HashMap<>();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
index 48523171345..da9c356190c 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java
@@ -325,6 +325,21 @@ public class SubscriptionInfoTopicValidationTest {
}
}
+ @Test
+ public void testAcceptAlteringRetainProgressAfterUnsubscribe() throws
Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> originalAttributes =
newIncrementalTableTopicAttributes();
+
originalAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"false");
+ subscriptionInfo.createTopic(
+ new CreateTopicPlan(new TopicMeta("table_topic", 1L,
originalAttributes)));
+
+ final Map<String, String> updatedAttributes =
newIncrementalTableTopicAttributes();
+ updatedAttributes.put("Retain.Progress.After.Unsubscribe", "true");
+
+ subscriptionInfo.validateBeforeAlteringTopic(
+ new TopicMeta("table_topic", 2L, updatedAttributes));
+ }
+
@Test
public void testRejectIllegalMode() {
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
@@ -385,6 +400,35 @@ public class SubscriptionInfoTopicValidationTest {
assertCreateRejected(subscriptionInfo, attributes, "only supported for
incremental topics");
}
+ @Test
+ public void testAcceptRetainingProgressOnIncrementalTopic() throws Exception
{
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes =
newIncrementalTableTopicAttributes();
+ attributes.put("Retain.Progress.After.Unsubscribe", " TRUE ");
+
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new
TCreateTopicReq("table_topic").setTopicAttributes(attributes)));
+ }
+
+ @Test
+ public void testRejectRetainingProgressOnInitialTopic() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes = newInitialTableTopicAttributes();
+ attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"true");
+
+ assertCreateRejected(subscriptionInfo, attributes, "only supported for
incremental topics");
+ }
+
+ @Test
+ public void testRejectIllegalRetainProgressValue() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> attributes =
newIncrementalTableTopicAttributes();
+ attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "yes");
+
+ assertCreateRejected(subscriptionInfo, attributes, "expected true or
false");
+ }
+
@Test
public void testRejectOwnerLeaseDurationBelowMin() {
final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index 9c0692132f0..8ee9ff96b7a 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -116,6 +116,12 @@ public class IoTConsensus implements IConsensus {
*/
public static volatile BiConsumer<ConsensusGroupId, IoTConsensusServerImpl>
onNewPeerCreated;
+ /**
+ * Optional callback invoked after a recovered local peer is constructed and
before it starts.
+ * Used by the subscription system to restore WAL-retention guards before
WAL cleanup is enabled.
+ */
+ public static volatile BiConsumer<ConsensusGroupId, IoTConsensusServerImpl>
onPeerRecovering;
+
/**
* Optional callback invoked before a local peer is deleted via {@link
#deleteLocalPeer}. Used by
* the subscription system to unbind and clean up prefetching queues before
the region is removed.
@@ -244,7 +250,14 @@ public class IoTConsensus implements IConsensus {
consensusGroupId ->
resetPeerListWithoutThrow.accept(consensusGroupId,
Collections.emptyList()));
}
- stateMachineMap.values().forEach(IoTConsensusServerImpl::start);
+ stateMachineMap.forEach(
+ (consensusGroupId, consensus) -> {
+ final BiConsumer<ConsensusGroupId, IoTConsensusServerImpl> callback
= onPeerRecovering;
+ if (callback != null) {
+ callback.accept(consensusGroupId, consensus);
+ }
+ consensus.start();
+ });
}
@Override
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
index ba4ad87b1fd..8e70866e379 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
@@ -359,7 +359,7 @@ public class IoTConsensusServerImpl {
subscriptionQueueRegistry.offer(indexedConsensusRequest);
logDispatcher.offer(indexedConsensusRequest, offeredToSubscription);
if (!offeredToSubscription
- && subscriptionQueueRegistry.isEmpty()
+ && !subscriptionQueueRegistry.hasQueues()
&& logger.isDebugEnabled()
&& indexedConsensusRequest.getSearchIndex() % 50 == 0) {
// Log periodically when no subscription queues are registered
@@ -1106,7 +1106,7 @@ public class IoTConsensusServerImpl {
}
public boolean hasSubscriptionConsumers() {
- return !subscriptionQueueRegistry.isEmpty();
+ return subscriptionQueueRegistry.hasQueues();
}
private long assignPhysicalTimeInMs() {
@@ -1280,6 +1280,16 @@ public class IoTConsensusServerImpl {
System.identityHashCode(this));
}
+ public void replaceDetachedSubscriptionRetentionWithQueue(
+ final String retentionId,
+ final BlockingQueue<IndexedConsensusRequest> queue,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier committedRetainedMinVersionIdSupplier) {
+ subscriptionQueueRegistry.replaceDetachedRetentionWithQueue(
+ retentionId, queue, retentionPolicy,
committedRetainedMinVersionIdSupplier);
+ checkAndUpdateSafeDeletedSearchIndex();
+ }
+
public void deregisterSubscriptionQueue(final
BlockingQueue<IndexedConsensusRequest> queue) {
subscriptionQueueRegistry.unregister(queue);
// Re-evaluate: with fewer subscribers, more WAL may be deletable
@@ -1291,6 +1301,30 @@ public class IoTConsensusServerImpl {
subscriptionQueueRegistry.size());
}
+ public void replaceSubscriptionQueueWithDetachedRetention(
+ final BlockingQueue<IndexedConsensusRequest> queue,
+ final String retentionId,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier committedRetainedMinVersionIdSupplier) {
+ subscriptionQueueRegistry.replaceQueueWithDetachedRetention(
+ queue, retentionId, retentionPolicy,
committedRetainedMinVersionIdSupplier);
+ checkAndUpdateSafeDeletedSearchIndex();
+ }
+
+ public void registerDetachedSubscriptionRetention(
+ final String retentionId,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier committedRetainedMinVersionIdSupplier) {
+ subscriptionQueueRegistry.registerDetachedRetention(
+ retentionId, retentionPolicy, committedRetainedMinVersionIdSupplier);
+ checkAndUpdateSafeDeletedSearchIndex();
+ }
+
+ public void deregisterDetachedSubscriptionRetention(final String
retentionId) {
+ subscriptionQueueRegistry.unregisterDetachedRetention(retentionId);
+ checkAndUpdateSafeDeletedSearchIndex();
+ }
+
public long getSyncLag() {
long minSyncIndex = getMinSyncIndex();
return getSearchIndex() - minSyncIndex;
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
index d2de8f93fd9..995b876cbd1 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
@@ -43,6 +43,8 @@ public class SubscriptionQueueRegistry {
private final String consensusGroupId;
private final Map<BlockingQueue<IndexedConsensusRequest>,
SubscriptionQueueRegistration> queues =
new ConcurrentHashMap<>();
+ private final Map<String, SubscriptionQueueRegistration>
detachedRetentionRegistrations =
+ new ConcurrentHashMap<>();
// offer is synchronized, so these counts remain scoped to one reason
without extra atomics.
private final long[] rejectedEntriesByReason =
new long[SubscriptionQueueRejectionReason.values().length];
@@ -79,8 +81,46 @@ public class SubscriptionQueueRegistry {
queues.remove(queue);
}
+ public synchronized void replaceQueueWithDetachedRetention(
+ final BlockingQueue<IndexedConsensusRequest> queue,
+ final String retentionId,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier retainedMinVersionIdSupplier) {
+ queues.remove(queue);
+ detachedRetentionRegistrations.put(
+ retentionId,
+ new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ }
+
+ public synchronized void replaceDetachedRetentionWithQueue(
+ final String retentionId,
+ final BlockingQueue<IndexedConsensusRequest> queue,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier retainedMinVersionIdSupplier) {
+ detachedRetentionRegistrations.remove(retentionId);
+ queues.put(
+ queue, new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ }
+
+ public synchronized void registerDetachedRetention(
+ final String retentionId,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier retainedMinVersionIdSupplier) {
+ detachedRetentionRegistrations.put(
+ retentionId,
+ new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ }
+
+ public synchronized void unregisterDetachedRetention(final String
retentionId) {
+ detachedRetentionRegistrations.remove(retentionId);
+ }
+
public synchronized boolean isEmpty() {
- return queues.isEmpty();
+ return queues.isEmpty() && detachedRetentionRegistrations.isEmpty();
+ }
+
+ public synchronized boolean hasQueues() {
+ return !queues.isEmpty();
}
public synchronized int size() {
@@ -93,6 +133,10 @@ public class SubscriptionQueueRegistry {
for (final SubscriptionQueueRegistration registration : queues.values()) {
retentionPolicies.add(registration.retentionPolicy);
}
+ for (final SubscriptionQueueRegistration registration :
+ detachedRetentionRegistrations.values()) {
+ retentionPolicies.add(registration.retentionPolicy);
+ }
return retentionPolicies;
}
@@ -102,6 +146,10 @@ public class SubscriptionQueueRegistry {
for (final SubscriptionQueueRegistration registration : queues.values())
{
suppliers.add(registration.retainedMinVersionIdSupplier);
}
+ for (final SubscriptionQueueRegistration registration :
+ detachedRetentionRegistrations.values()) {
+ suppliers.add(registration.retainedMinVersionIdSupplier);
+ }
}
final Collection<Long> retainedMinVersionIds = new
ArrayList<>(suppliers.size());
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistryTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistryTest.java
index e9d0f7f3b0f..214cc2ec8f2 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistryTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistryTest.java
@@ -28,6 +28,7 @@ import org.junit.Test;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.atomic.AtomicLong;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -63,6 +64,47 @@ public class SubscriptionQueueRegistryTest {
assertSame(request, queue.poll());
}
+ @Test
+ public void testDetachedRetentionProtectsWalWithoutReceivingRequests() {
+ final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("test");
+ final SubscriptionWalRetentionPolicy policy =
+ new SubscriptionWalRetentionPolicy("topic", 100, 200);
+ registry.registerDetachedRetention("retention", policy, () -> 7L);
+ final IndexedConsensusRequest request = newRequest();
+
+ assertFalse(registry.isEmpty());
+ assertFalse(registry.hasQueues());
+ assertFalse(registry.offer(request));
+ assertTrue(request.getSerializedRequests().isEmpty());
+ assertEquals(Collections.singletonList(policy),
registry.getRetentionPolicies());
+ assertEquals(Collections.singletonList(7L),
registry.getRetainedMinVersionIds());
+ }
+
+ @Test
+ public void testReplaceQueueAndDetachedRetentionAtomically() {
+ final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("test");
+ final InspectingQueue firstQueue = new InspectingQueue();
+ final InspectingQueue secondQueue = new InspectingQueue();
+ final SubscriptionWalRetentionPolicy policy =
+ new SubscriptionWalRetentionPolicy("topic", 100, 200);
+ final AtomicLong retainedVersion = new AtomicLong(3L);
+ registry.register(firstQueue, policy, retainedVersion::get);
+
+ registry.replaceQueueWithDetachedRetention(
+ firstQueue, "retention", policy, retainedVersion::get);
+ assertFalse(registry.hasQueues());
+ assertFalse(registry.isEmpty());
+ assertFalse(registry.offer(newRequest()));
+
+ retainedVersion.set(9L);
+ registry.replaceDetachedRetentionWithQueue(
+ "retention", secondQueue, policy, retainedVersion::get);
+ assertTrue(registry.hasQueues());
+ assertTrue(registry.offer(newRequest()));
+ assertEquals(1, secondQueue.size());
+ assertEquals(Collections.singletonList(9L),
registry.getRetainedMinVersionIds());
+ }
+
private static IndexedConsensusRequest newRequest() {
return new IndexedConsensusRequest(
1, Collections.singletonList(new
ByteBufferConsensusRequest(ByteBuffer.allocate(1))));
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 836ac88d84f..9d400f9d40b 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2692,6 +2692,8 @@ public final class DataNodePipeMessages {
"Topic metadata for %s is unavailable during consensus subscription
setup";
public static final String
EXCEPTION_TOPIC_CONFIG_FOR_ARG_IS_UNAVAILABLE_DURING_CONSENSUS_SUBSCRIPTION_SETUP_B94404EE
=
"Topic config for %s is unavailable during consensus subscription setup";
+ public static final String
LOG_FAILED_TO_RESTORE_DETACHED_CONSENSUS_SUBSCRIPTION_WAL_RETENTION_FOR_CONSUMER_GROUP_ARG_TOPIC_ARG_REGION_ARG_296129FE
=
+ "Failed to restore detached consensus subscription WAL retention for
consumer group {}, topic {}, region {}";
public static final String
LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0
=
"Failed to release TsFile parser memory for Pipe {} (creation time {})
in DataRegion {} because no reservation exists.";
public static final String
LOG_PIPE_PROCESSOR_WORKER_ARG_HAS_BEEN_PROCESSING_THE_SAME_EVENT_FOR_ARG_MS_PIPE_ARG_DATAREGION_ARG_SUBTASK_ARG_EVENT_ARG_THREAD_STATE_ARG_STACK_ARG_63B40775
=
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 5840da12daf..4891920e079 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2517,6 +2517,8 @@ public final class DataNodePipeMessages {
"共识订阅设置期间 topic %s 的元数据不可用";
public static final String
EXCEPTION_TOPIC_CONFIG_FOR_ARG_IS_UNAVAILABLE_DURING_CONSENSUS_SUBSCRIPTION_SETUP_B94404EE
=
"共识订阅设置期间 topic %s 的配置不可用";
+ public static final String
LOG_FAILED_TO_RESTORE_DETACHED_CONSENSUS_SUBSCRIPTION_WAL_RETENTION_FOR_CONSUMER_GROUP_ARG_TOPIC_ARG_REGION_ARG_296129FE
=
+ "为 consumer group {}、topic {}、region {} 恢复 detached consensus
subscription WAL retention 失败";
public static final String
LOG_FAILED_TO_RELEASE_TSFILE_PARSER_MEMORY_FOR_PIPE_ARG_CREATION_TIME_ARG_IN_DATAREGION_ARG_BECAUSE_NO_RESERVATION_EXISTS_BB8321C0
=
"无法释放 Pipe {}（创建时间 {}）在 DataRegion {} 中的 TsFile 解析器内存，因为不存在对应的预留。";
public static final String
LOG_PIPE_PROCESSOR_WORKER_ARG_HAS_BEEN_PROCESSING_THE_SAME_EVENT_FOR_ARG_MS_PIPE_ARG_DATAREGION_ARG_SUBTASK_ARG_EVENT_ARG_THREAD_STATE_ARG_STACK_ARG_63B40775
=
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 8291200b6ca..6bca0cace69 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -120,6 +120,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
import org.apache.iotdb.db.storageengine.load.active.ActiveLoadAgent;
import org.apache.iotdb.db.storageengine.rescon.disk.TierManager;
import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusSubscriptionSetupHandler;
import org.apache.iotdb.db.trigger.executor.TriggerExecutor;
import org.apache.iotdb.db.trigger.service.TriggerInformationUpdater;
import org.apache.iotdb.db.trigger.service.TriggerManagementService;
@@ -300,6 +301,9 @@ public class DataNode extends ServerCommandLine implements
DataNodeMBean {
// TierManager need DataNodeId to do some operations so the reset method
need to be invoked
// after DataNode adding
TierManager.getInstance().resetFolders();
+ if (SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+ ConsensusSubscriptionSetupHandler.ensureNewRegionListenerRegistered();
+ }
// Active DataNode
active();
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 93f64aa0381..1c429cd97dc 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
@@ -670,7 +670,8 @@ public class SubscriptionBrokerAgent {
final RegionProgress fallbackCommittedRegionProgress,
final long tailStartSearchIndex,
final long initialRuntimeVersion,
- final boolean initialActive) {
+ final boolean initialActive,
+ final boolean replaceDetachedRetention) {
final ConsensusPrefetchingQueue queue =
getOrCreateBroker(
consumerGroupId,
@@ -692,7 +693,8 @@ public class SubscriptionBrokerAgent {
fallbackCommittedRegionProgress,
tailStartSearchIndex,
initialRuntimeVersion,
- initialActive);
+ initialActive,
+ replaceDetachedRetention);
prefetchingQueueCount.invalidate();
return queue;
}
@@ -824,7 +826,9 @@ public class SubscriptionBrokerAgent {
}
public void unbindConsensusPrefetchingQueue(
- final String consumerGroupId, final String topicName) {
+ final String consumerGroupId,
+ final String topicName,
+ final boolean retainProgressAfterUnsubscribe) {
final ConsensusSubscriptionBroker broker =
getConsensusBroker(consumerGroupId);
if (Objects.isNull(broker)) {
LOGGER.warn(
@@ -833,7 +837,7 @@ public class SubscriptionBrokerAgent {
consumerGroupId);
return;
}
- broker.unbindConsensusPrefetchingQueue(topicName);
+ broker.unbindConsensusPrefetchingQueue(topicName,
retainProgressAfterUnsubscribe);
prefetchingQueueCount.invalidate();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
index 79f0c59ec7e..7917fa39a5c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java
@@ -118,6 +118,7 @@ public class SubscriptionConsumerAgent {
// if the creation time of consumer group meta on local agent is
inconsistent with meta from
// coordinator
if (metaInAgent.getCreationTime() !=
metaFromCoordinator.getCreationTime()) {
+ cleanupConsumerGroupSubscriptions(consumerGroupId, metaInAgent);
if (SubscriptionAgent.broker().isBrokerExist(consumerGroupId)) {
LOGGER.warn(
DataNodePipeMessages
@@ -242,6 +243,12 @@ public class SubscriptionConsumerAgent {
}
private void handleDropConsumerGroupInternal(final String consumerGroupId) {
+ final ConsumerGroupMeta consumerGroupMeta =
+ consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId);
+ if (Objects.nonNull(consumerGroupMeta)) {
+ cleanupConsumerGroupSubscriptions(consumerGroupId, consumerGroupMeta);
+ }
+
if (SubscriptionAgent.broker().isBrokerExist(consumerGroupId)) {
if (!SubscriptionAgent.broker().dropBroker(consumerGroupId)) {
final String exceptionMessage =
@@ -320,6 +327,22 @@ public class SubscriptionConsumerAgent {
private int referenceCount = 1;
}
+ private void cleanupConsumerGroupSubscriptions(
+ final String consumerGroupId, final ConsumerGroupMeta consumerGroupMeta)
{
+ final Set<String> consensusTopics = new LinkedHashSet<>();
+ for (final String topicName : consumerGroupMeta.getSubscribedTopicNames())
{
+ if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
+ topicName, consumerGroupMeta.visibleUnder(true))) {
+ consensusTopics.add(topicName);
+ } else {
+ SubscriptionAgent.broker().removePrefetchingQueue(consumerGroupId,
topicName);
+ }
+ }
+ ConsensusSubscriptionSetupHandler.teardownConsensusSubscriptions(
+ consumerGroupId, consensusTopics, false);
+
ConsensusSubscriptionSetupHandler.cleanupRetainedProgressForConsumerGroup(consumerGroupId);
+ }
+
public boolean isConsumerExisted(final String consumerGroupId, final String
consumerId) {
acquireReadLock();
try {
@@ -352,6 +375,25 @@ public class SubscriptionConsumerAgent {
}
}
+ public boolean containsConsumerGroup(final String consumerGroupId) {
+ acquireReadLock();
+ try {
+ return
consumerGroupMetaKeeper.containsConsumerGroupMeta(consumerGroupId);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
+ public boolean isTopicSubscribedByConsumerGroup(
+ final String consumerGroupId, final String topicName) {
+ acquireReadLock();
+ try {
+ return
consumerGroupMetaKeeper.isTopicSubscribedByConsumerGroup(topicName,
consumerGroupId);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
/**
* Get all active subscriptions: consumerGroupId → set of subscribed topic
names. Used by
* consensus subscription auto-binding when a new DataRegion is created.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java
index e942453f7bd..2962f525e70 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionRuntimeAgent.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.service.IService;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusSubscriptionSetupHandler;
import
org.apache.iotdb.db.subscription.task.execution.ConsensusSubscriptionPrefetchExecutorManager;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -72,6 +73,7 @@ public class SubscriptionRuntimeAgent implements IService {
SubscriptionAgentLauncher.launchSubscriptionTopicAgent();
SubscriptionAgentLauncher.launchSubscriptionConsumerAgent();
+ ConsensusSubscriptionSetupHandler.restoreDetachedRetentions();
isShutdown.set(false);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
index 256c87e521d..f4cb98d823e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
@@ -31,6 +31,7 @@ import
org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusSubscriptionSetupHandler;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaRespExceptionMessage;
import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
import org.apache.iotdb.rpc.RpcUtils;
@@ -121,6 +122,8 @@ public class SubscriptionTopicAgent {
SubscriptionAgent.broker()
.refreshConsensusQueueOrderMode(
topicName, isTableModel,
metaFromCoordinator.getConfig().getOrderMode());
+ ConsensusSubscriptionSetupHandler.refreshDetachedRetentionsForTopic(
+ topicName, isTableModel, metaFromCoordinator.getConfig());
}
static boolean shouldRefreshColumnFilter(final TopicMeta oldMeta, final
TopicMeta newMeta) {
@@ -225,6 +228,8 @@ public class SubscriptionTopicAgent {
Objects.isNull(isTableModel)
? topicMetaKeeper.getTopicMeta(topicName)
: topicMetaKeeper.getTopicMeta(topicName, isTableModel);
+ ConsensusSubscriptionSetupHandler.cleanupRetainedProgressForDroppedTopic(
+ topicName, Objects.nonNull(topicMeta) ?
topicMeta.visibleUnderTableModel() : isTableModel);
if (Objects.isNull(isTableModel)) {
topicMetaKeeper.removeTopicMeta(topicName);
} else {
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 2792dff39e2..cd7abf043cf 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
@@ -679,7 +679,8 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
final RegionProgress fallbackCommittedRegionProgress,
final long tailStartSearchIndex,
final long initialRuntimeVersion,
- final boolean initialActive) {
+ final boolean initialActive,
+ final boolean replaceDetachedRetention) {
synchronized (queueLifecycleLock) {
// Get or create the list of queues for this topic
final List<ConsensusPrefetchingQueue> queues =
@@ -713,7 +714,8 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
fallbackCommittedRegionProgress,
tailStartSearchIndex,
initialRuntimeVersion,
- initialActive);
+ initialActive,
+ replaceDetachedRetention);
queues.add(consensusQueue);
LOGGER.info(
DataNodePipeMessages
@@ -742,16 +744,16 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
}
}
- public void unbindConsensusPrefetchingQueue(final String topicName) {
- // Subscription setup teardown and rollback manage commit state after this
call returns. Let
- // queue close persist the current state, while avoiding an asynchronous
close deleting state
- // restored by a concurrent setup rollback.
- closeAndRemoveConsensusPrefetchingQueues(topicName, true, false);
+ public void unbindConsensusPrefetchingQueue(
+ final String topicName, final boolean retainProgressAfterUnsubscribe) {
+ // Subscription setup teardown and rollback manage commit state after this
call returns.
+ // Persist it on close so asynchronous cleanup cannot delete state
restored by a rollback.
+ closeAndRemoveConsensusPrefetchingQueues(topicName, true,
retainProgressAfterUnsubscribe);
}
@Override
public void unbind(final String topicName) {
- unbindConsensusPrefetchingQueue(topicName);
+ unbindConsensusPrefetchingQueue(topicName, false);
}
public int unbindByRegion(final ConsensusGroupId regionId) {
@@ -856,7 +858,9 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
}
private void closeAndRemoveConsensusPrefetchingQueues(
- final String topicName, final boolean warnIfMissing, final boolean
removeProgressAfterClose) {
+ final String topicName,
+ final boolean warnIfMissing,
+ final boolean retainProgressAfterUnsubscribe) {
final List<ConsensusPrefetchingQueue> queuesToClose;
synchronized (queueLifecycleLock) {
final List<ConsensusPrefetchingQueue> queues =
@@ -877,7 +881,7 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
}
for (final ConsensusPrefetchingQueue q : queuesToClose) {
- q.closeAsync(removeProgressAfterClose);
+ q.closeAsync(false, retainProgressAfterUnsubscribe);
}
LOGGER.info(
DataNodePipeMessages
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 a678646a735..c19420c130c 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
@@ -617,6 +617,36 @@ public class ConsensusPrefetchingQueue {
final long tailStartSearchIndex,
final long initialRuntimeVersion,
final boolean initialActive) {
+ this(
+ consumerGroupId,
+ topicName,
+ orderMode,
+ consensusGroupId,
+ serverImpl,
+ retentionPolicy,
+ converter,
+ commitManager,
+ fallbackCommittedRegionProgress,
+ tailStartSearchIndex,
+ initialRuntimeVersion,
+ initialActive,
+ false);
+ }
+
+ public ConsensusPrefetchingQueue(
+ final String consumerGroupId,
+ final String topicName,
+ final String orderMode,
+ final ConsensusGroupId consensusGroupId,
+ final IoTConsensusServerImpl serverImpl,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final ConsensusLogToTabletConverter converter,
+ final ConsensusSubscriptionCommitManager commitManager,
+ final RegionProgress fallbackCommittedRegionProgress,
+ final long tailStartSearchIndex,
+ final long initialRuntimeVersion,
+ final boolean initialActive,
+ final boolean replaceDetachedRetention) {
this.consumerGroupId = consumerGroupId;
this.topicName = topicName;
this.consensusGroupId = consensusGroupId;
@@ -649,8 +679,17 @@ public class ConsensusPrefetchingQueue {
this::requestPrefetchForRealtimeEntry,
this::getRealtimeAdmissionRejectionReason,
this::recordRealtimeAdmissionRejection);
- serverImpl.registerSubscriptionQueue(
- pendingEntries, retentionPolicy,
this::getRequiredRetainedMinVersionId);
+ if (replaceDetachedRetention) {
+ serverImpl.replaceDetachedSubscriptionRetentionWithQueue(
+ ConsensusSubscriptionWalRetention.generateRetentionId(
+ consumerGroupId, topicName, consensusGroupId),
+ pendingEntries,
+ retentionPolicy,
+ this::getRequiredRetainedMinVersionId);
+ } else {
+ serverImpl.registerSubscriptionQueue(
+ pendingEntries, retentionPolicy,
this::getRequiredRetainedMinVersionId);
+ }
LOGGER.info(
DataNodePipeMessages
@@ -4191,10 +4230,17 @@ public class ConsensusPrefetchingQueue {
}
public void close(final boolean removeProgressAfterClose) {
+ close(removeProgressAfterClose, false);
+ }
+
+ public void close(
+ final boolean removeProgressAfterClose, final boolean
retainProgressAfterUnsubscribe) {
if (!lock.writeLock().tryLock()) {
if (beginClose()) {
CLOSE_CLEANUP_EXECUTOR.execute(
- () -> finishCloseAfterWriteLock(removeProgressAfterClose, true));
+ () ->
+ finishCloseAfterWriteLock(
+ removeProgressAfterClose, retainProgressAfterUnsubscribe,
true));
}
return;
}
@@ -4211,7 +4257,8 @@ public class ConsensusPrefetchingQueue {
}
try {
- finishCloseRequest(seekRequestToFail, removeProgressAfterClose, false);
+ finishCloseRequest(
+ seekRequestToFail, removeProgressAfterClose,
retainProgressAfterUnsubscribe, false);
} catch (final Exception e) {
handleCloseFailure(e);
}
@@ -4219,9 +4266,16 @@ public class ConsensusPrefetchingQueue {
/** Starts teardown without waiting for an active prefetch round or
consensus queue writes. */
public void closeAsync(final boolean removeProgressAfterClose) {
+ closeAsync(removeProgressAfterClose, false);
+ }
+
+ public void closeAsync(
+ final boolean removeProgressAfterClose, final boolean
retainProgressAfterUnsubscribe) {
if (beginClose()) {
CLOSE_CLEANUP_EXECUTOR.execute(
- () -> finishCloseAfterWriteLock(removeProgressAfterClose, true));
+ () ->
+ finishCloseAfterWriteLock(
+ removeProgressAfterClose, retainProgressAfterUnsubscribe,
true));
}
}
@@ -4238,7 +4292,9 @@ public class ConsensusPrefetchingQueue {
}
private void finishCloseAfterWriteLock(
- final boolean removeProgressAfterClose, final boolean backgroundCleanup)
{
+ final boolean removeProgressAfterClose,
+ final boolean retainProgressAfterUnsubscribe,
+ final boolean backgroundCleanup) {
try {
final PendingSeekRequest seekRequestToFail;
acquireWriteLock();
@@ -4248,7 +4304,11 @@ public class ConsensusPrefetchingQueue {
} finally {
releaseWriteLock();
}
- finishCloseRequest(seekRequestToFail, removeProgressAfterClose,
backgroundCleanup);
+ finishCloseRequest(
+ seekRequestToFail,
+ removeProgressAfterClose,
+ retainProgressAfterUnsubscribe,
+ backgroundCleanup);
} catch (final Exception e) {
handleCloseFailure(e);
}
@@ -4257,6 +4317,7 @@ public class ConsensusPrefetchingQueue {
private void finishCloseRequest(
final PendingSeekRequest seekRequestToFail,
final boolean removeProgressAfterClose,
+ final boolean retainProgressAfterUnsubscribe,
final boolean backgroundCleanup) {
final Pair<ConsensusSubscriptionPrefetchExecutor,
ConsensusPrefetchSubtask> prefetchBinding =
detachPrefetchSubtask();
@@ -4278,22 +4339,26 @@ public class ConsensusPrefetchingQueue {
&& Objects.nonNull(prefetchBinding.right)
&& prefetchBinding.right.isScheduledOrRunning()) {
CLOSE_CLEANUP_EXECUTOR.execute(
- () -> finishCloseInBackground(prefetchBinding,
removeProgressAfterClose));
+ () ->
+ finishCloseInBackground(
+ prefetchBinding, removeProgressAfterClose,
retainProgressAfterUnsubscribe));
return;
}
if (backgroundCleanup) {
- finishCloseInBackground(prefetchBinding, removeProgressAfterClose);
+ finishCloseInBackground(
+ prefetchBinding, removeProgressAfterClose,
retainProgressAfterUnsubscribe);
} else {
- finishClose(prefetchBinding, removeProgressAfterClose);
+ finishClose(prefetchBinding, removeProgressAfterClose,
retainProgressAfterUnsubscribe);
}
}
private void finishCloseInBackground(
final Pair<ConsensusSubscriptionPrefetchExecutor,
ConsensusPrefetchSubtask> prefetchBinding,
- final boolean removeProgressAfterClose) {
+ final boolean removeProgressAfterClose,
+ final boolean retainProgressAfterUnsubscribe) {
try {
- finishClose(prefetchBinding, removeProgressAfterClose);
+ finishClose(prefetchBinding, removeProgressAfterClose,
retainProgressAfterUnsubscribe);
} catch (final Exception e) {
handleCloseFailure(e);
}
@@ -4312,12 +4377,13 @@ public class ConsensusPrefetchingQueue {
private void finishClose(
final Pair<ConsensusSubscriptionPrefetchExecutor,
ConsensusPrefetchSubtask> prefetchBinding,
- final boolean removeProgressAfterClose) {
+ final boolean removeProgressAfterClose,
+ final boolean retainProgressAfterUnsubscribe) {
// Queue deregistration may wait for consensus-side WAL/index work.
Metadata callers use
// closeAsync so this cleanup cannot hold up the consumer-group metadata
push.
// Requests left in pendingEntries have not advanced subscription progress
and can be replayed
// from WAL if this queue is created again.
- deregisterPendingEntriesFromConsensusServer();
+
deregisterPendingEntriesFromConsensusServer(retainProgressAfterUnsubscribe);
if (Objects.nonNull(prefetchBinding.right)) {
prefetchBinding.right.awaitIdle();
@@ -4362,9 +4428,24 @@ public class ConsensusPrefetchingQueue {
}
}
- private void deregisterPendingEntriesFromConsensusServer() {
+ private void deregisterPendingEntriesFromConsensusServer(
+ final boolean retainProgressAfterUnsubscribe) {
try {
- serverImpl.deregisterSubscriptionQueue(pendingEntries);
+ if (retainProgressAfterUnsubscribe) {
+ serverImpl.replaceSubscriptionQueueWithDetachedRetention(
+ pendingEntries,
+ ConsensusSubscriptionWalRetention.generateRetentionId(
+ consumerGroupId, topicName, consensusGroupId),
+ retentionPolicy,
+
ConsensusSubscriptionWalRetention.createCommittedRetainedMinVersionIdSupplier(
+ consensusGroupId,
+ () -> consensusReqReader,
+ () ->
+ commitManager.getCommittedRegionProgress(
+ consumerGroupId, topicName, consensusGroupId)));
+ } else {
+ serverImpl.deregisterSubscriptionQueue(pendingEntries);
+ }
} catch (final Exception e) {
LOGGER.warn(
DataNodePipeMessages.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_DURING_DEREGISTER_34C332E7,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
index 14e50ae44d0..ed49dda3f46 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java
@@ -68,6 +68,7 @@ import java.nio.file.StandardOpenOption;
import java.util.Base64;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -725,6 +726,46 @@ public class ConsensusSubscriptionCommitManager {
}
}
+ public void removeAllStatesForConsumerGroup(final String consumerGroupId) {
+ recoverAllTopicStatesIfNeeded();
+ final Set<String> topicNames = new HashSet<>();
+ for (final CommitStateKey stateKey : commitStateKeys.values()) {
+ if (Objects.equals(stateKey.consumerGroupId, consumerGroupId)) {
+ topicNames.add(stateKey.topicName);
+ }
+ }
+ for (final String topicName : topicNames) {
+ removeAllStatesForTopic(consumerGroupId, topicName);
+ }
+ }
+
+ public void removeAllStatesForRegion(final ConsensusGroupId regionId) {
+ final String regionIdStr = regionId.toString();
+ for (final PersistedStateKey stateKey : getPersistedStateKeys()) {
+ if (Objects.equals(stateKey.regionId, regionIdStr)) {
+ removeState(
+ stateKey.consumerGroupId,
+ stateKey.topicName,
+ ConsensusGroupId.Factory.createFromString(stateKey.regionId));
+ }
+ }
+ }
+
+ public List<PersistedStateKey> getPersistedStateKeys() {
+ recoverAllTopicStatesIfNeeded();
+ final List<PersistedStateKey> result = new java.util.ArrayList<>();
+ for (final CommitStateKey stateKey : commitStateKeys.values()) {
+ if (topicProgressIndexes
+ .getOrDefault(stateKey.topicFileKey, Collections.emptyMap())
+ .containsKey(stateKey.regionIdStr)) {
+ result.add(
+ new PersistedStateKey(
+ stateKey.consumerGroupId, stateKey.topicName,
stateKey.regionIdStr));
+ }
+ }
+ return Collections.unmodifiableList(result);
+ }
+
public void resetState(
final String consumerGroupId,
final String topicName,
@@ -1487,6 +1528,13 @@ public class ConsensusSubscriptionCommitManager {
this.consumerGroupId = consumerGroupId;
this.topicSnapshots = topicSnapshots;
}
+
+ Set<String> getRegionIds(final String topicName) {
+ final TopicSetupSnapshot topicSnapshot = topicSnapshots.get(topicName);
+ return Objects.isNull(topicSnapshot)
+ ? Collections.emptySet()
+ : Collections.unmodifiableSet(topicSnapshot.stateSnapshots.keySet());
+ }
}
private static final class TopicSetupSnapshot {
@@ -1559,6 +1607,34 @@ public class ConsensusSubscriptionCommitManager {
}
}
+ public static final class PersistedStateKey {
+
+ private final String consumerGroupId;
+
+ private final String topicName;
+
+ private final String regionId;
+
+ private PersistedStateKey(
+ final String consumerGroupId, final String topicName, final String
regionId) {
+ this.consumerGroupId = consumerGroupId;
+ this.topicName = topicName;
+ this.regionId = regionId;
+ }
+
+ public String getConsumerGroupId() {
+ return consumerGroupId;
+ }
+
+ public String getTopicName() {
+ return topicName;
+ }
+
+ public String getRegionId() {
+ return regionId;
+ }
+ }
+
private static final class TopicProgressSnapshot {
private final Map<String, ConsensusSubscriptionCommitState> states;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
index 87b3c66f698..92cfade03c2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java
@@ -119,6 +119,9 @@ public class ConsensusSubscriptionSetupHandler {
* region, and when a DataRegion is removed, all subscription queues are
properly cleaned up.
*/
public static void ensureNewRegionListenerRegistered() {
+ if (IoTConsensus.onPeerRecovering == null) {
+ IoTConsensus.onPeerRecovering =
ConsensusSubscriptionSetupHandler::onRecoveredRegionCreated;
+ }
if (IoTConsensus.onNewPeerCreated == null) {
IoTConsensus.onNewPeerCreated =
ConsensusSubscriptionSetupHandler::onNewRegionCreated;
LOGGER.info(
@@ -133,6 +136,37 @@ public class ConsensusSubscriptionSetupHandler {
}
}
+ private static void onRecoveredRegionCreated(
+ final ConsensusGroupId groupId, final IoTConsensusServerImpl serverImpl)
{
+ if (!(groupId instanceof DataRegionId)) {
+ return;
+ }
+ registerStartupDetachedRetentions(
+ groupId, serverImpl, ConsensusSubscriptionCommitManager.getInstance());
+ }
+
+ static void registerStartupDetachedRetentions(
+ final ConsensusGroupId groupId,
+ final IoTConsensusServerImpl serverImpl,
+ final ConsensusSubscriptionCommitManager commitManager) {
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
+ commitManager.getPersistedStateKeys()) {
+ if (!groupId.toString().equals(stateKey.getRegionId())) {
+ continue;
+ }
+ ConsensusSubscriptionWalRetention.registerDetached(
+ stateKey.getConsumerGroupId(),
+ stateKey.getTopicName(),
+ groupId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ stateKey.getTopicName(),
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ commitManager);
+ }
+ }
+
/**
* Callback invoked when a new DataRegion (IoTConsensusServerImpl) is
created locally. Queries
* existing subscription metadata to find all active consensus subscriptions
and binds prefetching
@@ -254,7 +288,8 @@ public class ConsensusSubscriptionSetupHandler {
committedRegionProgress,
tailStartSearchIndex,
initialRuntimeVersion,
- initialActive);
+ initialActive,
+ hasLocalPersistedState);
SubscriptionAgent.broker().applyRuntimeStateForRegion(groupId,
initialRuntimeState);
} catch (final Exception e) {
LOGGER.error(
@@ -287,6 +322,21 @@ public class ConsensusSubscriptionSetupHandler {
groupId);
try {
SubscriptionAgent.broker().unbindByRegion(groupId);
+ final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
+ if (dataRegionConsensus instanceof IoTConsensus) {
+ final IoTConsensusServerImpl serverImpl =
+ ((IoTConsensus) dataRegionConsensus).getImpl(groupId);
+ if (Objects.nonNull(serverImpl)) {
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey
stateKey :
+
ConsensusSubscriptionCommitManager.getInstance().getPersistedStateKeys()) {
+ if (groupId.toString().equals(stateKey.getRegionId())) {
+ ConsensusSubscriptionWalRetention.unregisterDetached(
+ stateKey.getConsumerGroupId(), stateKey.getTopicName(),
groupId, serverImpl);
+ }
+ }
+ }
+ }
+
ConsensusSubscriptionCommitManager.getInstance().removeAllStatesForRegion(groupId);
} catch (final Exception e) {
LOGGER.error(
DataNodePipeMessages
@@ -392,7 +442,8 @@ public class ConsensusSubscriptionSetupHandler {
RuntimeException rollbackFailure = null;
for (final String topicName : attemptedTopicNames) {
try {
-
SubscriptionAgent.broker().unbindConsensusPrefetchingQueue(consumerGroupId,
topicName);
+ SubscriptionAgent.broker()
+ .unbindConsensusPrefetchingQueue(consumerGroupId, topicName,
false);
} catch (final RuntimeException e) {
if (Objects.isNull(rollbackFailure)) {
rollbackFailure = e;
@@ -403,6 +454,8 @@ public class ConsensusSubscriptionSetupHandler {
}
try {
commitManager.restoreSetupSnapshot(setupSnapshot, attemptedTopicNames);
+ restoreStartupDetachedRetentions(
+ consumerGroupId, attemptedTopicNames, setupSnapshot, commitManager);
} catch (final RuntimeException e) {
if (Objects.isNull(rollbackFailure)) {
rollbackFailure = e;
@@ -415,6 +468,36 @@ public class ConsensusSubscriptionSetupHandler {
}
}
+ private static void restoreStartupDetachedRetentions(
+ final String consumerGroupId,
+ final Set<String> attemptedTopicNames,
+ final ConsensusSubscriptionCommitManager.SetupSnapshot setupSnapshot,
+ final ConsensusSubscriptionCommitManager commitManager) {
+ final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
+ if (!(dataRegionConsensus instanceof IoTConsensus)) {
+ return;
+ }
+ final IoTConsensus ioTConsensus = (IoTConsensus) dataRegionConsensus;
+ for (final String topicName : attemptedTopicNames) {
+ for (final String regionId : setupSnapshot.getRegionIds(topicName)) {
+ final ConsensusGroupId groupId =
ConsensusGroupId.Factory.createFromString(regionId);
+ final IoTConsensusServerImpl serverImpl =
ioTConsensus.getImpl(groupId);
+ if (Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.registerDetached(
+ consumerGroupId,
+ topicName,
+ groupId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ topicName,
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ commitManager);
+ }
+ }
+ }
+ }
+
static void setupConsensusTopics(
final String consumerGroupId,
final Set<String> topicNames,
@@ -604,7 +687,8 @@ public class ConsensusSubscriptionSetupHandler {
committedRegionProgress,
tailStartSearchIndex,
initialRuntimeVersion,
- initialActive);
+ initialActive,
+ hasLocalPersistedState);
commitManager.initializeStateFromTailProposal(
consumerGroupId, topicName, groupId,
queue.computeTailRegionProgress());
@@ -698,16 +782,20 @@ public class ConsensusSubscriptionSetupHandler {
private static long resolveRetentionValue(
final TopicConfig topicConfig, final String key, final long
defaultValue) {
- if (!topicConfig.hasAttribute(key)) {
+ final String rawValue =
+ topicConfig.getAttribute().entrySet().stream()
+ .filter(entry -> Objects.nonNull(entry.getKey()))
+ .filter(entry -> key.equalsIgnoreCase(entry.getKey()))
+ .map(Map.Entry::getValue)
+ .findFirst()
+ .orElse(null);
+ if (Objects.isNull(rawValue)) {
return normalizeRetentionValue(defaultValue);
}
- final long parsedValue =
Long.parseLong(topicConfig.getAttribute().get(key));
+ final long parsedValue = Long.parseLong(rawValue);
if (parsedValue == 0 || parsedValue <
SubscriptionWalRetentionPolicy.UNBOUNDED) {
throw new IllegalArgumentException(
- String.format(
- DataNodePipeMessages.PIPE_EXCEPTION_ILLEGAL_S_S_72D743AA,
- key,
- topicConfig.getAttribute().get(key)));
+
String.format(DataNodePipeMessages.PIPE_EXCEPTION_ILLEGAL_S_S_72D743AA, key,
rawValue));
}
return normalizeRetentionValue(parsedValue);
}
@@ -718,13 +806,32 @@ public class ConsensusSubscriptionSetupHandler {
public static void teardownConsensusSubscriptions(
final String consumerGroupId, final Set<String> topicNames) {
+ teardownConsensusSubscriptions(consumerGroupId, topicNames, true);
+ }
+
+ public static void teardownConsensusSubscriptions(
+ final String consumerGroupId,
+ final Set<String> topicNames,
+ final boolean retainConfiguredProgress) {
for (final String topicName : topicNames) {
try {
-
SubscriptionAgent.broker().unbindConsensusPrefetchingQueue(consumerGroupId,
topicName);
-
- // Clean up commit state for all regions of this topic
- ConsensusSubscriptionCommitManager.getInstance()
- .removeAllStatesForTopic(consumerGroupId, topicName);
+ final boolean isTableModel =
SubscriptionAgent.consumer().isTableModel(consumerGroupId);
+ final TopicConfig topicConfig =
+ SubscriptionAgent.topic()
+ .getTopicConfigs(Collections.singleton(topicName),
isTableModel)
+ .get(topicName);
+ final boolean retainProgressAfterUnsubscribe =
+ retainConfiguredProgress
+ && Objects.nonNull(topicConfig)
+ && topicConfig.isProgressRetainedAfterUnsubscribe();
+ SubscriptionAgent.broker()
+ .unbindConsensusPrefetchingQueue(
+ consumerGroupId, topicName, retainProgressAfterUnsubscribe);
+
+ if (!retainProgressAfterUnsubscribe) {
+ ConsensusSubscriptionCommitManager.getInstance()
+ .removeAllStatesForTopic(consumerGroupId, topicName);
+ }
LOGGER.info(
DataNodePipeMessages
@@ -742,6 +849,179 @@ public class ConsensusSubscriptionSetupHandler {
}
}
+ public static void restoreDetachedRetentions() {
+ final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
+ if (!(dataRegionConsensus instanceof IoTConsensus)) {
+ return;
+ }
+
+ ensureNewRegionListenerRegistered();
+ final IoTConsensus ioTConsensus = (IoTConsensus) dataRegionConsensus;
+ final ConsensusSubscriptionCommitManager commitManager =
+ ConsensusSubscriptionCommitManager.getInstance();
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
+ commitManager.getPersistedStateKeys()) {
+ try {
+ final String consumerGroupId = stateKey.getConsumerGroupId();
+ final String topicName = stateKey.getTopicName();
+ final ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromString(stateKey.getRegionId());
+
+ final IoTConsensusServerImpl serverImpl =
ioTConsensus.getImpl(groupId);
+ if
(!SubscriptionAgent.consumer().containsConsumerGroup(consumerGroupId)) {
+ if (Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.unregisterDetached(
+ consumerGroupId, topicName, groupId, serverImpl);
+ }
+ commitManager.removeState(consumerGroupId, topicName, groupId);
+ continue;
+ }
+
+ final boolean isTableModel =
SubscriptionAgent.consumer().isTableModel(consumerGroupId);
+ final TopicConfig topicConfig =
+ SubscriptionAgent.topic()
+ .getTopicConfigs(Collections.singleton(topicName),
isTableModel)
+ .get(topicName);
+ if (Objects.isNull(topicConfig)
+ || !topicConfig.isIncrementalMode()
+ || !topicConfig.isProgressRetainedAfterUnsubscribe()) {
+ if (Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.unregisterDetached(
+ consumerGroupId, topicName, groupId, serverImpl);
+ }
+ commitManager.removeState(consumerGroupId, topicName, groupId);
+ continue;
+ }
+ if (SubscriptionAgent.consumer()
+ .isTopicSubscribedByConsumerGroup(consumerGroupId, topicName)) {
+ continue;
+ }
+
+ if (Objects.isNull(serverImpl)) {
+ commitManager.removeState(consumerGroupId, topicName, groupId);
+ continue;
+ }
+ ConsensusSubscriptionWalRetention.registerDetached(
+ consumerGroupId,
+ topicName,
+ groupId,
+ serverImpl,
+ buildSubscriptionWalRetentionPolicy(topicName, topicConfig,
serverImpl),
+ commitManager);
+ } catch (final Exception e) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_FAILED_TO_RESTORE_DETACHED_CONSENSUS_SUBSCRIPTION_WAL_RETENTION_FOR_CONSUMER_GROUP_ARG_TOPIC_ARG_REGION_ARG_296129FE,
+ stateKey.getConsumerGroupId(),
+ stateKey.getTopicName(),
+ stateKey.getRegionId(),
+ e);
+ }
+ }
+ }
+
+ public static void cleanupRetainedProgressForDroppedTopic(
+ final String topicName, final Boolean isTableModel) {
+ final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
+ final IoTConsensus ioTConsensus =
+ dataRegionConsensus instanceof IoTConsensus ? (IoTConsensus)
dataRegionConsensus : null;
+ final ConsensusSubscriptionCommitManager commitManager =
+ ConsensusSubscriptionCommitManager.getInstance();
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
+ commitManager.getPersistedStateKeys()) {
+ if (!topicName.equals(stateKey.getTopicName())) {
+ continue;
+ }
+ final String consumerGroupId = stateKey.getConsumerGroupId();
+ if (Objects.nonNull(isTableModel)
+ &&
SubscriptionAgent.consumer().containsConsumerGroup(consumerGroupId)
+ && SubscriptionAgent.consumer().isTableModel(consumerGroupId) !=
isTableModel) {
+ continue;
+ }
+ final ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromString(stateKey.getRegionId());
+ if (Objects.nonNull(ioTConsensus)) {
+ final IoTConsensusServerImpl serverImpl =
ioTConsensus.getImpl(groupId);
+ if (Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.unregisterDetached(
+ consumerGroupId, topicName, groupId, serverImpl);
+ }
+ }
+ commitManager.removeState(consumerGroupId, topicName, groupId);
+ }
+ }
+
+ public static void cleanupRetainedProgressForConsumerGroup(final String
consumerGroupId) {
+ final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
+ final IoTConsensus ioTConsensus =
+ dataRegionConsensus instanceof IoTConsensus ? (IoTConsensus)
dataRegionConsensus : null;
+ final ConsensusSubscriptionCommitManager commitManager =
+ ConsensusSubscriptionCommitManager.getInstance();
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
+ commitManager.getPersistedStateKeys()) {
+ if (!consumerGroupId.equals(stateKey.getConsumerGroupId())) {
+ continue;
+ }
+ final ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromString(stateKey.getRegionId());
+ if (Objects.nonNull(ioTConsensus)) {
+ final IoTConsensusServerImpl serverImpl =
ioTConsensus.getImpl(groupId);
+ if (Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.unregisterDetached(
+ consumerGroupId, stateKey.getTopicName(), groupId, serverImpl);
+ }
+ }
+ }
+ commitManager.removeAllStatesForConsumerGroup(consumerGroupId);
+ }
+
+ public static void refreshDetachedRetentionsForTopic(
+ final String topicName, final boolean isTableModel, final TopicConfig
topicConfig) {
+ final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
+ final IoTConsensus ioTConsensus =
+ dataRegionConsensus instanceof IoTConsensus ? (IoTConsensus)
dataRegionConsensus : null;
+ final ConsensusSubscriptionCommitManager commitManager =
+ ConsensusSubscriptionCommitManager.getInstance();
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
+ commitManager.getPersistedStateKeys()) {
+ if (!topicName.equals(stateKey.getTopicName())) {
+ continue;
+ }
+ final String consumerGroupId = stateKey.getConsumerGroupId();
+ if
(!SubscriptionAgent.consumer().containsConsumerGroup(consumerGroupId)) {
+ continue;
+ }
+ if (SubscriptionAgent.consumer().isTableModel(consumerGroupId) !=
isTableModel) {
+ continue;
+ }
+ if (SubscriptionAgent.consumer()
+ .isTopicSubscribedByConsumerGroup(consumerGroupId, topicName)) {
+ continue;
+ }
+ final ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromString(stateKey.getRegionId());
+ final IoTConsensusServerImpl serverImpl =
+ Objects.nonNull(ioTConsensus) ? ioTConsensus.getImpl(groupId) : null;
+ if (topicConfig.isIncrementalMode()
+ && topicConfig.isProgressRetainedAfterUnsubscribe()
+ && Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.registerDetached(
+ consumerGroupId,
+ topicName,
+ groupId,
+ serverImpl,
+ buildSubscriptionWalRetentionPolicy(topicName, topicConfig,
serverImpl),
+ commitManager);
+ } else {
+ if (Objects.nonNull(serverImpl)) {
+ ConsensusSubscriptionWalRetention.unregisterDetached(
+ consumerGroupId, topicName, groupId, serverImpl);
+ }
+ commitManager.removeState(consumerGroupId, topicName, groupId);
+ }
+ }
+ }
+
public static void handleNewSubscriptions(
final String consumerGroupId, final Set<String> newTopicNames) {
if (newTopicNames == null || newTopicNames.isEmpty()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetention.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetention.java
new file mode 100644
index 00000000000..8e3042d5e95
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetention.java
@@ -0,0 +1,163 @@
+/*
+ * 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 org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+final class ConsensusSubscriptionWalRetention {
+
+ private ConsensusSubscriptionWalRetention() {}
+
+ static String generateRetentionId(
+ final String consumerGroupId, final String topicName, final
ConsensusGroupId regionId) {
+ return consumerGroupId + "\u0000" + topicName + "\u0000" + regionId;
+ }
+
+ static boolean isProgressMonotonic(
+ final RegionProgress previousProgress, final RegionProgress
currentProgress) {
+ if (Objects.isNull(previousProgress)) {
+ return true;
+ }
+ if (Objects.isNull(currentProgress)) {
+ return false;
+ }
+ for (final Map.Entry<WriterId, WriterProgress> entry :
+ previousProgress.getWriterPositions().entrySet()) {
+ final WriterProgress previous = entry.getValue();
+ final WriterProgress current =
currentProgress.getWriterPositions().get(entry.getKey());
+ if (Objects.isNull(current)
+ || current.getPhysicalTime() < previous.getPhysicalTime()
+ || (current.getPhysicalTime() == previous.getPhysicalTime()
+ && current.getLocalSeq() < previous.getLocalSeq())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static final class RetainedWalVersionCalculator {
+
+ private final ConsensusGroupId consensusGroupId;
+ private ConsensusReqReader lastConsensusReqReader;
+ private RegionProgress lastCommittedRegionProgress;
+ private long lastCoveredWalVersionId = -1L;
+
+ private RetainedWalVersionCalculator(final ConsensusGroupId
consensusGroupId) {
+ this.consensusGroupId = consensusGroupId;
+ }
+
+ private synchronized long compute(
+ final ConsensusReqReader consensusReqReader, final RegionProgress
committedRegionProgress) {
+ if (lastConsensusReqReader != consensusReqReader
+ || !isProgressMonotonic(lastCommittedRegionProgress,
committedRegionProgress)) {
+ lastCoveredWalVersionId = -1L;
+ }
+ lastConsensusReqReader = consensusReqReader;
+ lastCommittedRegionProgress = committedRegionProgress;
+ if (!(consensusReqReader instanceof WALNode)) {
+ return 0L;
+ }
+
+ final WALNode walNode = (WALNode) consensusReqReader;
+ final long currentWalVersion = walNode.getCurrentWALFileVersion();
+ if (currentWalVersion <= lastCoveredWalVersionId) {
+ lastCoveredWalVersionId = -1L;
+ }
+ final File[] walFiles = walNode.getSortedWalFilesSnapshot();
+ if (Objects.isNull(walFiles) || walFiles.length == 0) {
+ return Math.max(0L, currentWalVersion);
+ }
+
+ for (final File walFile : walFiles) {
+ final long versionId = WALFileUtils.parseVersionId(walFile.getName());
+ if (versionId >= currentWalVersion) {
+ return Math.max(0L, currentWalVersion);
+ }
+ if (versionId <= lastCoveredWalVersionId
+ || ProgressWALIterator.isHeaderOnlyWalFile(walFile)) {
+ continue;
+ }
+
+ try (final ProgressWALReader reader = new ProgressWALReader(walFile)) {
+ if (!ConsensusPrefetchingQueue.WalFileCommitRequirement.fromMetadata(
+ consensusGroupId.toString(), reader.getMetaData())
+ .isCoveredBy(committedRegionProgress)) {
+ return versionId;
+ }
+ lastCoveredWalVersionId = versionId;
+ } catch (final IOException e) {
+ return versionId;
+ }
+ }
+ return Math.max(0L, currentWalVersion);
+ }
+ }
+
+ static LongSupplier createCommittedRetainedMinVersionIdSupplier(
+ final ConsensusGroupId consensusGroupId,
+ final Supplier<ConsensusReqReader> consensusReqReaderSupplier,
+ final Supplier<RegionProgress> committedRegionProgressSupplier) {
+ final RetainedWalVersionCalculator calculator =
+ new RetainedWalVersionCalculator(consensusGroupId);
+ return () ->
+ calculator.compute(consensusReqReaderSupplier.get(),
committedRegionProgressSupplier.get());
+ }
+
+ static void registerDetached(
+ final String consumerGroupId,
+ final String topicName,
+ final ConsensusGroupId regionId,
+ final IoTConsensusServerImpl serverImpl,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final ConsensusSubscriptionCommitManager commitManager) {
+ serverImpl.registerDetachedSubscriptionRetention(
+ generateRetentionId(consumerGroupId, topicName, regionId),
+ retentionPolicy,
+ createCommittedRetainedMinVersionIdSupplier(
+ regionId,
+ serverImpl::getConsensusReqReader,
+ () -> commitManager.getCommittedRegionProgress(consumerGroupId,
topicName, regionId)));
+ }
+
+ static void unregisterDetached(
+ final String consumerGroupId,
+ final String topicName,
+ final ConsensusGroupId regionId,
+ final IoTConsensusServerImpl serverImpl) {
+ serverImpl.deregisterDetachedSubscriptionRetention(
+ generateRetentionId(consumerGroupId, topicName, regionId));
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
index 7f9c4139040..c8bd2734e91 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
@@ -269,6 +269,16 @@ public class ConsensusPrefetchingQueueTest {
@Test
public void testCloseDetachesBlockedConsensusQueueDeregistration() throws
Exception {
+ assertCloseDetachesBlockedConsensusRegistration(false);
+ }
+
+ @Test
+ public void testRetainingCloseDetachesBlockedConsensusQueueReplacement()
throws Exception {
+ assertCloseDetachesBlockedConsensusRegistration(true);
+ }
+
+ private void assertCloseDetachesBlockedConsensusRegistration(
+ final boolean retainProgressAfterUnsubscribe) throws Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
final File systemDir =
temporaryFolder.newFolder("async-prefetch-deregister");
final CountDownLatch deregistrationStarted = new CountDownLatch(1);
@@ -279,14 +289,19 @@ public class ConsensusPrefetchingQueueTest {
final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
when(serverImpl.getConsensusReqReader()).thenReturn(reader);
when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
- doAnswer(
- invocation -> {
- deregistrationStarted.countDown();
- releaseDeregistration.await(TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
- return null;
- })
- .when(serverImpl)
- .deregisterSubscriptionQueue(any());
+ final IoTConsensusServerImpl blockedServer =
+ doAnswer(
+ invocation -> {
+ deregistrationStarted.countDown();
+ releaseDeregistration.await(TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ return null;
+ })
+ .when(serverImpl);
+ if (retainProgressAfterUnsubscribe) {
+ blockedServer.replaceSubscriptionQueueWithDetachedRetention(any(),
any(), any(), any());
+ } else {
+ blockedServer.deregisterSubscriptionQueue(any());
+ }
final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
when(converter.getDatabaseName()).thenReturn("db");
@@ -318,7 +333,7 @@ public class ConsensusPrefetchingQueueTest {
setPrefetchBinding(queue, prefetchExecutor, subtask);
final long closeStartNanos = System.nanoTime();
- queue.closeAsync(false);
+ queue.closeAsync(false, retainProgressAfterUnsubscribe);
final long closeElapsedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - closeStartNanos);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java
index 24778e2c175..9734e19593f 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java
@@ -564,6 +564,41 @@ public class ConsensusSubscriptionCommitStateTest {
}
}
+ @Test
+ public void testPersistedStateKeysAndRegionCleanup() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("persistedStateKeys");
+ try {
+ final ConsensusSubscriptionCommitManager manager =
newCommitManager(systemDir);
+ final DataRegionId firstRegion = new DataRegionId(31);
+ final DataRegionId secondRegion = new DataRegionId(32);
+ manager.receiveProgressBroadcast(
+ "cg",
+ "topic",
+ firstRegion.toString(),
+ new WriterId(firstRegion.toString(), 7),
+ new WriterProgress(100L, 1L));
+ manager.receiveProgressBroadcast(
+ "cg",
+ "topic",
+ secondRegion.toString(),
+ new WriterId(secondRegion.toString(), 7),
+ new WriterProgress(200L, 2L));
+
+ assertEquals(2, manager.getPersistedStateKeys().size());
+ manager.removeAllStatesForRegion(firstRegion);
+
+ assertFalse(manager.hasPersistedState("cg", "topic", firstRegion));
+ assertTrue(manager.hasPersistedState("cg", "topic", secondRegion));
+ assertEquals(1, manager.getPersistedStateKeys().size());
+
+ manager.removeAllStatesForConsumerGroup("cg");
+ assertTrue(manager.getPersistedStateKeys().isEmpty());
+ } finally {
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testProgressFileNameEncodesForbiddenTopicComponents() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
index 30cbd487668..9401fc3e4af 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandlerTest.java
@@ -21,10 +21,14 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
+import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
+import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
import org.junit.Rule;
import org.junit.Test;
@@ -38,12 +42,18 @@ import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.LongSupplier;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
public class ConsensusSubscriptionSetupHandlerTest {
@@ -143,6 +153,43 @@ public class ConsensusSubscriptionSetupHandlerTest {
}
}
+ @Test
+ public void testStartupDetachedRetentionProtectsPersistedProgress() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("startupDetachedRetention");
+ try {
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath());
+ final ConsensusSubscriptionCommitManager commitManager =
+ new ConsensusSubscriptionCommitManager(
+ (consumerGroupId, topicName, regionId) ->
+
ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent());
+ final DataRegionId regionId = new DataRegionId(41);
+ commitManager.receiveProgressBroadcast(
+ "consumerGroup",
+ "topic",
+ regionId.toString(),
+ new WriterId(regionId.toString(), 7),
+ new WriterProgress(100L, 1L));
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+
+ ConsensusSubscriptionSetupHandler.registerStartupDetachedRetentions(
+ regionId, serverImpl, commitManager);
+
+ verify(serverImpl)
+ .registerDetachedSubscriptionRetention(
+ eq(
+ ConsensusSubscriptionWalRetention.generateRetentionId(
+ "consumerGroup", "topic", regionId)),
+ argThat(
+ policy ->
+ policy.getRetentionBytes() ==
SubscriptionWalRetentionPolicy.UNBOUNDED
+ && policy.getRetentionMs() ==
SubscriptionWalRetentionPolicy.UNBOUNDED),
+ any(LongSupplier.class));
+ } finally {
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testAuditDatabaseNeverMatchesTopic() {
final Map<String, String> tableTopicAttributes = new HashMap<>();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetentionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetentionTest.java
new file mode 100644
index 00000000000..18312370782
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionWalRetentionTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.DataRegionId;
+import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
+import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
+import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
+import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
+import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
+
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.function.LongSupplier;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class ConsensusSubscriptionWalRetentionTest {
+
+ private static final WriterId WRITER_ID = new WriterId("region", 7);
+
+ @Test
+ public void testProgressRegressionInvalidatesCoveredWalVersion() {
+ final RegionProgress previous = progress(100L, 2L);
+
+ assertTrue(ConsensusSubscriptionWalRetention.isProgressMonotonic(null,
previous));
+ assertTrue(ConsensusSubscriptionWalRetention.isProgressMonotonic(previous,
previous));
+ assertTrue(ConsensusSubscriptionWalRetention.isProgressMonotonic(previous,
progress(100L, 3L)));
+ assertTrue(ConsensusSubscriptionWalRetention.isProgressMonotonic(previous,
progress(101L, 0L)));
+ assertFalse(
+ ConsensusSubscriptionWalRetention.isProgressMonotonic(previous,
progress(100L, 1L)));
+
assertFalse(ConsensusSubscriptionWalRetention.isProgressMonotonic(previous,
progress(99L, 3L)));
+ assertFalse(
+ ConsensusSubscriptionWalRetention.isProgressMonotonic(
+ previous, new RegionProgress(Collections.emptyMap())));
+
assertFalse(ConsensusSubscriptionWalRetention.isProgressMonotonic(previous,
null));
+ }
+
+ @Test
+ public void testDetachedRetentionUsesSortedWalSnapshot() {
+ final DataRegionId regionId = new DataRegionId(1);
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ final WALNode walNode = mock(WALNode.class);
+ final ConsensusSubscriptionCommitManager commitManager =
+ mock(ConsensusSubscriptionCommitManager.class);
+ final SubscriptionWalRetentionPolicy retentionPolicy =
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+ when(walNode.getCurrentWALFileVersion()).thenReturn(7L);
+ when(walNode.getSortedWalFilesSnapshot()).thenReturn(new File[0]);
+ when(commitManager.getCommittedRegionProgress("group", "topic", regionId))
+ .thenReturn(new RegionProgress(Collections.emptyMap()));
+
+ ConsensusSubscriptionWalRetention.registerDetached(
+ "group", "topic", regionId, serverImpl, retentionPolicy,
commitManager);
+
+ final ArgumentCaptor<LongSupplier> supplierCaptor =
ArgumentCaptor.forClass(LongSupplier.class);
+ verify(serverImpl)
+ .registerDetachedSubscriptionRetention(
+ eq(ConsensusSubscriptionWalRetention.generateRetentionId("group",
"topic", regionId)),
+ same(retentionPolicy),
+ supplierCaptor.capture());
+ assertEquals(7L, supplierCaptor.getValue().getAsLong());
+ verify(walNode).getSortedWalFilesSnapshot();
+ }
+
+ private static RegionProgress progress(final long physicalTime, final long
localSeq) {
+ return new RegionProgress(
+ Collections.singletonMap(WRITER_ID, new WriterProgress(physicalTime,
localSeq)));
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java
index 2b291311fab..a0b16d9b7c1 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java
@@ -116,6 +116,21 @@ public class CommitProgressKeeper {
&& isValidRegionAndDataNodeProgressKey(key,
legacyTopicKeyPrefix)));
}
+ public synchronized void removeConsumerGroupProgress(final String
consumerGroupId) {
+ final String versionedConsumerGroupKeyPrefix =
+ VERSIONED_KEY_PREFIX + encodeKeyComponent(consumerGroupId) +
KEY_COMPONENT_SEPARATOR;
+ final String consumerGroupString = String.valueOf(consumerGroupId);
+ final String legacyConsumerGroupKeyPrefix =
+ consumerGroupString.contains(KEY_SEPARATOR) ? null :
consumerGroupString + KEY_SEPARATOR;
+ regionProgressMap
+ .keySet()
+ .removeIf(
+ key ->
+ key.startsWith(versionedConsumerGroupKeyPrefix)
+ || (legacyConsumerGroupKeyPrefix != null
+ && key.startsWith(legacyConsumerGroupKeyPrefix)));
+ }
+
private static boolean isValidRegionAndDataNodeProgressKey(
final String key, final String topicKeyPrefix) {
if (!key.startsWith(topicKeyPrefix)) {
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java
index ad040b4c8eb..7f18e4875d1 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeperTest.java
@@ -50,6 +50,7 @@ public class CommitProgressKeeperTest {
@Test
public void testStateAccessUsesOneInstanceMonitor() throws Exception {
assertSynchronized("removeTopicProgress", String.class, String.class);
+ assertSynchronized("removeConsumerGroupProgress", String.class);
assertSynchronized("updateRegionProgress", String.class, ByteBuffer.class);
assertSynchronized("getRegionProgress", String.class);
assertSynchronized("getAllRegionProgress");
@@ -116,6 +117,24 @@ public class CommitProgressKeeperTest {
assertNotNull(keeper.getRegionProgress(masqueradingLegacyKey));
}
+ @Test
+ public void testRemoveConsumerGroupProgressRemovesAllTopics() {
+ final CommitProgressKeeper keeper = new CommitProgressKeeper();
+ final ByteBuffer progress = ByteBuffer.wrap(new byte[] {1});
+ final String firstTopicKey = CommitProgressKeeper.generateKey("cg",
"topic1", "1_1", 1);
+ final String secondTopicKey = CommitProgressKeeper.generateLegacyKey("cg",
"topic2", "1_2", 2);
+ final String otherGroupKey = CommitProgressKeeper.generateKey("other",
"topic1", "1_1", 1);
+ keeper.updateRegionProgress(firstTopicKey, progress);
+ keeper.updateRegionProgress(secondTopicKey, progress);
+ keeper.updateRegionProgress(otherGroupKey, progress);
+
+ keeper.removeConsumerGroupProgress("cg");
+
+ assertNull(keeper.getRegionProgress(firstTopicKey));
+ assertNull(keeper.getRegionProgress(secondTopicKey));
+ assertNotNull(keeper.getRegionProgress(otherGroupKey));
+ }
+
@Test
public void testUpdateAndReplaceAllUseDefensiveCopies() throws Exception {
final CommitProgressKeeper keeper = new CommitProgressKeeper();
