You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/14 00:49:14 UTC
(pulsar) branch master updated: [fix][txn] Prevent queued snapshot writes from stalling (#26565)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ce7c74f7a393 [fix][txn] Prevent queued snapshot writes from stalling (#26565)
ce7c74f7a393 is described below
commit ce7c74f7a39372159805f9f1772215d6ed178dd4
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Sep 14 03:49:07 2026 +0300
[fix][txn] Prevent queued snapshot writes from stalling (#26565)
---
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 13 ++-
...apshotSegmentAbortedTxnProcessorWorkerTest.java | 124 +++++++++++++++++++++
2 files changed, 134 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
index 480cd5f71575..c59aecc68705 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
@@ -614,6 +614,11 @@ public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbor
if (STATE_UPDATER.compareAndSet(this, OperationState.None, OperationState.Operating)) {
//Double-check. Avoid NoSuchElementException due to the first task is completed by other thread.
if (taskQueue.isEmpty()) {
+ STATE_UPDATER.compareAndSet(this, OperationState.Operating, OperationState.None);
+ // An append may have observed Operating between the empty check and releasing ownership.
+ if (!taskQueue.isEmpty()) {
+ executeTask();
+ }
return;
}
Pair<OperationType, Pair<CompletableFuture<Void>, Supplier<CompletableFuture<Void>>>> firstTask =
@@ -630,12 +635,14 @@ public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbor
} else {
firstTask.getRight().getKey().complete(null);
taskQueue.removeFirst();
- //Execute the next task in the other thread.
- topic.getBrokerService().getPulsar().getTransactionExecutorProvider()
- .getExecutor(this).submit(this::executeTask);
}
STATE_UPDATER.compareAndSet(this, OperationState.Operating,
OperationState.None);
+ if (throwable == null) {
+ // Release the operation before scheduling: the executor may run the next task immediately.
+ topic.getBrokerService().getPulsar().getTransactionExecutorProvider()
+ .getExecutor(this).submit(this::executeTask);
+ }
});
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorWorkerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorWorkerTest.java
new file mode 100644
index 000000000000..08e43176c697
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorWorkerTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.pulsar.broker.transaction.buffer.impl;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.impl.SnapshotSegmentAbortedTxnProcessorImpl.PersistentWorker;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class SnapshotSegmentAbortedTxnProcessorWorkerTest {
+
+ private ExecutorService executor;
+ private SnapshotSegmentAbortedTxnProcessorImpl processor;
+ private PersistentWorker worker;
+
+ @BeforeMethod
+ public void setup() {
+ PersistentTopic topic = mock(PersistentTopic.class, RETURNS_DEEP_STUBS);
+ when(topic.getName()).thenReturn("persistent://public/default/snapshot-worker");
+ executor = MoreExecutors.newDirectExecutorService();
+ when(topic.getBrokerService().getPulsar().getTransactionExecutorProvider().getExecutor(any(Object.class)))
+ .thenReturn(executor);
+ processor = new SnapshotSegmentAbortedTxnProcessorImpl(topic);
+ worker = processor.new PersistentWorker(topic);
+ }
+
+ @Test(timeOut = 10_000)
+ public void testQueuedTasksContinueWhenExecutorRunsImmediately() throws Exception {
+ CompletableFuture<Void> firstWrite = new CompletableFuture<>();
+ CompletableFuture<Void> secondWrite = new CompletableFuture<>();
+ AtomicBoolean secondStarted = new AtomicBoolean();
+ AtomicBoolean thirdStarted = new AtomicBoolean();
+ CompletableFuture<Void> firstResult = worker.appendTask(PersistentWorker.OperationType.WriteSegment,
+ () -> firstWrite);
+ CompletableFuture<Void> secondResult = worker.appendTask(PersistentWorker.OperationType.WriteSegment, () -> {
+ secondStarted.set(true);
+ return secondWrite;
+ });
+ assertFalse(secondStarted.get());
+
+ firstWrite.complete(null);
+ firstResult.get(5, TimeUnit.SECONDS);
+ assertTrue(secondStarted.get(), "The queued write must start when the previous write completes");
+
+ CompletableFuture<Void> thirdResult = worker.appendTask(PersistentWorker.OperationType.WriteSegment, () -> {
+ thirdStarted.set(true);
+ return CompletableFuture.completedFuture(null);
+ });
+ assertFalse(thirdStarted.get(), "The next write must wait for the in-flight write");
+ secondWrite.complete(null);
+ secondResult.get(5, TimeUnit.SECONDS);
+ thirdResult.get(5, TimeUnit.SECONDS);
+ assertTrue(thirdStarted.get());
+ }
+
+ @Test(timeOut = 10_000)
+ public void testQueueDrainedBeforeAcquiringOperation() throws Exception {
+ AtomicInteger stage = new AtomicInteger();
+ AtomicReference<CompletableFuture<Void>> appendedWhileOperating = new AtomicReference<>();
+ worker.taskQueue = new ConcurrentLinkedDeque<>() {
+ @Override
+ public boolean isEmpty() {
+ boolean empty = super.isEmpty();
+ if (!empty && stage.compareAndSet(0, 1)) {
+ // Another caller drains the queue after the initial snapshot, before ownership is acquired.
+ worker.appendTask(PersistentWorker.OperationType.WriteSegment,
+ () -> CompletableFuture.completedFuture(null));
+ stage.set(2);
+ } else if (empty && stage.compareAndSet(2, 3)) {
+ // An append observes Operating after the second empty snapshot, before ownership is released.
+ appendedWhileOperating.set(worker.appendTask(PersistentWorker.OperationType.WriteSegment,
+ () -> CompletableFuture.completedFuture(null)));
+ }
+ return empty;
+ }
+ };
+ worker.appendTask(PersistentWorker.OperationType.WriteSegment,
+ () -> CompletableFuture.completedFuture(null)).get(5, TimeUnit.SECONDS);
+ assertEquals(stage.get(), 3, "Both queue observations must be exercised");
+ appendedWhileOperating.get().get(5, TimeUnit.SECONDS);
+ worker.appendTask(PersistentWorker.OperationType.WriteSegment,
+ () -> CompletableFuture.completedFuture(null)).get(5, TimeUnit.SECONDS);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() throws Exception {
+ worker.closeAsync().get(5, TimeUnit.SECONDS);
+ processor.closeAsync().get(5, TimeUnit.SECONDS);
+ executor.shutdownNow();
+ }
+}