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 9b807abb93f [Bugfix] Preserve retained subscription progress after the
last consumer closes (#18894)
9b807abb93f is described below
commit 9b807abb93f06ae1e1339dd41d990bcad93e897e
Author: Caideyipi <[email protected]>
AuthorDate: Fri Oct 16 10:30:42 2026 +0800
[Bugfix] Preserve retained subscription progress after the last consumer
closes (#18894)
* [Bugfix] Preserve retained subscription progress after the last consumer
closes
* Fix detached retention queue capture
---
.../IoTDBConsensusSubscriptionBasicTableIT.java | 60 +++++++++++++
.../persistence/subscription/SubscriptionInfo.java | 45 +++++++---
.../subscription/SubscriptionInfoTest.java | 99 ++++++++++++++++++++++
.../consensus/iot/IoTConsensusServerImpl.java | 15 +++-
.../subscription/SubscriptionQueueRegistry.java | 46 ++++++++--
.../SubscriptionQueueRegistryTest.java | 46 ++++++++++
.../agent/SubscriptionConsumerAgent.java | 44 +++++++---
.../consensus/ConsensusPrefetchingQueue.java | 22 +++--
.../ConsensusSubscriptionSetupHandler.java | 36 +++++++-
.../consensus/ConsensusPrefetchingQueueTest.java | 75 +++++++++++++++-
.../meta/consumer/CommitProgressKeeper.java | 68 ++++++++++++---
.../meta/consumer/ConsumerGroupMeta.java | 37 +++++---
.../consumer/ConsumerGroupDeSerTest.java | 39 +++++++++
.../meta/consumer/CommitProgressKeeperTest.java | 26 ++++++
14 files changed, 590 insertions(+), 68 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionBasicTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionBasicTableIT.java
index 1a64e70d4fd..aeb1850882e 100644
---
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionBasicTableIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionBasicTableIT.java
@@ -19,8 +19,12 @@
package org.apache.iotdb.subscription.it.consensus.local.tablemodel;
+import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
+import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
import
org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumer;
import
org.apache.iotdb.subscription.it.consensus.local.AbstractSubscriptionConsensusLocalIT;
@@ -31,12 +35,68 @@ import org.junit.runner.RunWith;
import java.time.Duration;
import java.util.LinkedHashSet;
+import java.util.Properties;
import java.util.Set;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class})
public class IoTDBConsensusSubscriptionBasicTableIT extends
AbstractSubscriptionConsensusLocalIT {
+ @Test
+ public void testLastConsumerCloseRetainsProgress() throws Exception {
+ for (final boolean unsubscribeFirst : new boolean[] {false, true}) {
+ final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
+
ConsensusSubscriptionTableITSupport.newIdentifiers("retained_last_close");
+ SubscriptionTablePullConsumer consumer = null;
+ try {
+ ConsensusSubscriptionTableITSupport.bootstrapDatabaseAndTable(
+ ids.getDatabase(), "t1",
ConsensusSubscriptionTableITSupport.DEFAULT_TABLE_SCHEMA);
+ ConsensusSubscriptionTableITSupport.createConsensusTopic(
+ ids.getTopic(), ids.getDatabase(), "t1");
+ try (final ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder()
+ .host(EnvFactory.getEnv().getIP())
+ .port(Integer.parseInt(EnvFactory.getEnv().getPort()))
+ .build()) {
+ session.open();
+ final Properties config = new Properties();
+
config.setProperty(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true");
+ session.alterTopic(ids.getTopic(), config);
+ }
+ consumer =
+ ConsensusSubscriptionTableITSupport.createConsumer(
+ ids.getConsumerId(), ids.getConsumerGroupId());
+ consumer.subscribe(ids.getTopic());
+ final Set<String> firstRows =
+ ConsensusSubscriptionTableITSupport.insertRows(ids.getDatabase(),
"t1", 100L, 10, true);
+ ConsensusSubscriptionTableITSupport.assertExactRowKeys(
+ firstRows,
+ ConsensusSubscriptionTableITSupport.pollAndCommitUntilAtLeast(
+ consumer, firstRows.size(), 40));
+ if (unsubscribeFirst) {
+ consumer.unsubscribe(ids.getTopic());
+ }
+ consumer.close();
+ consumer = null;
+
+ final Set<String> offlineRows =
+ ConsensusSubscriptionTableITSupport.insertRows(ids.getDatabase(),
"t1", 200L, 10, true);
+ consumer =
+ ConsensusSubscriptionTableITSupport.createConsumer(
+ ids.consumer("replacement"), ids.getConsumerGroupId());
+ consumer.subscribe(ids.getTopic());
+ ConsensusSubscriptionTableITSupport.assertExactRowKeys(
+ offlineRows,
+ ConsensusSubscriptionTableITSupport.pollAndCommitUntilAtLeast(
+ consumer, offlineRows.size(), 40));
+ ConsensusSubscriptionTableITSupport.assertNoMoreMessages(
+ consumer, 3, Duration.ofMillis(500));
+ } finally {
+ ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(),
ids.getDatabase());
+ }
+ }
+ }
+
@Test
public void testRealtimeOnlyAfterSubscribe() throws Exception {
final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
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 58fe0a10690..c0b6d10e848 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
@@ -929,6 +929,7 @@ public class SubscriptionInfo implements SnapshotProcessor {
plan.getTopicMeta().getTopicName())) {
commitProgressKeeper.removeTopicProgress(
consumerGroupMeta.getConsumerGroupId(),
plan.getTopicMeta().getTopicName());
+ removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta);
}
}
}
@@ -971,6 +972,7 @@ public class SubscriptionInfo implements SnapshotProcessor {
if (!plan.isTableModelSet() ||
consumerGroupMeta.visibleUnder(plan.isTableModel())) {
commitProgressKeeper.removeTopicProgress(
consumerGroupMeta.getConsumerGroupId(), plan.getTopicName());
+ removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta);
}
}
if (plan.isTableModelSet()) {
@@ -1146,24 +1148,32 @@ public class SubscriptionInfo implements
SnapshotProcessor {
? consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId)
: null;
if (Objects.nonNull(currentConsumerGroupMeta)) {
+ 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);
+ }
+ });
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);
- }
- });
+ final Set<String> retainedTopicNames = new HashSet<>();
+ for (final TopicMeta topicMeta :
topicMetaKeeper.getAllTopicMeta()) {
+ if (topicMeta.visibleUnderTableModel() ==
currentConsumerGroupMeta.visibleUnder(true)
+ &&
topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) {
+ retainedTopicNames.add(topicMeta.getTopicName());
+ }
+ }
+ commitProgressKeeper.removeConsumerGroupProgressExceptTopics(
+ consumerGroupId, retainedTopicNames);
}
}
consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId);
- if (!consumerGroupMeta.isEmpty()) {
+ if (!consumerGroupMeta.isEmpty()
+ || commitProgressKeeper.hasConsumerGroupProgress(consumerGroupId))
{
consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId,
consumerGroupMeta);
}
}
@@ -1173,6 +1183,13 @@ public class SubscriptionInfo implements
SnapshotProcessor {
}
}
+ private void removeEmptyConsumerGroupWithoutProgress(final ConsumerGroupMeta
consumerGroupMeta) {
+ if (consumerGroupMeta.isEmpty()
+ &&
!commitProgressKeeper.hasConsumerGroupProgress(consumerGroupMeta.getConsumerGroupId()))
{
+
consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupMeta.getConsumerGroupId());
+ }
+ }
+
public TSStatus
handleConsumerGroupMetaChanges(ConsumerGroupHandleMetaChangePlan plan) {
acquireWriteLock();
try {
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 087cb5868f5..35cd29dc617 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
@@ -34,10 +34,13 @@ import
org.apache.iotdb.confignode.rpc.thrift.TShowTopicInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp;
import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.junit.Assert;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import java.nio.ByteBuffer;
import java.util.Collections;
@@ -49,6 +52,102 @@ import java.util.UUID;
public class SubscriptionInfoTest {
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testLastConsumerClosePreservesConfiguredProgressAcrossSnapshot()
throws Exception {
+ for (final boolean tableModel : new boolean[] {false, true}) {
+ for (final boolean unsubscribeFirst : new boolean[] {false, true}) {
+ assertLastConsumerClosePreservesProgress(tableModel, unsubscribeFirst);
+ }
+ }
+ }
+
+ private void assertLastConsumerClosePreservesProgress(
+ final boolean tableModel, final boolean unsubscribeFirst) throws
Exception {
+ final SubscriptionInfo info = new SubscriptionInfo();
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(TopicConstant.MODE_KEY,
TopicConstant.MODE_INCREMENTAL_VALUE);
+ attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"true");
+ attributes.put(SystemConstant.SQL_DIALECT_KEY, tableModel ? "table" :
"tree");
+ info.createTopic(new CreateTopicPlan(new TopicMeta("kept", 1L,
attributes)));
+ final Map<String, String> discardedAttributes = new HashMap<>(attributes);
+
discardedAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"false");
+ info.createTopic(new CreateTopicPlan(new TopicMeta("discarded", 1L,
discardedAttributes)));
+ final ConsumerGroupMeta group =
+ new ConsumerGroupMeta(
+ "group",
+ 1L,
+ new ConsumerMeta(
+ "consumer",
+ 1L,
+ Collections.singletonMap(
+ ConsumerConstant.SQL_DIALECT_KEY, tableModel ? "table" :
"tree")));
+ group.addSubscription(
+ "consumer", new java.util.HashSet<>(java.util.Arrays.asList("kept",
"discarded")));
+ info.alterConsumerGroup(new AlterConsumerGroupPlan(group));
+ final String keptKey = CommitProgressKeeper.generateKey("group", "kept",
"DataRegion[1]", 1);
+ final String legacyKey =
+ CommitProgressKeeper.generateLegacyKey("group", "kept",
"DataRegion[1]", 2);
+ final String discardedKey =
+ CommitProgressKeeper.generateKey("group", "discarded",
"DataRegion[1]", 1);
+ for (final String key : new String[] {keptKey, legacyKey, discardedKey}) {
+ info.getCommitProgressKeeper().updateRegionProgress(key,
ByteBuffer.wrap(new byte[] {1}));
+ }
+ final ConsumerGroupMeta updated = group.deepCopy();
+ if (unsubscribeFirst) {
+ updated.removeSubscription(
+ "consumer", new java.util.HashSet<>(java.util.Arrays.asList("kept",
"discarded")));
+ info.alterConsumerGroup(new AlterConsumerGroupPlan(updated.deepCopy()));
+ }
+ updated.removeConsumer("consumer");
+ info.alterConsumerGroup(new AlterConsumerGroupPlan(updated));
+ Assert.assertNotNull(
+ "tableModel=" + tableModel + ", unsubscribeFirst=" + unsubscribeFirst,
+ info.getCommitProgressKeeper().getRegionProgress(keptKey));
+
Assert.assertNotNull(info.getCommitProgressKeeper().getRegionProgress(legacyKey));
+
Assert.assertNull(info.getCommitProgressKeeper().getRegionProgress(discardedKey));
+ Assert.assertTrue(info.getConsumerGroupMeta("group").isEmpty());
+
Assert.assertTrue(info.getConsumerGroupMeta("group").visibleUnder(tableModel));
+
+ final java.io.File snapshotDir = temporaryFolder.newFolder();
+ Assert.assertTrue(info.processTakeSnapshot(snapshotDir));
+ final SubscriptionInfo restored = new SubscriptionInfo();
+ restored.processLoadSnapshot(snapshotDir);
+
Assert.assertNotNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
+ final ConsumerGroupMeta restoredGroup =
restored.deepCopyConsumerGroupMeta("group");
+ Assert.assertTrue(restoredGroup.isEmpty());
+ Assert.assertTrue(restoredGroup.visibleUnder(tableModel));
+ final ConsumerMeta replacement =
+ new ConsumerMeta(
+ "replacement",
+ 2L,
+ Collections.singletonMap(
+ ConsumerConstant.SQL_DIALECT_KEY, tableModel ? "table" :
"tree"));
+ restoredGroup.checkAuthorityBeforeJoinConsumerGroup(replacement);
+ restoredGroup.addConsumer(replacement);
+ restored.alterConsumerGroup(new AlterConsumerGroupPlan(restoredGroup));
+
Assert.assertNotNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
+ restoredGroup.removeConsumer("replacement");
+ restored.alterConsumerGroup(new AlterConsumerGroupPlan(restoredGroup));
+
+ // Same-name topics in the other model must not remove this retained
group's progress.
+ final Map<String, String> otherAttributes = new HashMap<>(attributes);
+ otherAttributes.put(SystemConstant.SQL_DIALECT_KEY, tableModel ? "tree" :
"table");
+ restored.createTopic(new CreateTopicPlan(new TopicMeta("kept", 2L,
otherAttributes)));
+ restored.dropTopic(new DropTopicPlan("kept", !tableModel));
+
Assert.assertNotNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
+ if (unsubscribeFirst) {
+ attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY,
"false");
+ restored.alterTopic(new AlterTopicPlan(new TopicMeta("kept", 2L,
attributes)));
+ } else {
+ restored.dropTopic(new DropTopicPlan("kept", tableModel));
+ }
+
Assert.assertNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
+
Assert.assertNull(restored.getCommitProgressKeeper().getRegionProgress(legacyKey));
+ Assert.assertFalse(restored.isConsumerGroupExisted("group"));
+ }
+
@Test
public void testSameNameTreeAndTableTopicsCanBeManagedIndependently() {
final String topicName = "same-name-topic";
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 ca29931f1c9..aab4c7523cf 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
@@ -1267,7 +1267,20 @@ public class IoTConsensusServerImpl {
final BlockingQueue<IndexedConsensusRequest> queue,
final SubscriptionWalRetentionPolicy retentionPolicy,
final LongSupplier retainedMinVersionIdSupplier) {
- subscriptionQueueRegistry.register(queue, retentionPolicy,
retainedMinVersionIdSupplier);
+ registerSubscriptionQueue(queue, retentionPolicy,
retainedMinVersionIdSupplier, null);
+ }
+
+ public void registerSubscriptionQueue(
+ final BlockingQueue<IndexedConsensusRequest> queue,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier retainedMinVersionIdSupplier,
+ final String retentionId) {
+ if (retentionId == null) {
+ subscriptionQueueRegistry.register(queue, retentionPolicy,
retainedMinVersionIdSupplier);
+ } else {
+ subscriptionQueueRegistry.register(
+ queue, retentionPolicy, retainedMinVersionIdSupplier, retentionId);
+ }
// Immediately re-evaluate the safe delete index with new subscription
awareness
checkAndUpdateSafeDeletedSearchIndex();
logger.info(
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 995b876cbd1..e75324a8854 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
@@ -45,6 +45,8 @@ public class SubscriptionQueueRegistry {
new ConcurrentHashMap<>();
private final Map<String, SubscriptionQueueRegistration>
detachedRetentionRegistrations =
new ConcurrentHashMap<>();
+ private final Map<String, BlockingQueue<IndexedConsensusRequest>>
retentionOwners =
+ 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];
@@ -73,12 +75,25 @@ public class SubscriptionQueueRegistry {
final SubscriptionWalRetentionPolicy retentionPolicy,
final LongSupplier retainedMinVersionIdSupplier) {
queues.put(
- queue, new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ queue,
+ new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier, null));
+ }
+
+ public synchronized void register(
+ final BlockingQueue<IndexedConsensusRequest> queue,
+ final SubscriptionWalRetentionPolicy retentionPolicy,
+ final LongSupplier retainedMinVersionIdSupplier,
+ final String retentionId) {
+ replaceDetachedRetentionWithQueue(
+ retentionId, queue, retentionPolicy, retainedMinVersionIdSupplier);
}
// Shares the monitor with offer() so unregister() is a real stop-receiving
barrier.
public synchronized void unregister(final
BlockingQueue<IndexedConsensusRequest> queue) {
- queues.remove(queue);
+ final SubscriptionQueueRegistration registration = queues.remove(queue);
+ if (registration != null && registration.retentionId != null) {
+ retentionOwners.remove(registration.retentionId, queue);
+ }
}
public synchronized void replaceQueueWithDetachedRetention(
@@ -86,10 +101,18 @@ public class SubscriptionQueueRegistry {
final String retentionId,
final SubscriptionWalRetentionPolicy retentionPolicy,
final LongSupplier retainedMinVersionIdSupplier) {
- queues.remove(queue);
+ final SubscriptionQueueRegistration registration = queues.remove(queue);
+ // A later unsubscribe cleanup or a new queue may have superseded this
asynchronous close.
+ // The old queue must still stop receiving writes, but cannot recreate
detached retention.
+ if (registration == null
+ || (registration.retentionId != null
+ && !retentionOwners.remove(registration.retentionId, queue))) {
+ return;
+ }
detachedRetentionRegistrations.put(
retentionId,
- new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ new SubscriptionQueueRegistration(
+ retentionPolicy, retainedMinVersionIdSupplier, retentionId));
}
public synchronized void replaceDetachedRetentionWithQueue(
@@ -98,20 +121,26 @@ public class SubscriptionQueueRegistry {
final SubscriptionWalRetentionPolicy retentionPolicy,
final LongSupplier retainedMinVersionIdSupplier) {
detachedRetentionRegistrations.remove(retentionId);
+ retentionOwners.put(retentionId, queue);
queues.put(
- queue, new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ queue,
+ new SubscriptionQueueRegistration(
+ retentionPolicy, retainedMinVersionIdSupplier, retentionId));
}
public synchronized void registerDetachedRetention(
final String retentionId,
final SubscriptionWalRetentionPolicy retentionPolicy,
final LongSupplier retainedMinVersionIdSupplier) {
+ retentionOwners.remove(retentionId);
detachedRetentionRegistrations.put(
retentionId,
- new SubscriptionQueueRegistration(retentionPolicy,
retainedMinVersionIdSupplier));
+ new SubscriptionQueueRegistration(
+ retentionPolicy, retainedMinVersionIdSupplier, retentionId));
}
public synchronized void unregisterDetachedRetention(final String
retentionId) {
+ retentionOwners.remove(retentionId);
detachedRetentionRegistrations.remove(retentionId);
}
@@ -230,12 +259,15 @@ public class SubscriptionQueueRegistry {
private final SubscriptionWalRetentionPolicy retentionPolicy;
private final LongSupplier retainedMinVersionIdSupplier;
+ private final String retentionId;
private SubscriptionQueueRegistration(
final SubscriptionWalRetentionPolicy retentionPolicy,
- final LongSupplier retainedMinVersionIdSupplier) {
+ final LongSupplier retainedMinVersionIdSupplier,
+ final String retentionId) {
this.retentionPolicy = retentionPolicy;
this.retainedMinVersionIdSupplier = retainedMinVersionIdSupplier;
+ this.retentionId = retentionId;
}
}
}
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 214cc2ec8f2..86222dd1118 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
@@ -37,6 +37,52 @@ import static org.junit.Assert.assertTrue;
public class SubscriptionQueueRegistryTest {
+ @Test
+ public void testCancelledRetentionCannotBeRestoredByDelayedQueueClose() {
+ final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("test");
+ final InspectingQueue queue = new InspectingQueue();
+ final SubscriptionWalRetentionPolicy policy =
+ new SubscriptionWalRetentionPolicy("topic", -1, -1);
+ registry.register(queue, policy, () -> 3L, "retention");
+ registry.unregisterDetachedRetention("retention");
+ registry.replaceQueueWithDetachedRetention(queue, "retention", policy, ()
-> 3L);
+ assertTrue(registry.isEmpty());
+ }
+
+ @Test
+ public void testDelayedOldQueueCloseCannotReplaceNewQueueRetention() {
+ final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("test");
+ final InspectingQueue oldQueue = new InspectingQueue();
+ final InspectingQueue newQueue = new InspectingQueue();
+ final SubscriptionWalRetentionPolicy policy =
+ new SubscriptionWalRetentionPolicy("topic", -1, -1);
+ registry.register(oldQueue, policy, () -> 3L, "retention");
+ registry.replaceDetachedRetentionWithQueue("retention", newQueue, policy,
() -> 9L);
+ registry.replaceQueueWithDetachedRetention(oldQueue, "retention", policy,
() -> 3L);
+ assertEquals(1, registry.size());
+ assertEquals(Collections.singletonList(9L),
registry.getRetainedMinVersionIds());
+ registry.replaceQueueWithDetachedRetention(newQueue, "retention", policy,
() -> 9L);
+ assertFalse(registry.hasQueues());
+ assertEquals(Collections.singletonList(9L),
registry.getRetainedMinVersionIds());
+ registry.unregisterDetachedRetention("retention");
+ assertTrue(registry.isEmpty());
+ }
+
+ @Test
+ public void testUpdatedDetachedPolicyIsNotOverwrittenByOldQueueClose() {
+ final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("test");
+ final InspectingQueue queue = new InspectingQueue();
+ final SubscriptionWalRetentionPolicy oldPolicy =
+ new SubscriptionWalRetentionPolicy("topic", -1, -1);
+ final SubscriptionWalRetentionPolicy newPolicy =
+ new SubscriptionWalRetentionPolicy("topic", 100, 200);
+ registry.register(queue, oldPolicy, () -> 3L, "retention");
+ registry.registerDetachedRetention("retention", newPolicy, () -> 9L);
+ registry.replaceQueueWithDetachedRetention(queue, "retention", oldPolicy,
() -> 3L);
+ assertFalse(registry.hasQueues());
+ assertEquals(Collections.singletonList(newPolicy),
registry.getRetentionPolicies());
+ }
+
@Test
public void testOfferWithoutQueuesDoesNotSerializeRequest() {
final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("test");
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 7917fa39a5c..8af488c5923 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
@@ -76,13 +76,7 @@ public class SubscriptionConsumerAgent {
try {
executeWithConsumerGroupLock(
consumerGroupId,
- () -> {
- if (consumerGroupMetaFromCoordinator.isEmpty()) {
- handleDropConsumerGroupInternal(consumerGroupId);
- } else {
-
handleSingleConsumerGroupMetaChangesInternal(consumerGroupMetaFromCoordinator);
- }
- });
+ () ->
handleSingleConsumerGroupMetaChangesInternal(consumerGroupMetaFromCoordinator));
return null;
} catch (final Exception e) {
LOGGER.warn(
@@ -104,6 +98,22 @@ public class SubscriptionConsumerAgent {
final String consumerGroupId = metaFromCoordinator.getConsumerGroupId();
final ConsumerGroupMeta metaInAgent =
getConsumerGroupMeta(consumerGroupId);
+ if (metaFromCoordinator.isEmpty()) {
+ if (Objects.nonNull(metaInAgent)) {
+ cleanupConsumerGroupSubscriptions(consumerGroupId, metaInAgent, true);
+ }
+ final boolean retainProgress =
+
ConsensusSubscriptionSetupHandler.retainConfiguredProgressForConsumerGroup(
+ consumerGroupId, metaFromCoordinator.visibleUnder(true));
+ dropConsumerGroupBroker(consumerGroupId);
+ if (retainProgress) {
+ replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator);
+ } else {
+ removeConsumerGroupMeta(consumerGroupId);
+ }
+ return;
+ }
+
// if consumer group meta does not exist on local agent
if (Objects.isNull(metaInAgent)) {
SubscriptionAgent.broker().createPipeBrokerIfNotExist(consumerGroupId);
@@ -249,6 +259,11 @@ public class SubscriptionConsumerAgent {
cleanupConsumerGroupSubscriptions(consumerGroupId, consumerGroupMeta);
}
+ dropConsumerGroupBroker(consumerGroupId);
+ removeConsumerGroupMeta(consumerGroupId);
+ }
+
+ private void dropConsumerGroupBroker(final String consumerGroupId) {
if (SubscriptionAgent.broker().isBrokerExist(consumerGroupId)) {
if (!SubscriptionAgent.broker().dropBroker(consumerGroupId)) {
final String exceptionMessage =
@@ -262,8 +277,6 @@ public class SubscriptionConsumerAgent {
.PIPE_LOG_SUBSCRIPTION_BROKER_BOUND_TO_CONSUMER_GROUP_DOES_NOT_EXISTED_9F09E4DE,
consumerGroupId);
}
-
- removeConsumerGroupMeta(consumerGroupId);
}
private ConsumerGroupMeta getConsumerGroupMeta(final String consumerGroupId)
{
@@ -329,6 +342,13 @@ public class SubscriptionConsumerAgent {
private void cleanupConsumerGroupSubscriptions(
final String consumerGroupId, final ConsumerGroupMeta consumerGroupMeta)
{
+ cleanupConsumerGroupSubscriptions(consumerGroupId, consumerGroupMeta,
false);
+ }
+
+ private void cleanupConsumerGroupSubscriptions(
+ final String consumerGroupId,
+ final ConsumerGroupMeta consumerGroupMeta,
+ final boolean retainConfiguredProgress) {
final Set<String> consensusTopics = new LinkedHashSet<>();
for (final String topicName : consumerGroupMeta.getSubscribedTopicNames())
{
if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic(
@@ -339,8 +359,10 @@ public class SubscriptionConsumerAgent {
}
}
ConsensusSubscriptionSetupHandler.teardownConsensusSubscriptions(
- consumerGroupId, consensusTopics, false);
-
ConsensusSubscriptionSetupHandler.cleanupRetainedProgressForConsumerGroup(consumerGroupId);
+ consumerGroupId, consensusTopics, retainConfiguredProgress);
+ if (!retainConfiguredProgress) {
+
ConsensusSubscriptionSetupHandler.cleanupRetainedProgressForConsumerGroup(consumerGroupId);
+ }
}
public boolean isConsumerExisted(final String consumerGroupId, final String
consumerId) {
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 4e71f2e2fe9..9143e40796c 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
@@ -690,7 +690,11 @@ public class ConsensusPrefetchingQueue {
this::getRequiredRetainedMinVersionId);
} else {
serverImpl.registerSubscriptionQueue(
- pendingEntries, retentionPolicy,
this::getRequiredRetainedMinVersionId);
+ pendingEntries,
+ retentionPolicy,
+ this::getRequiredRetainedMinVersionId,
+ ConsensusSubscriptionWalRetention.generateRetentionId(
+ consumerGroupId, topicName, consensusGroupId));
}
LOGGER.info(
@@ -4474,17 +4478,25 @@ public class ConsensusPrefetchingQueue {
final boolean retainProgressAfterUnsubscribe) {
try {
if (retainProgressAfterUnsubscribe) {
+ // Detached retention outlives this queue. Capture only the reader and
progress manager
+ // required for WAL retention calculation instead of capturing the
queue through a field
+ // reference in either supplier.
+ final ConsensusReqReader retainedConsensusReqReader =
consensusReqReader;
+ final ConsensusSubscriptionCommitManager retainedCommitManager =
commitManager;
+ final String retainedConsumerGroupId = consumerGroupId;
+ final String retainedTopicName = topicName;
+ final ConsensusGroupId retainedConsensusGroupId = consensusGroupId;
serverImpl.replaceSubscriptionQueueWithDetachedRetention(
pendingEntries,
ConsensusSubscriptionWalRetention.generateRetentionId(
consumerGroupId, topicName, consensusGroupId),
retentionPolicy,
ConsensusSubscriptionWalRetention.createCommittedRetainedMinVersionIdSupplier(
- consensusGroupId,
- () -> consensusReqReader,
+ retainedConsensusGroupId,
+ () -> retainedConsensusReqReader,
() ->
- commitManager.getCommittedRegionProgress(
- consumerGroupId, topicName, consensusGroupId)));
+ retainedCommitManager.getCommittedRegionProgress(
+ retainedConsumerGroupId, retainedTopicName,
retainedConsensusGroupId)));
} else {
serverImpl.deregisterSubscriptionQueue(pendingEntries);
}
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 92cfade03c2..964bbcb2259 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
@@ -952,6 +952,34 @@ public class ConsensusSubscriptionSetupHandler {
}
public static void cleanupRetainedProgressForConsumerGroup(final String
consumerGroupId) {
+ cleanupConsumerGroupProgressExceptTopics(consumerGroupId,
Collections.emptySet());
+ }
+
+ public static boolean retainConfiguredProgressForConsumerGroup(
+ final String consumerGroupId, final boolean isTableModel) {
+ final Set<String> topicNames = new LinkedHashSet<>();
+ for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
+
ConsensusSubscriptionCommitManager.getInstance().getPersistedStateKeys()) {
+ if (consumerGroupId.equals(stateKey.getConsumerGroupId())) {
+ topicNames.add(stateKey.getTopicName());
+ }
+ }
+ final Set<String> retainedTopicNames = new LinkedHashSet<>();
+ SubscriptionAgent.topic()
+ .getTopicConfigs(topicNames, isTableModel)
+ .forEach(
+ (topicName, topicConfig) -> {
+ if (topicConfig.isIncrementalMode()
+ && topicConfig.isProgressRetainedAfterUnsubscribe()) {
+ retainedTopicNames.add(topicName);
+ }
+ });
+ cleanupConsumerGroupProgressExceptTopics(consumerGroupId,
retainedTopicNames);
+ return !retainedTopicNames.isEmpty();
+ }
+
+ private static void cleanupConsumerGroupProgressExceptTopics(
+ final String consumerGroupId, final Set<String> retainedTopicNames) {
final IConsensus dataRegionConsensus =
DataRegionConsensusImpl.getInstance();
final IoTConsensus ioTConsensus =
dataRegionConsensus instanceof IoTConsensus ? (IoTConsensus)
dataRegionConsensus : null;
@@ -959,7 +987,8 @@ public class ConsensusSubscriptionSetupHandler {
ConsensusSubscriptionCommitManager.getInstance();
for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey :
commitManager.getPersistedStateKeys()) {
- if (!consumerGroupId.equals(stateKey.getConsumerGroupId())) {
+ if (!consumerGroupId.equals(stateKey.getConsumerGroupId())
+ || retainedTopicNames.contains(stateKey.getTopicName())) {
continue;
}
final ConsensusGroupId groupId =
@@ -971,8 +1000,11 @@ public class ConsensusSubscriptionSetupHandler {
consumerGroupId, stateKey.getTopicName(), groupId, serverImpl);
}
}
+ commitManager.removeState(consumerGroupId, stateKey.getTopicName(),
groupId);
+ }
+ if (retainedTopicNames.isEmpty()) {
+ commitManager.removeAllStatesForConsumerGroup(consumerGroupId);
}
- commitManager.removeAllStatesForConsumerGroup(consumerGroupId);
}
public static void refreshDetachedRetentionsForTopic(
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 cc30193647a..a0696342501 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
@@ -33,6 +33,7 @@ import
org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager;
import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission;
+import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRegistry;
import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
@@ -280,6 +281,76 @@ public class ConsensusPrefetchingQueueTest {
assertCloseDetachesBlockedConsensusRegistration(true);
}
+ @Test
+ public void testRetainingAsyncCloseDoesNotRecreateCancelledWalRetention()
throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final SubscriptionQueueRegistry registry = new
SubscriptionQueueRegistry("DataRegion[1]");
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(new
FakeConsensusReqReader());
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.isTableModel()).thenReturn(true);
+ doAnswer(
+ invocation -> {
+ registry.register(
+ invocation.getArgument(0),
+ invocation.getArgument(1),
+ invocation.getArgument(2),
+ invocation.getArgument(3));
+ return null;
+ })
+ .when(serverImpl)
+ .registerSubscriptionQueue(any(), any(), any(), any());
+ doAnswer(
+ invocation -> {
+ registry.replaceQueueWithDetachedRetention(
+ invocation.getArgument(0),
+ invocation.getArgument(1),
+ invocation.getArgument(2),
+ invocation.getArgument(3));
+ return null;
+ })
+ .when(serverImpl)
+ .replaceSubscriptionQueueWithDetachedRetention(any(), any(), any(),
any());
+ ConsensusPrefetchingQueue queue = null;
+ ReentrantReadWriteLock queueLock = null;
+ try {
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy("topic", -1L, -1L),
+ converter,
+
newCommitManager(temporaryFolder.newFolder("cancelled-retention")),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+ final Field lockField =
ConsensusPrefetchingQueue.class.getDeclaredField("lock");
+ lockField.setAccessible(true);
+ queueLock = (ReentrantReadWriteLock) lockField.get(queue);
+ queueLock.readLock().lock();
+ queue.closeAsync(false, true);
+ registry.unregisterDetachedRetention(
+ ConsensusSubscriptionWalRetention.generateRetentionId(
+ "consumerGroup", "topic", new DataRegionId(1)));
+ queueLock.readLock().unlock();
+ awaitCloseFinished(queue);
+ assertTrue(registry.isEmpty());
+ } finally {
+ if (queueLock != null && queueLock.getReadHoldCount() > 0) {
+ queueLock.readLock().unlock();
+ }
+ if (queue != null) {
+ queue.close();
+ awaitCloseFinished(queue);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
private void assertCloseDetachesBlockedConsensusRegistration(
final boolean retainProgressAfterUnsubscribe) throws Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -610,7 +681,7 @@ public class ConsensusPrefetchingQueueTest {
return null;
})
.when(serverImpl)
- .registerSubscriptionQueue(any(), any(), any());
+ .registerSubscriptionQueue(any(), any(), any(), any());
final ConsensusSubscriptionCommitManager commitManager =
mock(ConsensusSubscriptionCommitManager.class);
@@ -702,7 +773,7 @@ public class ConsensusPrefetchingQueueTest {
return null;
})
.when(serverImpl)
- .registerSubscriptionQueue(any(), any(), any());
+ .registerSubscriptionQueue(any(), any(), any(), any());
final ConsensusSubscriptionCommitManager commitManager =
mock(ConsensusSubscriptionCommitManager.class);
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 a0b16d9b7c1..934ef58dd20 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
@@ -30,9 +30,12 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class CommitProgressKeeper {
@@ -97,38 +100,75 @@ public class CommitProgressKeeper {
public synchronized void removeTopicProgress(
final String consumerGroupId, final String topicName) {
+ final Set<String> prefixes = getTopicProgressKeyPrefixes(consumerGroupId,
topicName);
+ regionProgressMap
+ .keySet()
+ .removeIf(
+ key ->
+ prefixes.stream()
+ .anyMatch(prefix ->
isValidRegionAndDataNodeProgressKey(key, prefix)));
+ }
+
+ public synchronized boolean hasTopicProgress(
+ final String consumerGroupId, final String topicName) {
+ final Set<String> prefixes = getTopicProgressKeyPrefixes(consumerGroupId,
topicName);
+ return regionProgressMap.keySet().stream()
+ .anyMatch(
+ key ->
+ prefixes.stream()
+ .anyMatch(prefix ->
isValidRegionAndDataNodeProgressKey(key, prefix)));
+ }
+
+ private static Set<String> getTopicProgressKeyPrefixes(
+ final String consumerGroupId, final String topicName) {
+ final Set<String> prefixes = new HashSet<>();
final String versionedTopicKeyPrefix =
VERSIONED_KEY_PREFIX
+ encodeKeyComponent(consumerGroupId)
+ KEY_COMPONENT_SEPARATOR
+ encodeKeyComponent(topicName)
+ KEY_COMPONENT_SEPARATOR;
- final String legacyTopicKeyPrefix =
- String.valueOf(consumerGroupId) + KEY_SEPARATOR +
String.valueOf(topicName) + KEY_SEPARATOR;
- final boolean legacyTopicKeyIsUnambiguous =
- isLegacyKeyUnambiguous(consumerGroupId, topicName, "");
- regionProgressMap
- .keySet()
- .removeIf(
- key ->
- isValidRegionAndDataNodeProgressKey(key,
versionedTopicKeyPrefix)
- || (legacyTopicKeyIsUnambiguous
- && isValidRegionAndDataNodeProgressKey(key,
legacyTopicKeyPrefix)));
+ prefixes.add(versionedTopicKeyPrefix);
+ if (isLegacyKeyUnambiguous(consumerGroupId, topicName, "")) {
+ prefixes.add(consumerGroupId + KEY_SEPARATOR + topicName +
KEY_SEPARATOR);
+ }
+ return prefixes;
}
public synchronized void removeConsumerGroupProgress(final String
consumerGroupId) {
+ removeConsumerGroupProgressExceptTopics(consumerGroupId,
Collections.emptySet());
+ }
+
+ public synchronized void removeConsumerGroupProgressExceptTopics(
+ final String consumerGroupId, final Set<String> retainedTopicNames) {
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;
+ final Set<String> retainedPrefixes = new HashSet<>();
+ retainedTopicNames.forEach(
+ topic ->
retainedPrefixes.addAll(getTopicProgressKeyPrefixes(consumerGroupId, topic)));
regionProgressMap
.keySet()
.removeIf(
key ->
- key.startsWith(versionedConsumerGroupKeyPrefix)
- || (legacyConsumerGroupKeyPrefix != null
- && key.startsWith(legacyConsumerGroupKeyPrefix)));
+ (key.startsWith(versionedConsumerGroupKeyPrefix)
+ || (legacyConsumerGroupKeyPrefix != null
+ && key.startsWith(legacyConsumerGroupKeyPrefix)))
+ && retainedPrefixes.stream()
+ .noneMatch(prefix ->
isValidRegionAndDataNodeProgressKey(key, prefix)));
+ }
+
+ public synchronized boolean hasConsumerGroupProgress(final String
consumerGroupId) {
+ final String versionedPrefix =
+ VERSIONED_KEY_PREFIX + encodeKeyComponent(consumerGroupId) +
KEY_COMPONENT_SEPARATOR;
+ final String legacyPrefix = consumerGroupId + KEY_SEPARATOR;
+ return regionProgressMap.keySet().stream()
+ .anyMatch(
+ key ->
+ key.startsWith(versionedPrefix)
+ || (!consumerGroupId.contains(KEY_SEPARATOR) &&
key.startsWith(legacyPrefix)));
}
private static boolean isValidRegionAndDataNodeProgressKey(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
index 2efffb89182..a129c5ae70e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java
@@ -52,6 +52,7 @@ public class ConsumerGroupMeta {
private long creationTime;
private Map<String, Set<String>> topicNameToSubscribedConsumerIdSet;
private Map<String, ConsumerMeta> consumerIdToConsumerMeta;
+ private boolean tableModel;
private Map<String, Long> topicNameToSubscriptionCreationTime; // used when
creationTime < 0
public ConsumerGroupMeta() {
@@ -68,6 +69,7 @@ public class ConsumerGroupMeta {
this.creationTime = -creationTime;
consumerIdToConsumerMeta.put(firstConsumerMeta.getConsumerId(),
firstConsumerMeta);
+ tableModel = isTableConsumer(firstConsumerMeta);
}
public ConsumerGroupMeta deepCopy() {
@@ -80,6 +82,7 @@ public class ConsumerGroupMeta {
copied.topicNameToSubscribedConsumerIdSet.put(
topicName, new HashSet<>(subscribedConsumerIds)));
copied.consumerIdToConsumerMeta = new
ConcurrentHashMap<>(consumerIdToConsumerMeta);
+ copied.tableModel = tableModel;
copied.topicNameToSubscriptionCreationTime =
new ConcurrentHashMap<>(topicNameToSubscriptionCreationTime);
return copied;
@@ -145,11 +148,10 @@ public class ConsumerGroupMeta {
public void checkAuthorityBeforeJoinConsumerGroup(final ConsumerMeta
consumerMeta)
throws SubscriptionException {
- if (isEmpty()) {
- return;
- }
- final ConsumerMeta existedConsumerMeta =
consumerIdToConsumerMeta.values().iterator().next();
- final String expectedSqlDialect =
existedConsumerMeta.getConfig().getSqlDialect();
+ final String expectedSqlDialect =
+ visibleUnder(true)
+ ? SystemConstant.SQL_DIALECT_TABLE_VALUE
+ : SystemConstant.SQL_DIALECT_TREE_VALUE;
final String actualSqlDialect = consumerMeta.getConfig().getSqlDialect();
final boolean isExpectedTableModel =
SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(expectedSqlDialect);
@@ -169,6 +171,7 @@ public class ConsumerGroupMeta {
}
public void addConsumer(final ConsumerMeta consumerMeta) {
+ tableModel = isTableConsumer(consumerMeta);
consumerIdToConsumerMeta.put(consumerMeta.getConsumerId(), consumerMeta);
}
@@ -200,12 +203,12 @@ public class ConsumerGroupMeta {
}
public boolean visibleUnder(final boolean isTableModel) {
- if (consumerIdToConsumerMeta.isEmpty()) {
- return false;
- }
- final String sqlDialect =
-
consumerIdToConsumerMeta.values().iterator().next().getConfig().getSqlDialect();
- return isTableModel ==
SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(sqlDialect);
+ return isTableModel == tableModel;
+ }
+
+ private static boolean isTableConsumer(final ConsumerMeta consumerMeta) {
+ return SystemConstant.SQL_DIALECT_TABLE_VALUE.equalsIgnoreCase(
+ consumerMeta.getConfig().getSqlDialect());
}
////////////////////////// subscription //////////////////////////
@@ -345,7 +348,11 @@ public class ConsumerGroupMeta {
}
}
- ReadWriteIOUtils.write(consumerIdToConsumerMeta.size(), outputStream);
+ // A negative empty-consumer count preserves the dialect of a retained
table group. Older
+ // readers also treat this count as an empty map, without changing the
following fields.
+ ReadWriteIOUtils.write(
+ consumerIdToConsumerMeta.isEmpty() && tableModel ? -1 :
consumerIdToConsumerMeta.size(),
+ outputStream);
for (final Map.Entry<String, ConsumerMeta> entry :
consumerIdToConsumerMeta.entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), outputStream);
entry.getValue().serialize(outputStream);
@@ -382,10 +389,12 @@ public class ConsumerGroupMeta {
consumerGroupMeta.consumerIdToConsumerMeta = new ConcurrentHashMap<>();
size = ReadWriteIOUtils.readInt(inputStream);
+ consumerGroupMeta.tableModel = size < 0;
for (int i = 0; i < size; ++i) {
final String key = ReadWriteIOUtils.readString(inputStream);
final ConsumerMeta value = ConsumerMeta.deserialize(inputStream);
consumerGroupMeta.consumerIdToConsumerMeta.put(key, value);
+ consumerGroupMeta.tableModel = isTableConsumer(value);
}
consumerGroupMeta.topicNameToSubscriptionCreationTime = new
ConcurrentHashMap<>();
@@ -423,10 +432,12 @@ public class ConsumerGroupMeta {
consumerGroupMeta.consumerIdToConsumerMeta = new ConcurrentHashMap<>();
size = ReadWriteIOUtils.readInt(byteBuffer);
+ consumerGroupMeta.tableModel = size < 0;
for (int i = 0; i < size; ++i) {
final String key = ReadWriteIOUtils.readString(byteBuffer);
final ConsumerMeta value = ConsumerMeta.deserialize(byteBuffer);
consumerGroupMeta.consumerIdToConsumerMeta.put(key, value);
+ consumerGroupMeta.tableModel = isTableConsumer(value);
}
consumerGroupMeta.topicNameToSubscriptionCreationTime = new
ConcurrentHashMap<>();
@@ -455,6 +466,7 @@ public class ConsumerGroupMeta {
final ConsumerGroupMeta that = (ConsumerGroupMeta) obj;
return Objects.equals(this.consumerGroupId, that.consumerGroupId)
&& this.creationTime == that.creationTime
+ && this.tableModel == that.tableModel
&& Objects.equals(
this.topicNameToSubscribedConsumerIdSet,
that.topicNameToSubscribedConsumerIdSet)
&& Objects.equals(this.consumerIdToConsumerMeta,
that.consumerIdToConsumerMeta)
@@ -467,6 +479,7 @@ public class ConsumerGroupMeta {
return Objects.hash(
consumerGroupId,
creationTime,
+ tableModel,
topicNameToSubscribedConsumerIdSet,
consumerIdToConsumerMeta,
topicNameToSubscriptionCreationTime);
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
index 589571ab2b2..2d2014d7e97 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
import org.junit.Assert;
import org.junit.Test;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
@@ -38,6 +39,44 @@ import static org.junit.Assert.fail;
public class ConsumerGroupDeSerTest {
+ @Test
+ public void testEmptyGroupPreservesDialectAcrossSerializationAndRejoin()
throws Exception {
+ for (final String dialect : new String[] {"tree", "table"}) {
+ final Map<String, String> attributes =
+ Collections.singletonMap(ConsumerConstant.SQL_DIALECT_KEY, dialect);
+ final ConsumerGroupMeta group =
+ new ConsumerGroupMeta("retained", 1L, new ConsumerMeta("consumer",
1L, attributes));
+ group.removeConsumer("consumer");
+ final ByteBuffer buffer = group.serialize();
+ final byte[] bytes = new byte[buffer.remaining()];
+ buffer.duplicate().get(bytes);
+ final ConsumerGroupMeta[] restoredGroups = {
+ group.deepCopy(),
+ ConsumerGroupMeta.deserialize(buffer),
+ ConsumerGroupMeta.deserialize(new ByteArrayInputStream(bytes))
+ };
+ for (final ConsumerGroupMeta restored : restoredGroups) {
+ Assert.assertEquals(group, restored);
+ Assert.assertTrue(restored.isEmpty());
+ Assert.assertEquals("table".equals(dialect),
restored.visibleUnder(true));
+ final ConsumerMeta replacement = new ConsumerMeta("replacement", 2L,
attributes);
+ restored.checkAuthorityBeforeJoinConsumerGroup(replacement);
+ Assert.assertThrows(
+ SubscriptionException.class,
+ () ->
+ restored.checkAuthorityBeforeJoinConsumerGroup(
+ new ConsumerMeta(
+ "other",
+ 2L,
+ Collections.singletonMap(
+ ConsumerConstant.SQL_DIALECT_KEY,
+ "table".equals(dialect) ? "tree" : "table"))));
+ restored.addConsumer(replacement);
+ Assert.assertEquals("table".equals(dialect),
restored.visibleUnder(true));
+ }
+ }
+ }
+
@Test
public void test() throws IOException {
Map<String, String> consumerAttributes = new HashMap<>();
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 7f18e4875d1..2ca78d29264 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
@@ -35,6 +35,7 @@ import java.lang.reflect.Modifier;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -47,6 +48,31 @@ import static org.junit.Assert.assertTrue;
public class CommitProgressKeeperTest {
+ @Test
+ public void testConsumerGroupCleanupRetainsOnlyConfiguredTopics() {
+ final CommitProgressKeeper keeper = new CommitProgressKeeper();
+ final String retained = CommitProgressKeeper.generateKey("group",
"kept##topic", "1_1", 1);
+ final String retainedLegacy =
CommitProgressKeeper.generateLegacyKey("group", "kept", "1_2", 2);
+ final String discarded = CommitProgressKeeper.generateKey("group",
"discarded", "1_1", 1);
+ final String otherGroup = CommitProgressKeeper.generateKey("other",
"kept", "1_1", 1);
+ for (final String key : new String[] {retained, retainedLegacy, discarded,
otherGroup}) {
+ keeper.updateRegionProgress(key, ByteBuffer.wrap(new byte[] {1}));
+ }
+ keeper.removeConsumerGroupProgressExceptTopics(
+ "group", new java.util.HashSet<>(java.util.Arrays.asList("kept",
"kept##topic")));
+ assertTrue(keeper.hasTopicProgress("group", "kept##topic"));
+ assertTrue(keeper.hasTopicProgress("group", "kept"));
+ assertFalse(keeper.hasTopicProgress("group", "discarded"));
+ assertTrue(keeper.hasConsumerGroupProgress("group"));
+ assertNotNull(keeper.getRegionProgress(retained));
+ assertNotNull(keeper.getRegionProgress(retainedLegacy));
+ assertNull(keeper.getRegionProgress(discarded));
+ assertNotNull(keeper.getRegionProgress(otherGroup));
+ keeper.removeConsumerGroupProgressExceptTopics("group",
Collections.emptySet());
+ assertFalse(keeper.hasConsumerGroupProgress("group"));
+ assertNotNull(keeper.getRegionProgress(otherGroup));
+ }
+
@Test
public void testStateAccessUsesOneInstanceMonitor() throws Exception {
assertSynchronized("removeTopicProgress", String.class, String.class);
