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 2b6d628e022 [Pipe] Fix deletion buffer close flush race (#18662)
2b6d628e022 is described below
commit 2b6d628e02230e33eb8a38c0075ac74b36dbf305
Author: Caideyipi <[email protected]>
AuthorDate: Thu Sep 17 12:29:20 2026 +0800
[Pipe] Fix deletion buffer close flush race (#18662)
---
.../deletion/persist/PageCacheDeletionBuffer.java | 67 +++++++++++-----
.../persist/PageCacheDeletionBufferTest.java | 88 ++++++++++++++++++++++
2 files changed, 137 insertions(+), 18 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
index d58c65158ac..f306ee15eee 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
@@ -42,7 +42,6 @@ import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.StandardOpenOption;
import java.util.List;
-import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
@@ -82,8 +81,11 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
// single thread to serialize WALEntry to workingBuffer
private final ExecutorService persistThread;
private final Lock buffersLock = new ReentrantLock();
+ private final Lock lifecycleLock = new ReentrantLock();
// Total size of this batch.
private final AtomicInteger totalSize = new AtomicInteger(0);
+ // Number of accepted deletions that have not completed their persist tasks.
+ private final AtomicInteger unflushedDeletionCount = new AtomicInteger(0);
// All deletions that will be handled in a single persist task
private final List<DeletionResource> pendingDeletionsInOneTask = new
CopyOnWriteArrayList<>();
@@ -145,13 +147,7 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
@Override
public boolean isAllDeletionFlushed() {
- buffersLock.lock();
- try {
- int pos =
Optional.ofNullable(serializeBuffer).map(ByteBuffer::position).orElse(0);
- return deletionResources.isEmpty() && pos == 0;
- } finally {
- buffersLock.unlock();
- }
+ return unflushedDeletionCount.get() == 0;
}
private void allocateBuffers() {
@@ -166,13 +162,24 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
}
public void registerDeletionResource(DeletionResource deletionResource) {
- if (isClosed) {
- LOGGER.error(
-
DataNodePipeMessages.FAIL_TO_REGISTER_DELETIONRESOURCE_INTO_DELETIONBUFFER_BECAUSE,
- dataRegionId);
- return;
+ lifecycleLock.lock();
+ try {
+ if (isClosed) {
+ LOGGER.error(
+
DataNodePipeMessages.FAIL_TO_REGISTER_DELETIONRESOURCE_INTO_DELETIONBUFFER_BECAUSE,
+ dataRegionId);
+ return;
+ }
+ unflushedDeletionCount.incrementAndGet();
+ try {
+ deletionResources.add(deletionResource);
+ } catch (RuntimeException e) {
+ unflushedDeletionCount.decrementAndGet();
+ throw e;
+ }
+ } finally {
+ lifecycleLock.unlock();
}
- deletionResources.add(deletionResource);
}
private void appendCurrentBatch() throws IOException {
@@ -264,12 +271,22 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
@Override
public void close() {
- isClosed = true;
+ lifecycleLock.lock();
+ try {
+ isClosed = true;
+ } finally {
+ lifecycleLock.unlock();
+ }
// Force sync existing data in memory to disk.
// first waiting serialize and sync tasks finished, then release all
resources
waitUntilFlushAllDeletionsOrTimeOut();
if (persistThread != null) {
- persistThread.shutdownNow();
+ lifecycleLock.lock();
+ try {
+ persistThread.shutdownNow();
+ } finally {
+ lifecycleLock.unlock();
+ }
try {
if (!persistThread.awaitTermination(30, TimeUnit.SECONDS)) {
LOGGER.warn(DataNodePipeMessages.PERSISTTHREAD_DID_NOT_TERMINATE_WITHIN_S, 30);
@@ -306,19 +323,31 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
private class PersistTask implements Runnable {
// Batch size in current task, used to roll back.
private final AtomicInteger currentTaskBatchSize = new AtomicInteger(0);
+ private int currentTaskDeletionCount = 0;
@Override
public void run() {
+ boolean taskFinished = false;
try {
persistDeletion();
+ taskFinished = true;
} catch (IOException e) {
LOGGER.warn(DataNodePipeMessages.DELETION_PERSIST_CANNOT_WRITE_TO_MAY_CAUSE,
logFile, e);
// if any exception occurred, this batch will not be written to disk
and lost.
pendingDeletionsInOneTask.forEach(deletionResource ->
deletionResource.onPersistFailed(e));
rollbackFileAttribute(currentTaskBatchSize.get());
+ taskFinished = true;
} finally {
- if (!isClosed) {
- persistThread.submit(new PersistTask());
+ if (taskFinished) {
+ unflushedDeletionCount.addAndGet(-currentTaskDeletionCount);
+ }
+ lifecycleLock.lock();
+ try {
+ if ((!isClosed || unflushedDeletionCount.get() > 0) &&
!persistThread.isShutdown()) {
+ persistThread.submit(new PersistTask());
+ }
+ } finally {
+ lifecycleLock.unlock();
}
}
}
@@ -347,6 +376,7 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
// size of serializeBuffer.
serializeDeletionToBatchBuffer(firstDeletionResource);
pendingDeletionsInOneTask.add(firstDeletionResource);
+ currentTaskDeletionCount++;
maxProgressIndexInCurrentFile =
maxProgressIndexInCurrentFile.updateToMinimumEqualOrIsAfterProgressIndex(
firstDeletionResource.getProgressIndex());
@@ -389,6 +419,7 @@ public class PageCacheDeletionBuffer implements
DeletionBuffer {
return;
}
pendingDeletionsInOneTask.add(deletionResource);
+ currentTaskDeletionCount++;
// Update max progressIndex in current file if serialized successfully.
maxProgressIndexInCurrentFile =
maxProgressIndexInCurrentFile.updateToMinimumEqualOrIsAfterProgressIndex(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java
new file mode 100644
index 00000000000..b63dff6a5b6
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.db.pipe.consensus.deletion.persist;
+
+import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
+import org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
+
+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;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class PageCacheDeletionBufferTest {
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testDequeuedDeletionIsNotReportedAsFlushed() throws Exception {
+ final CountDownLatch serializationStarted = new CountDownLatch(1);
+ final CountDownLatch continueSerialization = new CountDownLatch(1);
+ final int dataRegionId = 1;
+ final DeleteDataNode deleteDataNode =
+ new DeleteDataNode(
+ new PlanNodeId("1"),
+ Collections.singletonList(new
MeasurementPath("root.vehicle.d2.s0")),
+ 50,
+ 150);
+ deleteDataNode.setProgressIndex(
+ new RecoverProgressIndex(
+ IoTDBDescriptor.getInstance().getConfig().getDataNodeId(),
+ new SimpleProgressIndex(0, 1)));
+ final DeletionResource deletionResource =
+ new DeletionResource(deleteDataNode, ignored -> {}, dataRegionId) {
+ @Override
+ public ByteBuffer serialize() {
+ serializationStarted.countDown();
+ try {
+ continueSerialization.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return super.serialize();
+ }
+ };
+ final PageCacheDeletionBuffer deletionBuffer =
+ new PageCacheDeletionBuffer(
+ dataRegionId,
temporaryFolder.newFolder("deletions").getAbsolutePath());
+
+ deletionBuffer.start();
+ try {
+ deletionBuffer.registerDeletionResource(deletionResource);
+ Assert.assertTrue(serializationStarted.await(10, TimeUnit.SECONDS));
+
+ Assert.assertFalse(deletionBuffer.isAllDeletionFlushed());
+ } finally {
+ continueSerialization.countDown();
+ deletionBuffer.close();
+ }
+ Assert.assertSame(DeletionResource.Status.SUCCESS,
deletionResource.waitForResult());
+ }
+}
