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 b94aa09ab66 [Fix] Prevent consensus subscription teardown from
blocking unsubscribe (#18804)
b94aa09ab66 is described below
commit b94aa09ab66a3b563c7c61c6bfb294152634fbce
Author: Caideyipi <[email protected]>
AuthorDate: Sat Oct 10 14:32:40 2026 +0800
[Fix] Prevent consensus subscription teardown from blocking unsubscribe
(#18804)
* [Fix] Prevent consensus subscription teardown from blocking unsubscribe
* Add consensus subscription unsubscribe integration test
---
...TDBConsensusSubscriptionUnsubscribeTableIT.java | 253 +++++++++++++++++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../broker/ConsensusSubscriptionBroker.java | 9 +-
.../consensus/ConsensusPrefetchingQueue.java | 147 ++++++++----
.../ConsensusSubscriptionPrefetchExecutor.java | 10 +
.../task/subtask/ConsensusPrefetchSubtask.java | 1 +
.../consensus/ConsensusPrefetchingQueueTest.java | 183 +++++++++++++++
8 files changed, 560 insertions(+), 47 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionUnsubscribeTableIT.java
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionUnsubscribeTableIT.java
new file mode 100644
index 00000000000..afa740b983e
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionUnsubscribeTableIT.java
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.subscription.it.consensus.local.tablemodel;
+
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+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.session.subscription.model.Subscription;
+import org.apache.iotdb.subscription.it.AbstractSubscriptionIT;
+
+import org.awaitility.Awaitility;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableClusterIT.class})
+public class IoTDBConsensusSubscriptionUnsubscribeTableIT extends
AbstractSubscriptionIT {
+
+ private static final long MAX_UNSUBSCRIBE_MS = 30_000L;
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaReplicationFactor(1)
+ .setDataReplicationFactor(1)
+ .setAutoCreateSchemaEnabled(true)
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
+ EnvFactory.getEnv().initClusterEnvironment(3, 1);
+ }
+
+ @Override
+ @After
+ public void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ super.tearDown();
+ }
+
+ @Test
+ public void testLastConsumerUnsubscribeDuringWritesAndImmediateResubscribe()
throws Exception {
+ final ConsensusSubscriptionTableITSupport.TestIdentifiers ids =
+
ConsensusSubscriptionTableITSupport.newIdentifiers("unsubscribe_during_writes");
+ final String database = ids.getDatabase();
+ final String table = "table_0";
+ final String firstConsumerId = ids.consumer("first");
+ final String lastConsumerId = ids.consumer("last");
+ final ExecutorService writer = Executors.newSingleThreadExecutor();
+ final ExecutorService unsubscribeExecutor =
+ Executors.newSingleThreadExecutor(
+ runnable -> {
+ final Thread thread = new Thread(runnable,
"subscription-unsubscribe-it");
+ thread.setDaemon(true);
+ return thread;
+ });
+ final CountDownLatch writesStarted = new CountDownLatch(1);
+ final AtomicBoolean keepWriting = new AtomicBoolean(true);
+ SubscriptionTablePullConsumer consumer1 = null;
+ SubscriptionTablePullConsumer consumer2 = null;
+ Future<?> writeFuture = null;
+
+ try {
+ ConsensusSubscriptionTableITSupport.bootstrapDatabaseAndTable(
+ database, table,
ConsensusSubscriptionTableITSupport.DEFAULT_TABLE_SCHEMA);
+ ConsensusSubscriptionTableITSupport.createConsensusTopic(ids.getTopic(),
database, table);
+ consumer1 =
+ ConsensusSubscriptionTableITSupport.createConsumer(
+ firstConsumerId, ids.getConsumerGroupId());
+ consumer2 =
+ ConsensusSubscriptionTableITSupport.createConsumer(
+ lastConsumerId, ids.getConsumerGroupId());
+ consumer1.subscribe(ids.getTopic());
+ consumer2.subscribe(ids.getTopic());
+ awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId(),
firstConsumerId, lastConsumerId);
+
+ writeFuture =
+ writer.submit(
+ (Callable<Void>)
+ () -> {
+ writeWhileSubscribed(database, table, keepWriting,
writesStarted);
+ return null;
+ });
+ Assert.assertTrue("Writer did not start", writesStarted.await(30,
TimeUnit.SECONDS));
+ if (writeFuture.isDone()) {
+ writeFuture.get();
+ }
+ Assert.assertFalse("Writer finished before unsubscribe",
writeFuture.isDone());
+ consumer1.poll(Duration.ofMillis(100));
+
+ unsubscribeWithin(consumer1, ids.getTopic(), unsubscribeExecutor);
+ awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId(),
lastConsumerId);
+ Assert.assertFalse("Writer finished before last unsubscribe",
writeFuture.isDone());
+ unsubscribeWithin(consumer2, ids.getTopic(), unsubscribeExecutor);
+ awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId());
+
+ // A previous queue may still be closing. The new queue must retain its
task registration.
+ consumer1.subscribe(ids.getTopic());
+ awaitConsumerIds(ids.getTopic(), ids.getConsumerGroupId(),
firstConsumerId);
+
+ keepWriting.set(false);
+ writeFuture.get(30, TimeUnit.SECONDS);
+ final Set<String> expectedRows =
+ ConsensusSubscriptionTableITSupport.insertRows(database, table,
1_000_000_000L, 20, true);
+ final ConsensusSubscriptionTableITSupport.ConsumedRecords consumed =
+ ConsensusSubscriptionTableITSupport.pollAndCommitUntilContains(
+ consumer1, expectedRows, 60);
+ Assert.assertTrue(consumed.toString(),
consumed.getRowKeys().containsAll(expectedRows));
+ } finally {
+ keepWriting.set(false);
+ if (writeFuture != null && !writeFuture.isDone()) {
+ writeFuture.cancel(true);
+ }
+ writer.shutdownNow();
+ writer.awaitTermination(10, TimeUnit.SECONDS);
+ unsubscribeExecutor.shutdownNow();
+ closeQuietly(consumer2);
+ closeQuietly(consumer1);
+ ConsensusSubscriptionTableITSupport.cleanup(null, ids.getTopic(),
database);
+ }
+ }
+
+ private static void writeWhileSubscribed(
+ final String database,
+ final String table,
+ final AtomicBoolean keepWriting,
+ final CountDownLatch writesStarted)
+ throws Exception {
+ try (final ITableSession session =
EnvFactory.getEnv().getTableSessionConnection()) {
+ session.executeNonQueryStatement("use " + database);
+ for (long row = 0; keepWriting.get(); row++) {
+ final long timestamp = 1_000L + row;
+ session.executeNonQueryStatement(
+ String.format(
+ Locale.ROOT,
+ "insert into %s(tag1, s1, time) values ('writing', %d, %d)",
+ table,
+ timestamp,
+ timestamp));
+ if (row == 20) {
+ writesStarted.countDown();
+ }
+ Thread.sleep(10L);
+ }
+ session.executeNonQueryStatement("flush");
+ } finally {
+ writesStarted.countDown();
+ }
+ }
+
+ private static void unsubscribeWithin(
+ final SubscriptionTablePullConsumer consumer,
+ final String topicName,
+ final ExecutorService executor)
+ throws Exception {
+ final Future<?> future = executor.submit(() ->
consumer.unsubscribe(topicName));
+ try {
+ future.get(MAX_UNSUBSCRIBE_MS, TimeUnit.MILLISECONDS);
+ } finally {
+ future.cancel(true);
+ }
+ }
+
+ private static void closeQuietly(final SubscriptionTablePullConsumer
consumer) {
+ if (consumer != null) {
+ try {
+ consumer.close();
+ } catch (final Exception ignored) {
+ // Best effort cleanup after a failed assertion.
+ }
+ }
+ }
+
+ private static void awaitConsumerIds(
+ final String topicName, final String consumerGroupId, final String...
expectedConsumerIds)
+ throws Exception {
+ try (final ISubscriptionTableSession session =
+ new SubscriptionTableSessionBuilder()
+ .host(EnvFactory.getEnv().getIP())
+ .port(Integer.parseInt(EnvFactory.getEnv().getPort()))
+ .build()) {
+ session.open();
+ Awaitility.await()
+ .pollInSameThread()
+ .pollInterval(Duration.ofMillis(500))
+ .atMost(Duration.ofSeconds(30))
+ .untilAsserted(
+ () -> {
+ final Set<Subscription> subscriptions =
session.getSubscriptions(topicName);
+ if (expectedConsumerIds.length == 0) {
+ Assert.assertTrue(subscriptions.toString(),
subscriptions.isEmpty());
+ } else {
+ Assert.assertEquals(subscriptions.toString(), 1,
subscriptions.size());
+ final Subscription subscription =
subscriptions.iterator().next();
+ Assert.assertEquals(consumerGroupId,
subscription.getConsumerGroupId());
+ final String consumerIds = subscription.getConsumerIds();
+ Assert.assertTrue(
+ subscription.toString(),
+ consumerIds.startsWith("[") &&
consumerIds.endsWith("]"));
+ final String idList = consumerIds.substring(1,
consumerIds.length() - 1);
+ Assert.assertEquals(
+ subscription.toString(),
+ new HashSet<>(Arrays.asList(expectedConsumerIds)),
+ new HashSet<>(Arrays.asList(idList.split(",\\s*"))));
+ }
+ });
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 3cd7ae14585..a8896d1cf41 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2188,6 +2188,8 @@ public final class DataNodePipeMessages {
+ "frontier";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_DURING_DEREGISTER_34C332E7 =
"ConsensusPrefetchingQueue {}: error during deregister";
+ public static final String
LOG_FAILED_TO_FINISH_CLOSING_CONSENSUS_PREFETCHING_QUEUE_ARG_3C31731C =
+ "Failed to finish closing consensus prefetching queue {}";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_FLUSH_LINGERING_BATCH_F97D8AA7 =
"ConsensusPrefetchingQueue {}: failed to flush lingering batch during
close, discarding it";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_ROUND_FAILED_TYPE_MESSAGE_63BC909B =
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 23f58a00c51..f9a4df76fb7 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2031,6 +2031,8 @@ public final class DataNodePipeMessages {
"ConsensusPrefetchingQueue {}：不支持 seekAfterRegionProgress（没有 WAL 目录）";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_READ_WAL_METADATA_FROM_A2ED50D1 =
"ConsensusPrefetchingQueue {}：计算 seekToEnd frontier 时，从 {} 读取 WAL
metadata 失败";
+ public static final String
LOG_FAILED_TO_FINISH_CLOSING_CONSENSUS_PREFETCHING_QUEUE_ARG_3C31731C =
+ "未能完成共识预取队列 {} 的关闭";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_DURING_DEREGISTER_34C332E7 =
"ConsensusPrefetchingQueue {}：注销期间出错";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_FLUSH_LINGERING_BATCH_F97D8AA7 =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
index fd04724488c..2792dff39e2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java
@@ -743,7 +743,10 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
}
public void unbindConsensusPrefetchingQueue(final String topicName) {
- closeAndRemoveConsensusPrefetchingQueues(topicName, true, true);
+ // Subscription setup teardown and rollback manage commit state after this
call returns. Let
+ // queue close persist the current state, while avoiding an asynchronous
close deleting state
+ // restored by a concurrent setup rollback.
+ closeAndRemoveConsensusPrefetchingQueues(topicName, true, false);
}
@Override
@@ -774,7 +777,7 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
}
for (int i = 0; i < queuesToClose.size(); i++) {
- queuesToClose.get(i).close();
+ queuesToClose.get(i).closeAsync(false);
LOGGER.info(
DataNodePipeMessages
.PIPE_LOG_SUBSCRIPTION_CLOSED_CONSENSUS_PREFETCHING_QUEUE_FOR_TOPIC_3A9DDEC5,
@@ -874,7 +877,7 @@ public class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
}
for (final ConsensusPrefetchingQueue q : queuesToClose) {
- q.close(removeProgressAfterClose);
+ q.closeAsync(removeProgressAfterClose);
}
LOGGER.info(
DataNodePipeMessages
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index b8cd6b258d3..af3bf6344a0 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
@@ -222,6 +222,10 @@ public class ConsensusPrefetchingQueue {
*/
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
+ private final Object closeLifecycleLock = new Object();
+
+ private boolean closeInProgress;
+
private volatile boolean isClosed = false;
private volatile boolean closeRequested = false;
@@ -779,7 +783,7 @@ public class ConsensusPrefetchingQueue {
&& Objects.nonNull(staleExecutor)
&& (staleExecutor != currentExecutor || staleSubtask.isClosed())
&& !staleExecutor.isShutdown()) {
- staleExecutor.deregister(staleSubtask.getTaskId());
+ staleExecutor.deregister(staleSubtask);
}
final ConsensusPrefetchSubtask newSubtask = new
ConsensusPrefetchSubtask(this);
@@ -2315,6 +2319,9 @@ public class ConsensusPrefetchingQueue {
final long commitLocalSeq,
final long retainedBytes,
final long expectedSeekGeneration) {
+ if (closeRequested || isClosed) {
+ return false;
+ }
if (seekGeneration.get() != expectedSeekGeneration) {
LOGGER.debug(
DataNodePipeMessages
@@ -3965,28 +3972,75 @@ public class ConsensusPrefetchingQueue {
}
public void close(final boolean removeProgressAfterClose) {
- final PendingSeekRequest seekRequestToFail;
- final Pair<ConsensusSubscriptionPrefetchExecutor,
ConsensusPrefetchSubtask> prefetchBinding;
+ if (!lock.writeLock().tryLock()) {
+ if (beginClose()) {
+ CLOSE_CLEANUP_EXECUTOR.execute(
+ () -> finishCloseAfterWriteLock(removeProgressAfterClose, true));
+ }
+ return;
+ }
- acquireWriteLock();
+ final PendingSeekRequest seekRequestToFail;
try {
- if (closeRequested || (isClosed && !closeCleanupPending)) {
+ if (!beginClose()) {
return;
}
- closeRequested = true;
- closeCleanupPending = true;
seekRequestToFail = pendingSeekRequest;
pendingSeekRequest = null;
} finally {
releaseWriteLock();
}
- // Stop receiving real-time in-memory writes before flushing close-time
batches. Requests that
- // remain in pendingEntries have not advanced subscription progress and
can be replayed from WAL
- // if this queue is created again.
- deregisterPendingEntriesFromConsensusServer();
+ try {
+ finishCloseRequest(seekRequestToFail, removeProgressAfterClose, false);
+ } catch (final Exception e) {
+ handleCloseFailure(e);
+ }
+ }
+
+ /** Starts teardown without waiting for an active prefetch round or
consensus queue writes. */
+ public void closeAsync(final boolean removeProgressAfterClose) {
+ if (beginClose()) {
+ CLOSE_CLEANUP_EXECUTOR.execute(
+ () -> finishCloseAfterWriteLock(removeProgressAfterClose, true));
+ }
+ }
+
+ private boolean beginClose() {
+ synchronized (closeLifecycleLock) {
+ if (closeInProgress || closeRequested || (isClosed &&
!closeCleanupPending)) {
+ return false;
+ }
+ closeInProgress = true;
+ closeRequested = true;
+ closeCleanupPending = true;
+ return true;
+ }
+ }
+
+ private void finishCloseAfterWriteLock(
+ final boolean removeProgressAfterClose, final boolean backgroundCleanup)
{
+ try {
+ final PendingSeekRequest seekRequestToFail;
+ acquireWriteLock();
+ try {
+ seekRequestToFail = pendingSeekRequest;
+ pendingSeekRequest = null;
+ } finally {
+ releaseWriteLock();
+ }
+ finishCloseRequest(seekRequestToFail, removeProgressAfterClose,
backgroundCleanup);
+ } catch (final Exception e) {
+ handleCloseFailure(e);
+ }
+ }
- prefetchBinding = detachPrefetchSubtask();
+ private void finishCloseRequest(
+ final PendingSeekRequest seekRequestToFail,
+ final boolean removeProgressAfterClose,
+ final boolean backgroundCleanup) {
+ final Pair<ConsensusSubscriptionPrefetchExecutor,
ConsensusPrefetchSubtask> prefetchBinding =
+ detachPrefetchSubtask();
if (Objects.nonNull(seekRequestToFail)) {
seekRequestToFail.fail(
@@ -3999,14 +4053,21 @@ public class ConsensusPrefetchingQueue {
if (Objects.nonNull(prefetchBinding.right)) {
prefetchBinding.right.cancelPendingExecution();
- if (prefetchBinding.right.isScheduledOrRunning()) {
- CLOSE_CLEANUP_EXECUTOR.execute(
- () -> finishCloseInBackground(prefetchBinding,
removeProgressAfterClose));
- return;
- }
}
- finishClose(prefetchBinding, removeProgressAfterClose);
+ if (!backgroundCleanup
+ && Objects.nonNull(prefetchBinding.right)
+ && prefetchBinding.right.isScheduledOrRunning()) {
+ CLOSE_CLEANUP_EXECUTOR.execute(
+ () -> finishCloseInBackground(prefetchBinding,
removeProgressAfterClose));
+ return;
+ }
+
+ if (backgroundCleanup) {
+ finishCloseInBackground(prefetchBinding, removeProgressAfterClose);
+ } else {
+ finishClose(prefetchBinding, removeProgressAfterClose);
+ }
}
private void finishCloseInBackground(
@@ -4015,24 +4076,38 @@ public class ConsensusPrefetchingQueue {
try {
finishClose(prefetchBinding, removeProgressAfterClose);
} catch (final Exception e) {
- LOGGER.warn("Failed to finish closing consensus prefetching queue {}",
this, e);
+ handleCloseFailure(e);
}
}
+ private void handleCloseFailure(final Exception e) {
+ closeRequested = false;
+ synchronized (closeLifecycleLock) {
+ closeInProgress = false;
+ }
+ LOGGER.warn(
+
DataNodePipeMessages.LOG_FAILED_TO_FINISH_CLOSING_CONSENSUS_PREFETCHING_QUEUE_ARG_3C31731C,
+ this,
+ e);
+ }
+
private void finishClose(
final Pair<ConsensusSubscriptionPrefetchExecutor,
ConsensusPrefetchSubtask> prefetchBinding,
final boolean removeProgressAfterClose) {
+ // Queue deregistration may wait for consensus-side WAL/index work.
Metadata callers use
+ // closeAsync so this cleanup cannot hold up the consumer-group metadata
push.
+ // Requests left in pendingEntries have not advanced subscription progress
and can be replayed
+ // from WAL if this queue is created again.
+ deregisterPendingEntriesFromConsensusServer();
+
if (Objects.nonNull(prefetchBinding.right)) {
prefetchBinding.right.awaitIdle();
}
try {
acquireWriteLock();
try {
- if (!isClosed
- && pendingSeekRequest == null
- && seekGeneration.get() == observedSeekGeneration) {
- flushLingeringBatchOnCloseUnderWriteLock();
- }
+ // Any uncommitted linger batch is discarded by cleanUp() and replayed
from WAL if the
+ // topic is subscribed again. Emitting it during close could recreate
removed commit state.
markClosed();
} finally {
releaseWriteLock();
@@ -4043,7 +4118,7 @@ public class ConsensusPrefetchingQueue {
if (Objects.nonNull(prefetchBinding.left) &&
Objects.nonNull(prefetchBinding.right)) {
if (!prefetchBinding.left.isShutdown()) {
- prefetchBinding.left.deregister(prefetchBinding.right.getTaskId());
+ prefetchBinding.left.deregister(prefetchBinding.right);
} else {
prefetchBinding.right.close();
}
@@ -4062,6 +4137,9 @@ public class ConsensusPrefetchingQueue {
closeCleanupPending = false;
} finally {
closeRequested = false;
+ synchronized (closeLifecycleLock) {
+ closeInProgress = false;
+ }
}
}
@@ -4076,25 +4154,6 @@ public class ConsensusPrefetchingQueue {
}
}
- private void flushLingeringBatchOnCloseUnderWriteLock() {
- if (lingerBatch.isEmpty()) {
- return;
- }
- LOGGER.info(
- DataNodePipeMessages
-
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FLUSHING_LINGERING_TABLETS_DURING_4C4AF235,
- this,
- lingerBatch.tablets.size());
- if (!flushBatch(lingerBatch, observedSeekGeneration)) {
- LOGGER.warn(
- DataNodePipeMessages
-
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_FLUSH_LINGERING_BATCH_F97D8AA7,
- this);
- discardBatch(lingerBatch);
- resetBatchWriterProgress();
- }
- }
-
private SubscriptionEvent generateErrorResponse(final String errorMessage) {
return new SubscriptionEvent(
SubscriptionPollResponseType.ERROR.getType(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/execution/ConsensusSubscriptionPrefetchExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/execution/ConsensusSubscriptionPrefetchExecutor.java
index 01700febb45..d34773e5510 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/execution/ConsensusSubscriptionPrefetchExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/execution/ConsensusSubscriptionPrefetchExecutor.java
@@ -104,6 +104,16 @@ public class ConsensusSubscriptionPrefetchExecutor {
subtask.close();
}
+ /** Removes a detached queue's task only if it still owns the registration.
*/
+ public synchronized void deregister(final ConsensusPrefetchSubtask subtask) {
+ if (!taskIdToSubtask.remove(subtask.getTaskId(), subtask)) {
+ return;
+ }
+ readyQueue.remove(subtask);
+ subtask.cancelPendingExecution();
+ subtask.close();
+ }
+
public void enqueue(final ConsensusPrefetchSubtask subtask) {
if (shutdown.get() || subtask.isClosed()) {
return;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
index e2de603eda1..703289b736b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
@@ -191,6 +191,7 @@ public class ConsensusPrefetchSubtask {
public void cancelPendingExecution() {
synchronized (monitor) {
+ closed = true;
delayedWakeToken++;
wakeupPending = false;
if (scheduledOrRunning && !running) {
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 2b3dc9f0ef5..371bb149c4c 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
@@ -109,6 +109,28 @@ public class ConsensusPrefetchingQueueTest {
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @Test
+ public void testLateOldPrefetchDeregistrationKeepsReplacementTask() {
+ final ConsensusSubscriptionPrefetchExecutor executor =
+ new ConsensusSubscriptionPrefetchExecutor();
+ try {
+ final ConsensusPrefetchSubtask oldSubtask =
mock(ConsensusPrefetchSubtask.class);
+ final ConsensusPrefetchSubtask replacementSubtask =
mock(ConsensusPrefetchSubtask.class);
+ when(oldSubtask.getTaskId()).thenReturn("same-task-id");
+ when(replacementSubtask.getTaskId()).thenReturn("same-task-id");
+
+ assertTrue(executor.register(oldSubtask));
+ executor.deregister(oldSubtask);
+ assertTrue(executor.register(replacementSubtask));
+
+ executor.deregister(oldSubtask);
+ verify(replacementSubtask, never()).close();
+ verify(replacementSubtask, never()).cancelPendingExecution();
+ } finally {
+ executor.shutdown();
+ }
+ }
+
@Test
public void testCloseDetachesBlockedPrefetchCleanup() throws Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -170,12 +192,152 @@ public class ConsensusPrefetchingQueueTest {
releaseCleanup.countDown();
verify(subtask,
timeout(TimeUnit.SECONDS.toMillis(TEST_TIMEOUT_SECONDS))).close();
+ awaitCloseFinished(queue);
assertTrue(queue.isClosed());
} finally {
releaseCleanup.countDown();
if (queue != null && !queue.isClosed()) {
queue.close();
}
+ if (queue != null) {
+ awaitCloseFinished(queue);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testCloseDoesNotWaitForActivePrefetchReadLock() throws Exception
{
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("async-prefetch-read-lock");
+ final ExecutorService closeExecutor = Executors.newSingleThreadExecutor();
+ ConsensusPrefetchingQueue queue = null;
+ ReentrantReadWriteLock queueLock = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.getDatabaseName()).thenReturn("db");
+ when(converter.isTableModel()).thenReturn(true);
+
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ 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();
+
+ final ConsensusPrefetchingQueue queueToClose = queue;
+ final Future<?> closeFuture = closeExecutor.submit((Runnable)
queueToClose::close);
+ closeFuture.get(1, TimeUnit.SECONDS);
+ assertFalse(queue.isClosed());
+ } finally {
+ if (queueLock != null && queueLock.getReadHoldCount() > 0) {
+ queueLock.readLock().unlock();
+ }
+ closeExecutor.shutdownNow();
+ closeExecutor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ if (queue != null && !queue.isClosed()) {
+ queue.close();
+ }
+ if (queue != null) {
+ awaitCloseFinished(queue);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ assertTrue(queue.isClosed());
+ }
+
+ @Test
+ public void testCloseDetachesBlockedConsensusQueueDeregistration() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("async-prefetch-deregister");
+ final CountDownLatch deregistrationStarted = new CountDownLatch(1);
+ final CountDownLatch releaseDeregistration = new CountDownLatch(1);
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ doAnswer(
+ invocation -> {
+ deregistrationStarted.countDown();
+ releaseDeregistration.await(TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ return null;
+ })
+ .when(serverImpl)
+ .deregisterSubscriptionQueue(any());
+
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.getDatabaseName()).thenReturn("db");
+ when(converter.isTableModel()).thenReturn(true);
+
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final ConsensusPrefetchSubtask subtask =
mock(ConsensusPrefetchSubtask.class);
+ final ConsensusSubscriptionPrefetchExecutor prefetchExecutor =
+ mock(ConsensusSubscriptionPrefetchExecutor.class);
+ when(subtask.isScheduledOrRunning()).thenReturn(false);
+ when(prefetchExecutor.isShutdown()).thenReturn(true);
+ setPrefetchBinding(queue, prefetchExecutor, subtask);
+
+ final long closeStartNanos = System.nanoTime();
+ queue.closeAsync(false);
+ final long closeElapsedMillis =
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - closeStartNanos);
+
+ assertTrue(closeElapsedMillis < 1000L);
+ assertTrue(deregistrationStarted.await(TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
+ assertFalse(queue.isClosed());
+
+ releaseDeregistration.countDown();
+ verify(subtask,
timeout(TimeUnit.SECONDS.toMillis(TEST_TIMEOUT_SECONDS))).close();
+ awaitCloseFinished(queue);
+ assertTrue(queue.isClosed());
+ } finally {
+ releaseDeregistration.countDown();
+ if (queue != null && !queue.isClosed()) {
+ queue.close();
+ }
+ if (queue != null) {
+ awaitCloseFinished(queue);
+ }
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
}
}
@@ -2916,6 +3078,27 @@ public class ConsensusPrefetchingQueueTest {
subtaskField.set(queue, subtask);
}
+ private static void awaitCloseFinished(final ConsensusPrefetchingQueue
queue) throws Exception {
+ final Field closeLifecycleLockField =
+ ConsensusPrefetchingQueue.class.getDeclaredField("closeLifecycleLock");
+ closeLifecycleLockField.setAccessible(true);
+ final Object closeLifecycleLock = closeLifecycleLockField.get(queue);
+
+ final Field closeInProgressField =
+ ConsensusPrefetchingQueue.class.getDeclaredField("closeInProgress");
+ closeInProgressField.setAccessible(true);
+ final long closeDeadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(TEST_TIMEOUT_SECONDS);
+ while (System.nanoTime() < closeDeadline) {
+ synchronized (closeLifecycleLock) {
+ if (!closeInProgressField.getBoolean(queue)) {
+ return;
+ }
+ }
+ Thread.sleep(10L);
+ }
+ throw new AssertionError("Timed out waiting for queue close cleanup");
+ }
+
private static ProgressWALIterator subscriptionWalIterator(final
ConsensusPrefetchingQueue queue)
throws Exception {
final Field field =
ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator");
