Skip to content

Commit b94aa09

Browse files
CaideyipiCaideyipi
authored
[Fix] Prevent consensus subscription teardown from blocking unsubscribe (#18804)
* [Fix] Prevent consensus subscription teardown from blocking unsubscribe * Add consensus subscription unsubscribe integration test
1 parent 21d53dd commit b94aa09

8 files changed

Lines changed: 560 additions & 47 deletions

File tree

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

Lines changed: 253 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,253 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.subscription.it.consensus.local.tablemodel;
21+
22+
import org.apache.iotdb.consensus.ConsensusFactory;
23+
import org.apache.iotdb.isession.ITableSession;
24+
import org.apache.iotdb.it.env.EnvFactory;
25+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
26+
import org.apache.iotdb.itbase.category.TableClusterIT;
27+
import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
28+
import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;
29+
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumer;
30+
import org.apache.iotdb.session.subscription.model.Subscription;
31+
import org.apache.iotdb.subscription.it.AbstractSubscriptionIT;
32+
33+
import org.awaitility.Awaitility;
34+
import org.junit.After;
35+
import org.junit.Assert;
36+
import org.junit.Before;
37+
import org.junit.Test;
38+
import org.junit.experimental.categories.Category;
39+
import org.junit.runner.RunWith;
40+
41+
import java.time.Duration;
42+
import java.util.Arrays;
43+
import java.util.HashSet;
44+
import java.util.Locale;
45+
import java.util.Set;
46+
import java.util.concurrent.Callable;
47+
import java.util.concurrent.CountDownLatch;
48+
import java.util.concurrent.ExecutorService;
49+
import java.util.concurrent.Executors;
50+
import java.util.concurrent.Future;
51+
import java.util.concurrent.TimeUnit;
52+
import java.util.concurrent.atomic.AtomicBoolean;
53+
54+
@RunWith(IoTDBTestRunner.class)
55+
@Category({TableClusterIT.class})
56+
public class IoTDBConsensusSubscriptionUnsubscribeTableIT extends AbstractSubscriptionIT {
57+
58+
private static final long MAX_UNSUBSCRIBE_MS = 30_000L;
59+
60+
@Override
61+
@Before
62+
public void setUp() throws Exception {
63+
super.setUp();
64+
EnvFactory.getEnv()
65+
.getConfig()
66+
.getCommonConfig()
67+
.setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
68+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
69+
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
70+
.setSchemaReplicationFactor(1)
71+
.setDataReplicationFactor(1)
72+
.setAutoCreateSchemaEnabled(true)
73+
.setPipeMemoryManagementEnabled(false)
74+
.setIsPipeEnableMemoryCheck(false);
75+
EnvFactory.getEnv().initClusterEnvironment(3, 1);
76+
}
77+
78+
@Override
79+
@After
80+
public void tearDown() throws Exception {
81+
EnvFactory.getEnv().cleanClusterEnvironment();
82+
super.tearDown();
83+
}
84+
85+
@Test
86+
public void testLastConsumerUnsubscribeDuringWritesAndImmediateResubscribe() throws Exception {
87+
final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
88+
ConsensusSubscriptionTableITSupport.newIdentifiers("unsubscribe_during_writes");
89+
final String database = ids.getDatabase();
90+
final String table = "table_0";
91+
final String firstConsumerId = ids.consumer("first");
92+
final String lastConsumerId = ids.consumer("last");
93+
final ExecutorService writer = Executors.newSingleThreadExecutor();
94+
final ExecutorService unsubscribeExecutor =
95+
Executors.newSingleThreadExecutor(
96+
runnable -> {
97+
final Thread thread = new Thread(runnable, "subscription-unsubscribe-it");
98+
thread.setDaemon(true);
99+
return thread;
100+
});
101+
final CountDownLatch writesStarted = new CountDownLatch(1);
102+
final AtomicBoolean keepWriting = new AtomicBoolean(true);
103+
SubscriptionTablePullConsumer consumer1 = null;
104+
SubscriptionTablePullConsumer consumer2 = null;
105+
Future<?> writeFuture = null;
106+
107+
try {
108+
ConsensusSubscriptionTableITSupport.bootstrapDatabaseAndTable(
109+
database, table, ConsensusSubscriptionTableITSupport.DEFAULT_TABLE_SCHEMA);
110+
ConsensusSubscriptionTableITSupport.createConsensusTopic(ids.getTopic(), database, table);
111+
consumer1 =
112+
ConsensusSubscriptionTableITSupport.createConsumer(
113+
firstConsumerId, ids.getConsumerGroupId());
114+
consumer2 =
115+
ConsensusSubscriptionTableITSupport.createConsumer(
116+
lastConsumerId, ids.getConsumerGroupId());
117+
consumer1.subscribe(ids.getTopic());
118+
consumer2.subscribe(ids.getTopic());
119+
awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId(), firstConsumerId, lastConsumerId);
120+
121+
writeFuture =
122+
writer.submit(
123+
(Callable<Void>)
124+
() -> {
125+
writeWhileSubscribed(database, table, keepWriting, writesStarted);
126+
return null;
127+
});
128+
Assert.assertTrue("Writer did not start", writesStarted.await(30, TimeUnit.SECONDS));
129+
if (writeFuture.isDone()) {
130+
writeFuture.get();
131+
}
132+
Assert.assertFalse("Writer finished before unsubscribe", writeFuture.isDone());
133+
consumer1.poll(Duration.ofMillis(100));
134+
135+
unsubscribeWithin(consumer1, ids.getTopic(), unsubscribeExecutor);
136+
awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId(), lastConsumerId);
137+
Assert.assertFalse("Writer finished before last unsubscribe", writeFuture.isDone());
138+
unsubscribeWithin(consumer2, ids.getTopic(), unsubscribeExecutor);
139+
awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId());
140+
141+
// A previous queue may still be closing. The new queue must retain its task registration.
142+
consumer1.subscribe(ids.getTopic());
143+
awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId(), firstConsumerId);
144+
145+
keepWriting.set(false);
146+
writeFuture.get(30, TimeUnit.SECONDS);
147+
final Set<String> expectedRows =
148+
ConsensusSubscriptionTableITSupport.insertRows(database, table, 1_000_000_000L, 20, true);
149+
final ConsensusSubscriptionTableITSupport.ConsumedRecords consumed =
150+
ConsensusSubscriptionTableITSupport.pollAndCommitUntilContains(
151+
consumer1, expectedRows, 60);
152+
Assert.assertTrue(consumed.toString(), consumed.getRowKeys().containsAll(expectedRows));
153+
} finally {
154+
keepWriting.set(false);
155+
if (writeFuture != null && !writeFuture.isDone()) {
156+
writeFuture.cancel(true);
157+
}
158+
writer.shutdownNow();
159+
writer.awaitTermination(10, TimeUnit.SECONDS);
160+
unsubscribeExecutor.shutdownNow();
161+
closeQuietly(consumer2);
162+
closeQuietly(consumer1);
163+
ConsensusSubscriptionTableITSupport.cleanup(null, ids.getTopic(), database);
164+
}
165+
}
166+
167+
private static void writeWhileSubscribed(
168+
final String database,
169+
final String table,
170+
final AtomicBoolean keepWriting,
171+
final CountDownLatch writesStarted)
172+
throws Exception {
173+
try (final ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
174+
session.executeNonQueryStatement("use " + database);
175+
for (long row = 0; keepWriting.get(); row++) {
176+
final long timestamp = 1_000L + row;
177+
session.executeNonQueryStatement(
178+
String.format(
179+
Locale.ROOT,
180+
"insert into %s(tag1, s1, time) values ('writing', %d, %d)",
181+
table,
182+
timestamp,
183+
timestamp));
184+
if (row == 20) {
185+
writesStarted.countDown();
186+
}
187+
Thread.sleep(10L);
188+
}
189+
session.executeNonQueryStatement("flush");
190+
} finally {
191+
writesStarted.countDown();
192+
}
193+
}
194+
195+
private static void unsubscribeWithin(
196+
final SubscriptionTablePullConsumer consumer,
197+
final String topicName,
198+
final ExecutorService executor)
199+
throws Exception {
200+
final Future<?> future = executor.submit(() -> consumer.unsubscribe(topicName));
201+
try {
202+
future.get(MAX_UNSUBSCRIBE_MS, TimeUnit.MILLISECONDS);
203+
} finally {
204+
future.cancel(true);
205+
}
206+
}
207+
208+
private static void closeQuietly(final SubscriptionTablePullConsumer consumer) {
209+
if (consumer != null) {
210+
try {
211+
consumer.close();
212+
} catch (final Exception ignored) {
213+
// Best effort cleanup after a failed assertion.
214+
}
215+
}
216+
}
217+
218+
private static void awaitConsumerIds(
219+
final String topicName, final String consumerGroupId, final String... expectedConsumerIds)
220+
throws Exception {
221+
try (final ISubscriptionTableSession session =
222+
new SubscriptionTableSessionBuilder()
223+
.host(EnvFactory.getEnv().getIP())
224+
.port(Integer.parseInt(EnvFactory.getEnv().getPort()))
225+
.build()) {
226+
session.open();
227+
Awaitility.await()
228+
.pollInSameThread()
229+
.pollInterval(Duration.ofMillis(500))
230+
.atMost(Duration.ofSeconds(30))
231+
.untilAsserted(
232+
() -> {
233+
final Set<Subscription> subscriptions = session.getSubscriptions(topicName);
234+
if (expectedConsumerIds.length == 0) {
235+
Assert.assertTrue(subscriptions.toString(), subscriptions.isEmpty());
236+
} else {
237+
Assert.assertEquals(subscriptions.toString(), 1, subscriptions.size());
238+
final Subscription subscription = subscriptions.iterator().next();
239+
Assert.assertEquals(consumerGroupId, subscription.getConsumerGroupId());
240+
final String consumerIds = subscription.getConsumerIds();
241+
Assert.assertTrue(
242+
subscription.toString(),
243+
consumerIds.startsWith("[") && consumerIds.endsWith("]"));
244+
final String idList = consumerIds.substring(1, consumerIds.length() - 1);
245+
Assert.assertEquals(
246+
subscription.toString(),
247+
new HashSet<>(Arrays.asList(expectedConsumerIds)),
248+
new HashSet<>(Arrays.asList(idList.split(",\\s*"))));
249+
}
250+
});
251+
}
252+
}
253+
}

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2188,6 +2188,8 @@ private DataNodePipeMessages() {}
21882188
+ "frontier";
21892189
public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_DURING_DEREGISTER_34C332E7 =
21902190
"ConsensusPrefetchingQueue {}: error during deregister";
2191+
public static final String LOG_FAILED_TO_FINISH_CLOSING_CONSENSUS_PREFETCHING_QUEUE_ARG_3C31731C =
2192+
"Failed to finish closing consensus prefetching queue {}";
21912193
public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_FLUSH_LINGERING_BATCH_F97D8AA7 =
21922194
"ConsensusPrefetchingQueue {}: failed to flush lingering batch during close, discarding it";
21932195
public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_ROUND_FAILED_TYPE_MESSAGE_63BC909B =

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2031,6 +2031,8 @@ private DataNodePipeMessages() {}
20312031
"ConsensusPrefetchingQueue {}：不支持 seekAfterRegionProgress（没有 WAL 目录）";
20322032
public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_READ_WAL_METADATA_FROM_A2ED50D1 =
20332033
"ConsensusPrefetchingQueue {}：计算 seekToEnd frontier 时，从 {} 读取 WAL metadata 失败";
2034+
public static final String LOG_FAILED_TO_FINISH_CLOSING_CONSENSUS_PREFETCHING_QUEUE_ARG_3C31731C =
2035+
"未能完成共识预取队列 {} 的关闭";
20342036
public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_DURING_DEREGISTER_34C332E7 =
20352037
"ConsensusPrefetchingQueue {}：注销期间出错";
20362038
public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_FLUSH_LINGERING_BATCH_F97D8AA7 =

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -743,7 +743,10 @@ public void refreshConsensusQueueOrderMode(final String topicName, final String
743743
}
744744

745745
public void unbindConsensusPrefetchingQueue(final String topicName) {
746-
closeAndRemoveConsensusPrefetchingQueues(topicName, true, true);
746+
// Subscription setup teardown and rollback manage commit state after this call returns. Let
747+
// queue close persist the current state, while avoiding an asynchronous close deleting state
748+
// restored by a concurrent setup rollback.
749+
closeAndRemoveConsensusPrefetchingQueues(topicName, true, false);
747750
}
748751

749752
@Override
@@ -774,7 +777,7 @@ public int unbindByRegion(final ConsensusGroupId regionId) {
774777
}
775778

776779
for (int i = 0; i < queuesToClose.size(); i++) {
777-
queuesToClose.get(i).close();
780+
queuesToClose.get(i).closeAsync(false);
778781
LOGGER.info(
779782
DataNodePipeMessages
780783
.PIPE_LOG_SUBSCRIPTION_CLOSED_CONSENSUS_PREFETCHING_QUEUE_FOR_TOPIC_3A9DDEC5,
@@ -874,7 +877,7 @@ private void closeAndRemoveConsensusPrefetchingQueues(
874877
}
875878

876879
for (final ConsensusPrefetchingQueue q : queuesToClose) {
877-
q.close(removeProgressAfterClose);
880+
q.closeAsync(removeProgressAfterClose);
878881
}
879882
LOGGER.info(
880883
DataNodePipeMessages

0 commit comments

Comments
 (0)