Skip to content

Commit 9b807ab

Browse files
CaideyipiCaideyipi
authored
[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
1 parent e492e44 commit 9b807ab

14 files changed

Lines changed: 590 additions & 68 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionBasicTableIT.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,12 @@
1919

2020
package org.apache.iotdb.subscription.it.consensus.local.tablemodel;
2121

22+
import org.apache.iotdb.it.env.EnvFactory;
2223
import org.apache.iotdb.it.framework.IoTDBTestRunner;
2324
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
25+
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
26+
import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
27+
import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
2428
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumer;
2529
import org.apache.iotdb.subscription.it.consensus.local.AbstractSubscriptionConsensusLocalIT;
2630

@@ -31,12 +35,68 @@
3135

3236
import java.time.Duration;
3337
import java.util.LinkedHashSet;
38+
import java.util.Properties;
3439
import java.util.Set;
3540

3641
@RunWith(IoTDBTestRunner.class)
3742
@Category({LocalStandaloneIT.class})
3843
public class IoTDBConsensusSubscriptionBasicTableIT extends AbstractSubscriptionConsensusLocalIT {
3944

45+
@Test
46+
public void testLastConsumerCloseRetainsProgress() throws Exception {
47+
for (final boolean unsubscribeFirst : new boolean[] {false, true}) {
48+
final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
49+
ConsensusSubscriptionTableITSupport.newIdentifiers("retained_last_close");
50+
SubscriptionTablePullConsumer consumer = null;
51+
try {
52+
ConsensusSubscriptionTableITSupport.bootstrapDatabaseAndTable(
53+
ids.getDatabase(), "t1", ConsensusSubscriptionTableITSupport.DEFAULT_TABLE_SCHEMA);
54+
ConsensusSubscriptionTableITSupport.createConsensusTopic(
55+
ids.getTopic(), ids.getDatabase(), "t1");
56+
try (final ISubscriptionTableSession session =
57+
new SubscriptionTableSessionBuilder()
58+
.host(EnvFactory.getEnv().getIP())
59+
.port(Integer.parseInt(EnvFactory.getEnv().getPort()))
60+
.build()) {
61+
session.open();
62+
final Properties config = new Properties();
63+
config.setProperty(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true");
64+
session.alterTopic(ids.getTopic(), config);
65+
}
66+
consumer =
67+
ConsensusSubscriptionTableITSupport.createConsumer(
68+
ids.getConsumerId(), ids.getConsumerGroupId());
69+
consumer.subscribe(ids.getTopic());
70+
final Set<String> firstRows =
71+
ConsensusSubscriptionTableITSupport.insertRows(ids.getDatabase(), "t1", 100L, 10, true);
72+
ConsensusSubscriptionTableITSupport.assertExactRowKeys(
73+
firstRows,
74+
ConsensusSubscriptionTableITSupport.pollAndCommitUntilAtLeast(
75+
consumer, firstRows.size(), 40));
76+
if (unsubscribeFirst) {
77+
consumer.unsubscribe(ids.getTopic());
78+
}
79+
consumer.close();
80+
consumer = null;
81+
82+
final Set<String> offlineRows =
83+
ConsensusSubscriptionTableITSupport.insertRows(ids.getDatabase(), "t1", 200L, 10, true);
84+
consumer =
85+
ConsensusSubscriptionTableITSupport.createConsumer(
86+
ids.consumer("replacement"), ids.getConsumerGroupId());
87+
consumer.subscribe(ids.getTopic());
88+
ConsensusSubscriptionTableITSupport.assertExactRowKeys(
89+
offlineRows,
90+
ConsensusSubscriptionTableITSupport.pollAndCommitUntilAtLeast(
91+
consumer, offlineRows.size(), 40));
92+
ConsensusSubscriptionTableITSupport.assertNoMoreMessages(
93+
consumer, 3, Duration.ofMillis(500));
94+
} finally {
95+
ConsensusSubscriptionTableITSupport.cleanup(consumer, ids.getTopic(), ids.getDatabase());
96+
}
97+
}
98+
}
99+
40100
@Test
41101
public void testRealtimeOnlyAfterSubscribe() throws Exception {
42102
final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java‎

Lines changed: 31 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -929,6 +929,7 @@ private TSStatus alterTopicInternal(final AlterTopicPlan plan) {
929929
plan.getTopicMeta().getTopicName())) {
930930
commitProgressKeeper.removeTopicProgress(
931931
consumerGroupMeta.getConsumerGroupId(), plan.getTopicMeta().getTopicName());
932+
removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta);
932933
}
933934
}
934935
}
@@ -971,6 +972,7 @@ public TSStatus dropTopic(DropTopicPlan plan) {
971972
if (!plan.isTableModelSet() || consumerGroupMeta.visibleUnder(plan.isTableModel())) {
972973
commitProgressKeeper.removeTopicProgress(
973974
consumerGroupMeta.getConsumerGroupId(), plan.getTopicName());
975+
removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta);
974976
}
975977
}
976978
if (plan.isTableModelSet()) {
@@ -1146,24 +1148,32 @@ public TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) {
11461148
? consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId)
11471149
: null;
11481150
if (Objects.nonNull(currentConsumerGroupMeta)) {
1151+
ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta, consumerGroupMeta)
1152+
.forEach(
1153+
topicName -> {
1154+
final TopicMeta topicMeta =
1155+
topicMetaKeeper.getTopicMeta(
1156+
topicName, currentConsumerGroupMeta.visibleUnder(true));
1157+
if (Objects.isNull(topicMeta)
1158+
|| !topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) {
1159+
commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName);
1160+
}
1161+
});
11491162
if (consumerGroupMeta.isEmpty()) {
1150-
commitProgressKeeper.removeConsumerGroupProgress(consumerGroupId);
1151-
} else {
1152-
ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta, consumerGroupMeta)
1153-
.forEach(
1154-
topicName -> {
1155-
final TopicMeta topicMeta =
1156-
topicMetaKeeper.getTopicMeta(
1157-
topicName, currentConsumerGroupMeta.visibleUnder(true));
1158-
if (Objects.isNull(topicMeta)
1159-
|| !topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) {
1160-
commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName);
1161-
}
1162-
});
1163+
final Set<String> retainedTopicNames = new HashSet<>();
1164+
for (final TopicMeta topicMeta : topicMetaKeeper.getAllTopicMeta()) {
1165+
if (topicMeta.visibleUnderTableModel() == currentConsumerGroupMeta.visibleUnder(true)
1166+
&& topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) {
1167+
retainedTopicNames.add(topicMeta.getTopicName());
1168+
}
1169+
}
1170+
commitProgressKeeper.removeConsumerGroupProgressExceptTopics(
1171+
consumerGroupId, retainedTopicNames);
11631172
}
11641173
}
11651174
consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId);
1166-
if (!consumerGroupMeta.isEmpty()) {
1175+
if (!consumerGroupMeta.isEmpty()
1176+
|| commitProgressKeeper.hasConsumerGroupProgress(consumerGroupId)) {
11671177
consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId, consumerGroupMeta);
11681178
}
11691179
}
@@ -1173,6 +1183,13 @@ public TSStatus alterConsumerGroup(AlterConsumerGroupPlan plan) {
11731183
}
11741184
}
11751185

1186+
private void removeEmptyConsumerGroupWithoutProgress(final ConsumerGroupMeta consumerGroupMeta) {
1187+
if (consumerGroupMeta.isEmpty()
1188+
&& !commitProgressKeeper.hasConsumerGroupProgress(consumerGroupMeta.getConsumerGroupId())) {
1189+
consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupMeta.getConsumerGroupId());
1190+
}
1191+
}
1192+
11761193
public TSStatus handleConsumerGroupMetaChanges(ConsumerGroupHandleMetaChangePlan plan) {
11771194
acquireWriteLock();
11781195
try {

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,13 @@
3434
import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp;
3535
import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
3636
import org.apache.iotdb.rpc.TSStatusCode;
37+
import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
3738
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
3839

3940
import org.junit.Assert;
41+
import org.junit.Rule;
4042
import org.junit.Test;
43+
import org.junit.rules.TemporaryFolder;
4144

4245
import java.nio.ByteBuffer;
4346
import java.util.Collections;
@@ -49,6 +52,102 @@
4952

5053
public class SubscriptionInfoTest {
5154

55+
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
56+
57+
@Test
58+
public void testLastConsumerClosePreservesConfiguredProgressAcrossSnapshot() throws Exception {
59+
for (final boolean tableModel : new boolean[] {false, true}) {
60+
for (final boolean unsubscribeFirst : new boolean[] {false, true}) {
61+
assertLastConsumerClosePreservesProgress(tableModel, unsubscribeFirst);
62+
}
63+
}
64+
}
65+
66+
private void assertLastConsumerClosePreservesProgress(
67+
final boolean tableModel, final boolean unsubscribeFirst) throws Exception {
68+
final SubscriptionInfo info = new SubscriptionInfo();
69+
final Map<String, String> attributes = new HashMap<>();
70+
attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE);
71+
attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true");
72+
attributes.put(SystemConstant.SQL_DIALECT_KEY, tableModel ? "table" : "tree");
73+
info.createTopic(new CreateTopicPlan(new TopicMeta("kept", 1L, attributes)));
74+
final Map<String, String> discardedAttributes = new HashMap<>(attributes);
75+
discardedAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "false");
76+
info.createTopic(new CreateTopicPlan(new TopicMeta("discarded", 1L, discardedAttributes)));
77+
final ConsumerGroupMeta group =
78+
new ConsumerGroupMeta(
79+
"group",
80+
1L,
81+
new ConsumerMeta(
82+
"consumer",
83+
1L,
84+
Collections.singletonMap(
85+
ConsumerConstant.SQL_DIALECT_KEY, tableModel ? "table" : "tree")));
86+
group.addSubscription(
87+
"consumer", new java.util.HashSet<>(java.util.Arrays.asList("kept", "discarded")));
88+
info.alterConsumerGroup(new AlterConsumerGroupPlan(group));
89+
final String keptKey = CommitProgressKeeper.generateKey("group", "kept", "DataRegion[1]", 1);
90+
final String legacyKey =
91+
CommitProgressKeeper.generateLegacyKey("group", "kept", "DataRegion[1]", 2);
92+
final String discardedKey =
93+
CommitProgressKeeper.generateKey("group", "discarded", "DataRegion[1]", 1);
94+
for (final String key : new String[] {keptKey, legacyKey, discardedKey}) {
95+
info.getCommitProgressKeeper().updateRegionProgress(key, ByteBuffer.wrap(new byte[] {1}));
96+
}
97+
final ConsumerGroupMeta updated = group.deepCopy();
98+
if (unsubscribeFirst) {
99+
updated.removeSubscription(
100+
"consumer", new java.util.HashSet<>(java.util.Arrays.asList("kept", "discarded")));
101+
info.alterConsumerGroup(new AlterConsumerGroupPlan(updated.deepCopy()));
102+
}
103+
updated.removeConsumer("consumer");
104+
info.alterConsumerGroup(new AlterConsumerGroupPlan(updated));
105+
Assert.assertNotNull(
106+
"tableModel=" + tableModel + ", unsubscribeFirst=" + unsubscribeFirst,
107+
info.getCommitProgressKeeper().getRegionProgress(keptKey));
108+
Assert.assertNotNull(info.getCommitProgressKeeper().getRegionProgress(legacyKey));
109+
Assert.assertNull(info.getCommitProgressKeeper().getRegionProgress(discardedKey));
110+
Assert.assertTrue(info.getConsumerGroupMeta("group").isEmpty());
111+
Assert.assertTrue(info.getConsumerGroupMeta("group").visibleUnder(tableModel));
112+
113+
final java.io.File snapshotDir = temporaryFolder.newFolder();
114+
Assert.assertTrue(info.processTakeSnapshot(snapshotDir));
115+
final SubscriptionInfo restored = new SubscriptionInfo();
116+
restored.processLoadSnapshot(snapshotDir);
117+
Assert.assertNotNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
118+
final ConsumerGroupMeta restoredGroup = restored.deepCopyConsumerGroupMeta("group");
119+
Assert.assertTrue(restoredGroup.isEmpty());
120+
Assert.assertTrue(restoredGroup.visibleUnder(tableModel));
121+
final ConsumerMeta replacement =
122+
new ConsumerMeta(
123+
"replacement",
124+
2L,
125+
Collections.singletonMap(
126+
ConsumerConstant.SQL_DIALECT_KEY, tableModel ? "table" : "tree"));
127+
restoredGroup.checkAuthorityBeforeJoinConsumerGroup(replacement);
128+
restoredGroup.addConsumer(replacement);
129+
restored.alterConsumerGroup(new AlterConsumerGroupPlan(restoredGroup));
130+
Assert.assertNotNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
131+
restoredGroup.removeConsumer("replacement");
132+
restored.alterConsumerGroup(new AlterConsumerGroupPlan(restoredGroup));
133+
134+
// Same-name topics in the other model must not remove this retained group's progress.
135+
final Map<String, String> otherAttributes = new HashMap<>(attributes);
136+
otherAttributes.put(SystemConstant.SQL_DIALECT_KEY, tableModel ? "tree" : "table");
137+
restored.createTopic(new CreateTopicPlan(new TopicMeta("kept", 2L, otherAttributes)));
138+
restored.dropTopic(new DropTopicPlan("kept", !tableModel));
139+
Assert.assertNotNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
140+
if (unsubscribeFirst) {
141+
attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "false");
142+
restored.alterTopic(new AlterTopicPlan(new TopicMeta("kept", 2L, attributes)));
143+
} else {
144+
restored.dropTopic(new DropTopicPlan("kept", tableModel));
145+
}
146+
Assert.assertNull(restored.getCommitProgressKeeper().getRegionProgress(keptKey));
147+
Assert.assertNull(restored.getCommitProgressKeeper().getRegionProgress(legacyKey));
148+
Assert.assertFalse(restored.isConsumerGroupExisted("group"));
149+
}
150+
52151
@Test
53152
public void testSameNameTreeAndTableTopicsCanBeManagedIndependently() {
54153
final String topicName = "same-name-topic";

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1267,7 +1267,20 @@ public void registerSubscriptionQueue(
12671267
final BlockingQueue<IndexedConsensusRequest> queue,
12681268
final SubscriptionWalRetentionPolicy retentionPolicy,
12691269
final LongSupplier retainedMinVersionIdSupplier) {
1270-
subscriptionQueueRegistry.register(queue, retentionPolicy, retainedMinVersionIdSupplier);
1270+
registerSubscriptionQueue(queue, retentionPolicy, retainedMinVersionIdSupplier, null);
1271+
}
1272+
1273+
public void registerSubscriptionQueue(
1274+
final BlockingQueue<IndexedConsensusRequest> queue,
1275+
final SubscriptionWalRetentionPolicy retentionPolicy,
1276+
final LongSupplier retainedMinVersionIdSupplier,
1277+
final String retentionId) {
1278+
if (retentionId == null) {
1279+
subscriptionQueueRegistry.register(queue, retentionPolicy, retainedMinVersionIdSupplier);
1280+
} else {
1281+
subscriptionQueueRegistry.register(
1282+
queue, retentionPolicy, retainedMinVersionIdSupplier, retentionId);
1283+
}
12711284
// Immediately re-evaluate the safe delete index with new subscription awareness
12721285
checkAndUpdateSafeDeletedSearchIndex();
12731286
logger.info(

0 commit comments

Comments
 (0)