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 1a64e70d4fd6..aeb1850882e6 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 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 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 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 58fe0a10690c..c0b6d10e848a 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 @@ private TSStatus alterTopicInternal(final AlterTopicPlan plan) { plan.getTopicMeta().getTopicName())) { commitProgressKeeper.removeTopicProgress( consumerGroupMeta.getConsumerGroupId(), plan.getTopicMeta().getTopicName()); + removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta); } } } @@ -971,6 +972,7 @@ public TSStatus dropTopic(DropTopicPlan plan) { if (!plan.isTableModelSet() || consumerGroupMeta.visibleUnder(plan.isTableModel())) { commitProgressKeeper.removeTopicProgress( consumerGroupMeta.getConsumerGroupId(), plan.getTopicName()); + removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta); } } if (plan.isTableModelSet()) { @@ -1146,24 +1148,32 @@ public TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) { ? 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 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 TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) { } } + 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 087cb5868f50..35cd29dc6178 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.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 @@ 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 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 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 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 ca29931f1c9e..aab4c7523cff 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 void registerSubscriptionQueue( final BlockingQueue queue, final SubscriptionWalRetentionPolicy retentionPolicy, final LongSupplier retainedMinVersionIdSupplier) { - subscriptionQueueRegistry.register(queue, retentionPolicy, retainedMinVersionIdSupplier); + registerSubscriptionQueue(queue, retentionPolicy, retainedMinVersionIdSupplier, null); + } + + public void registerSubscriptionQueue( + final BlockingQueue 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 995b876cbd1b..e75324a8854c 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 detachedRetentionRegistrations = new ConcurrentHashMap<>(); + private final Map> 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 synchronized void register( 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 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 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 synchronized void replaceQueueWithDetachedRetention( 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 synchronized void replaceDetachedRetentionWithQueue( 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 @@ private static final class SubscriptionQueueRegistration { 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 214cc2ec8f2b..86222dd1118e 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 @@ 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 7917fa39a5cf..8af488c5923b 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 TPushConsumerGroupMetaRespExceptionMessage handleSingleConsumerGroupMetaC 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 @@ private void handleSingleConsumerGroupMetaChangesInternal( 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 @@ private void handleDropConsumerGroupInternal(final String consumerGroupId) { 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 @@ private void handleDropConsumerGroupInternal(final String consumerGroupId) { .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 @@ private static final class ConsumerGroupOperationLock { 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 consensusTopics = new LinkedHashSet<>(); for (final String topicName : consumerGroupMeta.getSubscribedTopicNames()) { if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic( @@ -339,8 +359,10 @@ private void cleanupConsumerGroupSubscriptions( } } 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 4e71f2e2fe97..9143e40796c7 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 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 @@ private void deregisterPendingEntriesFromConsensusServer( 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 92cfade03c28..964bbcb22597 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 static void cleanupRetainedProgressForDroppedTopic( } public static void cleanupRetainedProgressForConsumerGroup(final String consumerGroupId) { + cleanupConsumerGroupProgressExceptTopics(consumerGroupId, Collections.emptySet()); + } + + public static boolean retainConfiguredProgressForConsumerGroup( + final String consumerGroupId, final boolean isTableModel) { + final Set topicNames = new LinkedHashSet<>(); + for (final ConsensusSubscriptionCommitManager.PersistedStateKey stateKey : + ConsensusSubscriptionCommitManager.getInstance().getPersistedStateKeys()) { + if (consumerGroupId.equals(stateKey.getConsumerGroupId())) { + topicNames.add(stateKey.getTopicName()); + } + } + final Set 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 retainedTopicNames) { final IConsensus dataRegionConsensus = DataRegionConsensusImpl.getInstance(); final IoTConsensus ioTConsensus = dataRegionConsensus instanceof IoTConsensus ? (IoTConsensus) dataRegionConsensus : null; @@ -959,7 +987,8 @@ public static void cleanupRetainedProgressForConsumerGroup(final String consumer 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 static void cleanupRetainedProgressForConsumerGroup(final String consumer 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 cc30193647af..a0696342501a 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.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 void testRetainingCloseDetachesBlockedConsensusQueueReplacement() throws 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 void testReplayCursorBoundsWalRetentionWhenCommittedProgressIsAhead() thr return null; }) .when(serverImpl) - .registerSubscriptionQueue(any(), any(), any()); + .registerSubscriptionQueue(any(), any(), any(), any()); final ConsensusSubscriptionCommitManager commitManager = mock(ConsensusSubscriptionCommitManager.class); @@ -702,7 +773,7 @@ public void testCommittedWalRetentionReusesUncoveredBoundaryRequirement() throws 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 a0b16d9b7c1a..934ef58dd204 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.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 static boolean isValidDataNodeProgressKey(final String key, final String public synchronized void removeTopicProgress( final String consumerGroupId, final String topicName) { + final Set 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 prefixes = getTopicProgressKeyPrefixes(consumerGroupId, topicName); + return regionProgressMap.keySet().stream() + .anyMatch( + key -> + prefixes.stream() + .anyMatch(prefix -> isValidRegionAndDataNodeProgressKey(key, prefix))); + } + + private static Set getTopicProgressKeyPrefixes( + final String consumerGroupId, final String topicName) { + final Set 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 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 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 2efffb89182a..a129c5ae70ea 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> topicNameToSubscribedConsumerIdSet; private Map consumerIdToConsumerMeta; + private boolean tableModel; private Map topicNameToSubscriptionCreationTime; // used when creationTime < 0 public ConsumerGroupMeta() { @@ -68,6 +69,7 @@ public ConsumerGroupMeta( this.creationTime = -creationTime; consumerIdToConsumerMeta.put(firstConsumerMeta.getConsumerId(), firstConsumerMeta); + tableModel = isTableConsumer(firstConsumerMeta); } public ConsumerGroupMeta deepCopy() { @@ -80,6 +82,7 @@ public ConsumerGroupMeta deepCopy() { 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 static Set getTopicsNewlySubByGroup( 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 void checkAuthorityBeforeJoinConsumerGroup(final ConsumerMeta consumerMet } public void addConsumer(final ConsumerMeta consumerMeta) { + tableModel = isTableConsumer(consumerMeta); consumerIdToConsumerMeta.put(consumerMeta.getConsumerId(), consumerMeta); } @@ -200,12 +203,12 @@ public ConsumerMeta getConsumerMeta(final String consumerId) { } 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 void serialize(final OutputStream outputStream) throws IOException { } } - 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 entry : consumerIdToConsumerMeta.entrySet()) { ReadWriteIOUtils.write(entry.getKey(), outputStream); entry.getValue().serialize(outputStream); @@ -382,10 +389,12 @@ public static ConsumerGroupMeta deserialize(final InputStream inputStream) throw 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 static ConsumerGroupMeta deserialize(final ByteBuffer byteBuffer) { 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 boolean equals(final Object obj) { 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 int hashCode() { 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 589571ab2b22..2d2014d7e976 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.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 @@ public class ConsumerGroupDeSerTest { + @Test + public void testEmptyGroupPreservesDialectAcrossSerializationAndRejoin() throws Exception { + for (final String dialect : new String[] {"tree", "table"}) { + final Map 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 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 7f18e4875d11..2ca78d29264b 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.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 @@ 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);