You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/09 16:40:31 UTC
(pulsar) 03/04: [fix][broker] Cancel queued transaction snapshot recovery on topic close (#26335)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 001c89c2cb1ad2914c9332d0755ba1aa42ab358e
Author: Kai Wang <kw...@apache.org>
AuthorDate: Wed Sep 9 15:34:26 2026 +0800
[fix][broker] Cancel queued transaction snapshot recovery on topic close (#26335)
(cherry picked from commit d2bc2076a28a191c930f61a14fa88621b4e98400)
---
.../impl/AbstractSnapshotAbortedTxnProcessor.java | 186 +++++++++++++
.../SingleSnapshotAbortedTxnProcessorImpl.java | 46 +---
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 105 +++----
.../buffer/impl/TopicTransactionBuffer.java | 212 +++++++++++----
.../pulsar/broker/transaction/TransactionTest.java | 7 +-
.../AbstractSnapshotAbortedTxnProcessorTest.java | 302 +++++++++++++++++++++
...napshotSegmentAbortedTxnProcessorCloseTest.java | 173 ++++++++++++
.../impl/TopicTransactionBufferCloseTest.java | 277 +++++++++++++++++++
.../impl/TopicTransactionBufferRecoveryTest.java | 88 ++++++
9 files changed, 1258 insertions(+), 138 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java
new file mode 100644
index 000000000000..2c8e9acd4a1e
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java
@@ -0,0 +1,186 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Coordinates snapshot recovery and resource closure so resources remain available until processor-owned recovery
+ * work finishes.
+ */
+abstract class AbstractSnapshotAbortedTxnProcessor implements AbortedTxnProcessor {
+
+ private enum State {
+ IDLE,
+ RECOVERY_QUEUED,
+ RECOVERY_RUNNING,
+ CLOSED
+ }
+
+ private final ScheduledExecutorService recoveryExecutor;
+
+ private volatile State state = State.IDLE;
+ private Future<?> recoveryTask;
+ private CompletableFuture<Position> recoveryFuture = CompletableFuture.completedFuture(null);
+ // Completes before the recovery result, so close waits only for processor-owned work.
+ private CompletableFuture<Void> recoveryWorkFinishedFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+ AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) {
+ this.recoveryExecutor = recoveryExecutor;
+ }
+
+ @Override
+ public final CompletableFuture<Position> recoverFromSnapshot() {
+ CompletableFuture<Position> newRecoveryFuture;
+ CompletableFuture<Void> newRecoveryWorkFinishedFuture;
+ RejectedExecutionException submissionFailure = null;
+ synchronized (this) {
+ if (this.state == State.CLOSED) {
+ return CompletableFuture.failedFuture(closedException());
+ }
+ if (!recoveryFuture.isDone()) {
+ return recoveryFuture.copy();
+ }
+ // Bind the task to this recovery attempt so a later retry cannot complete the wrong future.
+ newRecoveryFuture = new CompletableFuture<>();
+ newRecoveryWorkFinishedFuture = new CompletableFuture<>();
+ this.recoveryFuture = newRecoveryFuture;
+ this.recoveryWorkFinishedFuture = newRecoveryWorkFinishedFuture;
+ this.state = State.RECOVERY_QUEUED;
+ try {
+ this.recoveryTask = recoveryExecutor.submit(
+ () -> runRecovery(newRecoveryFuture, newRecoveryWorkFinishedFuture));
+ } catch (RejectedExecutionException e) {
+ this.state = State.IDLE;
+ submissionFailure = e;
+ }
+ }
+ if (submissionFailure != null) {
+ newRecoveryWorkFinishedFuture.complete(null);
+ newRecoveryFuture.completeExceptionally(submissionFailure);
+ }
+ // Do not let callers complete the internal recovery result.
+ return newRecoveryFuture.copy();
+ }
+
+ private void runRecovery(CompletableFuture<Position> recoveryResult,
+ CompletableFuture<Void> recoveryWorkFinished) {
+ Position recoveredPosition = null;
+ Throwable recoveryFailure = null;
+ boolean recoveryFinished = false;
+ if (tryStartRecovery()) {
+ try {
+ recoveredPosition = doRecoverFromSnapshot(recoveryExecutor);
+ } catch (Throwable throwable) {
+ recoveryFailure = throwable;
+ }
+ recoveryFinished = tryMarkRecoveryFinished();
+ }
+ recoveryWorkFinished.complete(null);
+ if (!recoveryFinished) {
+ failRecoveryAfterClose(recoveryResult);
+ } else if (recoveryFailure == null) {
+ recoveryResult.complete(recoveredPosition);
+ } else {
+ recoveryResult.completeExceptionally(recoveryFailure);
+ }
+ }
+
+ /**
+ * Recovers the processor state synchronously on the recovery executor thread.
+ * If closure wins while this method is running, its result is discarded and recovery fails as closed.
+ *
+ * @param executor executor used by recovery-related operations
+ * @return the recovery position, or {@code null} when no snapshot exists
+ */
+ abstract Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception;
+
+ final boolean isClosed() {
+ return this.state == State.CLOSED;
+ }
+
+ private synchronized boolean tryStartRecovery() {
+ if (this.state != State.RECOVERY_QUEUED) {
+ return false;
+ }
+ this.state = State.RECOVERY_RUNNING;
+ return true;
+ }
+
+ private synchronized boolean tryMarkRecoveryFinished() {
+ if (this.state != State.RECOVERY_RUNNING) {
+ return false;
+ }
+ this.state = State.IDLE;
+ return true;
+ }
+
+ @Override
+ public final CompletableFuture<Void> closeAsync() {
+ State previousState;
+ CompletableFuture<Position> currentRecoveryFuture;
+ CompletableFuture<Void> currentRecoveryWorkFinishedFuture;
+ synchronized (this) {
+ if (this.state == State.CLOSED) {
+ return closeFuture;
+ }
+ previousState = this.state;
+ this.state = State.CLOSED;
+ currentRecoveryFuture = this.recoveryFuture;
+ currentRecoveryWorkFinishedFuture = this.recoveryWorkFinishedFuture;
+ if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) {
+ this.recoveryTask.cancel(false);
+ }
+ }
+ FutureUtil.completeAfter(closeFuture,
+ currentRecoveryWorkFinishedFuture.thenCompose(v -> closeResources()));
+ if (previousState == State.RECOVERY_QUEUED) {
+ failRecoveryAfterClose(currentRecoveryFuture);
+ }
+ if (previousState != State.RECOVERY_RUNNING) {
+ currentRecoveryWorkFinishedFuture.complete(null);
+ }
+ return closeFuture;
+ }
+
+ private void failRecoveryAfterClose(CompletableFuture<Position> recoveryResult) {
+ closeFuture.whenComplete((__, ___) -> recoveryResult.completeExceptionally(closedException()));
+ }
+
+ /**
+ * Closes resources after processor-owned recovery work has finished.
+ *
+ * <p>This method may be invoked on the caller of {@link #closeAsync()} or on the recovery executor thread.
+ * Implementations must not rely on thread affinity and should return without blocking.
+ */
+ abstract CompletableFuture<Void> closeResources();
+
+ private static BrokerServiceException closedException() {
+ return new BrokerServiceException.ServiceUnitNotReadyException(
+ "Transaction buffer snapshot processor is closed");
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
index 5d360a5822df..fb84be7f0bae 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
@@ -30,7 +30,6 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
-import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.metadata.AbortTxnMetadata;
import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.transaction.TxnID;
@@ -40,7 +39,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
@Slf4j
-public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcessor {
+public class SingleSnapshotAbortedTxnProcessorImpl extends AbstractSnapshotAbortedTxnProcessor {
private final PersistentTopic topic;
private final ReferenceCountedWriter<TransactionBufferSnapshot> takeSnapshotWriter;
/**
@@ -51,11 +50,11 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
private volatile long lastSnapshotTimestamps;
- private volatile boolean isClosed = false;
-
public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) {
+ super(topic.getBrokerService().getPulsar().getTransactionSnapshotRecoverExecutorProvider()
+ .chooseThread(TopicName.get(topic.getName()).getNamespace()));
this.topic = topic;
- this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar()
+ this.takeSnapshotWriter = topic.getBrokerService().getPulsar()
.getTransactionBufferSnapshotServiceFactory()
.getTxnBufferSnapshotService().getReferenceWriter(TopicName.get(topic.getName()).getNamespaceObject());
this.takeSnapshotWriter.getFuture().exceptionally((ex) -> {
@@ -89,29 +88,15 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
}
@Override
- public CompletableFuture<Position> recoverFromSnapshot() {
- final var future = new CompletableFuture<Position>();
+ Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception {
final var pulsar = topic.getBrokerService().getPulsar();
- final String namespace = TopicName.get(topic.getName()).getNamespace();
- final ScheduledExecutorService scheduledExecutor = pulsar.getTransactionSnapshotRecoverExecutorProvider()
- .chooseThread(namespace);
- scheduledExecutor.execute(() -> {
- try {
- final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
- .getTableView(scheduledExecutor).readLatest(topic.getName());
- if (snapshot != null) {
- handleSnapshot(snapshot);
- final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),
- snapshot.getMaxReadPositionEntryId());
- future.complete(startReadCursorPosition);
- } else {
- future.complete(null);
- }
- } catch (Throwable e) {
- future.completeExceptionally(e);
- }
- });
- return future;
+ final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
+ .getTableView(executor).readLatest(topic.getName());
+ if (isClosed() || snapshot == null) {
+ return null;
+ }
+ handleSnapshot(snapshot);
+ return PositionFactory.create(snapshot.getMaxReadPositionLedgerId(), snapshot.getMaxReadPositionEntryId());
}
@Override
@@ -171,11 +156,8 @@ public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcesso
}
@Override
- public synchronized CompletableFuture<Void> closeAsync() {
- if (!isClosed) {
- isClosed = true;
- takeSnapshotWriter.release();
- }
+ CompletableFuture<Void> closeResources() {
+ takeSnapshotWriter.release();
return CompletableFuture.completedFuture(null);
}
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 38ae39583e57..eb31bed1a03f 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
@@ -49,7 +49,6 @@ import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.Refe
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
-import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot;
import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndex;
import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes;
@@ -72,7 +71,7 @@ import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.FutureUtil;
@Slf4j
-public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcessor {
+public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbortedTxnProcessor {
/**
* Stored the unsealed aborted transaction IDs Whose size is always less than the snapshotSegmentCapacity.
@@ -138,10 +137,14 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
* <p> Clear all snapshot segment. </p>
*/
private final PersistentWorker persistentWorker;
+ // A failed recovery can be retried, so close must retain updates started by every attempt.
+ private CompletableFuture<Void> recoveryIndexUpdatesFuture = CompletableFuture.completedFuture(null);
private static final String SNAPSHOT_PREFIX = "multiple-";
public SnapshotSegmentAbortedTxnProcessorImpl(PersistentTopic topic) {
+ super(topic.getBrokerService().getPulsar().getTransactionSnapshotRecoverExecutorProvider()
+ .chooseThread(TopicName.get(topic.getName()).getNamespace()));
this.topic = topic;
this.persistentWorker = new PersistentWorker(topic);
/*
@@ -229,48 +232,48 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
@Override
- public CompletableFuture<Position> recoverFromSnapshot() {
+ Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception {
final var pulsar = topic.getBrokerService().getPulsar();
- final var future = new CompletableFuture<Position>();
- final String namespace = TopicName.get(topic.getName()).getNamespace();
- final ScheduledExecutorService scheduledExecutor = pulsar.getTransactionSnapshotRecoverExecutorProvider()
- .chooseThread(namespace);
- scheduledExecutor.execute(() -> {
- try {
- final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory()
- .getTxnBufferSnapshotIndexService().getTableView(scheduledExecutor)
- .readLatest(topic.getName());
- if (indexes == null) {
- // Try recovering from the old format snapshot
- future.complete(recoverOldSnapshot());
- return;
- }
- final var snapshot = indexes.getSnapshot();
- final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),
- snapshot.getMaxReadPositionEntryId());
- this.unsealedTxnIds = convertTypeToTxnID(snapshot.getAborts());
- // Read snapshot segment to recover aborts
- final var snapshotSegmentTopicName = TopicName.get(TopicDomain.persistent.toString(),
- TopicName.get(topic.getName()).getNamespaceObject(),
- SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS);
- readSegmentEntries(snapshotSegmentTopicName, indexes);
- if (!this.indexes.isEmpty()) {
- // If there is no segment index, the persistent worker will write segment begin from 0.
- persistentWorker.sequenceID.set(this.indexes.get(this.indexes.lastKey()).sequenceID + 1);
- }
- unsealedTxnIds.forEach(txnID -> aborts.put(txnID, txnID));
- future.complete(startReadCursorPosition);
- } catch (Throwable throwable) {
- future.completeExceptionally(throwable);
- }
- });
- return future;
+ final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory()
+ .getTxnBufferSnapshotIndexService().getTableView(executor)
+ .readLatest(topic.getName());
+ if (isClosed()) {
+ return null;
+ }
+ if (indexes == null) {
+ // Try recovering from the old format snapshot
+ return recoverOldSnapshot(executor);
+ }
+ final var snapshot = indexes.getSnapshot();
+ final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),
+ snapshot.getMaxReadPositionEntryId());
+ this.unsealedTxnIds = convertTypeToTxnID(snapshot.getAborts());
+ // Read snapshot segment to recover aborts
+ final var snapshotSegmentTopicName = TopicName.get(TopicDomain.persistent.toString(),
+ TopicName.get(topic.getName()).getNamespaceObject(),
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS);
+ readSegmentEntries(snapshotSegmentTopicName, indexes);
+ if (isClosed()) {
+ return null;
+ }
+ if (!this.indexes.isEmpty()) {
+ // If there is no segment index, the persistent worker will write segment begin from 0.
+ persistentWorker.sequenceID.set(this.indexes.get(this.indexes.lastKey()).sequenceID + 1);
+ }
+ unsealedTxnIds.forEach(txnID -> aborts.put(txnID, txnID));
+ return startReadCursorPosition;
}
private void readSegmentEntries(TopicName topicName, TransactionBufferSnapshotIndexes indexes) throws Exception {
+ if (isClosed()) {
+ return;
+ }
final var managedLedger = openReadOnlyManagedLedger(topicName);
boolean hasInvalidIndex = false;
for (var index : indexes.getIndexList()) {
+ if (isClosed()) {
+ return;
+ }
final var position = PositionFactory.create(index.getSegmentLedgerID(), index.getSegmentEntryID());
final var abortedPosition = PositionFactory.create(index.abortedMarkLedgerID, index.abortedMarkEntryID);
try {
@@ -293,10 +296,11 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
}
}
- if (hasInvalidIndex) {
+ if (hasInvalidIndex && !isClosed()) {
// Update the snapshot segment index if there exist invalid indexes.
- persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex,
- () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot()));
+ recoveryIndexUpdatesFuture = CompletableFuture.allOf(recoveryIndexUpdatesFuture,
+ persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex,
+ () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot())));
}
}
@@ -346,22 +350,24 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
// This method will be deprecated and removed in version 4.x.0
- private Position recoverOldSnapshot() throws Exception {
- final String namespace = TopicName.get(topic.getName()).getNamespace();
- final ScheduledExecutorService scheduledExecutor = topic.getBrokerService().getPulsar()
- .getTransactionSnapshotRecoverExecutorProvider()
- .chooseThread(namespace);
+ private Position recoverOldSnapshot(ScheduledExecutorService executor) throws Exception {
+ if (isClosed()) {
+ return null;
+ }
final var pulsar = topic.getBrokerService().getPulsar();
final var topicName = TopicName.get(topic.getName());
final var topics = wait(pulsar.getPulsarResources().getTopicResources().listPersistentTopicsAsync(
NamespaceName.get(topicName.getNamespace())), "list persistent topics");
+ if (isClosed()) {
+ return null;
+ }
if (!topics.contains(TopicDomain.persistent + "://" + topicName.getNamespace() + "/"
+ SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT)) {
return null;
}
final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
- .getTableView(scheduledExecutor).readLatest(topic.getName());
- if (snapshot == null) {
+ .getTableView(executor).readLatest(topic.getName());
+ if (isClosed() || snapshot == null) {
return null;
}
handleOldSnapshot(snapshot);
@@ -409,8 +415,9 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
}
@Override
- public CompletableFuture<Void> closeAsync() {
- return persistentWorker.closeAsync();
+ CompletableFuture<Void> closeResources() {
+ return recoveryIndexUpdatesFuture.handle((__, throwable) -> null)
+ .thenCompose(__ -> persistentWorker.closeAsync());
}
private void handleSnapshotSegmentEntry(Entry entry) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index 385396f3727c..fb8314a420af 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -26,12 +26,13 @@ import io.netty.util.TimerTask;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
-import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
@@ -52,6 +53,7 @@ import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSna
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
import org.apache.pulsar.common.protocol.Commands;
@@ -90,7 +92,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private final int takeSnapshotIntervalTime;
-
+ // Complete outside the buffer monitor: inline continuations may acquire the topic monitor.
private final CompletableFuture<Void> transactionBufferFuture = new CompletableFuture<>();
/**
@@ -106,6 +108,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private final AbortedTxnProcessor.SnapshotType snapshotType;
private final MaxReadPositionCallBack maxReadPositionCallBack;
+ private Future<?> recoveryReplayTask;
/** if the first snapshot is in progress, it will pending following publishing tasks. **/
private final LinkedList<PendingAppendingTxnBufferTask> pendingAppendingTxnBufferTasks = new LinkedList<>();
@@ -148,37 +151,47 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
.execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() {
@Override
public void recoverComplete() {
+ BrokerServiceException recoveryFailure = null;
synchronized (TopicTransactionBuffer.this) {
+ if (checkIfClosed()) {
+ return;
+ }
if (ongoingTxns.isEmpty()) {
updateMaxReadPositionAfterRecovery();
}
if (!changeToReadyState()) {
log.error("[{}]Transaction buffer recover fail, current state: {}",
topic.getName(), getState());
- getTransactionBufferFuture().completeExceptionally
- (new BrokerServiceException.ServiceUnitNotReadyException(
- "Transaction buffer recover failed to change the status to Ready,"
- + "current state is: " + getState()));
+ recoveryFailure = new BrokerServiceException.ServiceUnitNotReadyException(
+ "Transaction buffer recover failed to change the status to Ready,"
+ + "current state is: " + getState());
} else {
timer.newTimeout(TopicTransactionBuffer.this,
takeSnapshotIntervalTime, TimeUnit.MILLISECONDS);
- getTransactionBufferFuture().complete(null);
recoverTime.setRecoverEndTime(System.currentTimeMillis());
}
}
+ if (recoveryFailure != null) {
+ getTransactionBufferFuture().completeExceptionally(recoveryFailure);
+ } else {
+ getTransactionBufferFuture().complete(null);
+ }
}
@Override
public void noNeedToRecover() {
synchronized (TopicTransactionBuffer.this) {
+ if (checkIfClosed()) {
+ return;
+ }
updateMaxReadPositionAfterRecovery();
if (!changeToNoSnapshotState()) {
log.error("[{}]Transaction buffer recover fail", topic.getName());
- } else {
- getTransactionBufferFuture().complete(null);
- recoverTime.setRecoverEndTime(System.currentTimeMillis());
+ return;
}
+ recoverTime.setRecoverEndTime(System.currentTimeMillis());
}
+ getTransactionBufferFuture().complete(null);
}
@Override
public void handleTxnEntry(Entry entry) {
@@ -192,8 +205,12 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
if (msgMetadata != null && msgMetadata.hasTxnidMostBits() && msgMetadata.hasTxnidLeastBits()) {
TxnID txnID = new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits());
Position position = PositionFactory.create(entry.getLedgerId(), entry.getEntryId());
+ boolean isTxnMarker = Markers.isTxnMarker(msgMetadata);
synchronized (TopicTransactionBuffer.this) {
- if (Markers.isTxnMarker(msgMetadata)) {
+ if (checkIfClosed()) {
+ return;
+ }
+ if (isTxnMarker) {
if (Markers.isTxnAbortMarker(msgMetadata)) {
snapshotAbortedTxnProcessor.putAbortedTxnAndPosition(txnID, position);
}
@@ -202,14 +219,14 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
handleTransactionMessage(txnID, position);
}
}
+ if (isTxnMarker) {
+ updateLastDispatchablePosition(null);
+ }
}
}
@Override
public void recoverExceptionally(Throwable e) {
-
- log.warn("Closing topic {} due to read transaction buffer snapshot while recovering the "
- + "transaction buffer throw exception", topic.getName(), e);
// when create reader or writer fail throw PulsarClientException,
// should close this topic and then reinit this topic
if (e instanceof PulsarClientException) {
@@ -222,11 +239,25 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
getTransactionBufferFuture().completeExceptionally(e);
}
recoverTime.setRecoverEndTime(System.currentTimeMillis());
+ if (checkIfClosed() || topic.isClosingOrDeleting()) {
+ return;
+ }
+ log.warn("Closing topic {} due to read transaction buffer snapshot while recovering the "
+ + "transaction buffer throw exception", topic.getName(), e);
topic.close(true);
}
}, this.topic, this, snapshotAbortedTxnProcessor));
}
+ private synchronized void submitRecoveryReplay(Runnable replay) {
+ if (checkIfClosed()) {
+ return;
+ }
+ // Replay polls for ledger reads, so it must stay off the live transaction executor.
+ recoveryReplayTask = topic.getBrokerService().getPulsar().getTransactionSnapshotRecoverExecutorProvider()
+ .chooseThread(TopicName.get(topic.getName()).getNamespace()).submit(replay);
+ }
+
@Override
public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) {
return CompletableFuture.completedFuture(null);
@@ -309,6 +340,11 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
PendingAppendingTxnBufferTask pendingTask = null;
try {
synchronized (pendingAppendingTxnBufferTasks) {
+ if (checkIfClosed()) {
+ failPendingTasks.accept(
+ new BrokerServiceException.ServiceUnitNotReadyException("Topic is closed"));
+ return;
+ }
while ((pendingTask = pendingAppendingTxnBufferTasks.poll()) != null) {
final ByteBuf data = pendingTask.buffer;
final CompletableFuture<Position> pendingFuture =
@@ -434,7 +470,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
- // ThreadSafe
+ // Acquires the topic monitor: never call while holding the buffer monitor, since topic close takes both.
private void updateLastDispatchablePosition(Position position) {
topic.updateLastDispatchablePosition(position);
}
@@ -464,6 +500,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
snapshotAbortedTxnProcessor.trimExpiredAbortedTxns();
takeSnapshotByChangeTimes();
}
+ updateLastDispatchablePosition(null);
txnCommittedCounter.increment();
completableFuture.complete(null);
}
@@ -509,8 +546,13 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
removeTxnAndUpdateMaxReadPosition(txnID);
snapshotAbortedTxnProcessor.trimExpiredAbortedTxns();
takeSnapshotByChangeTimes();
- txnAbortedCounter.increment();
- completableFuture.complete(null);
+ }
+ updateLastDispatchablePosition(null);
+ txnAbortedCounter.increment();
+ completableFuture.complete(null);
+ // Complete first so the low-water-mark abort continuation releases its permit before
+ // we scan for the next transaction. Completion must remain outside the buffer monitor.
+ synchronized (TopicTransactionBuffer.this) {
handleLowWaterMark(txnID, lowWaterMark);
}
}
@@ -601,8 +643,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
} else {
updateMaxReadPosition(topic.getManagedLedger().getLastConfirmedEntry(), false);
}
- // Update the last dispatchable position to null if there is a TXN finished.
- updateLastDispatchablePosition(null);
}
/**
@@ -679,17 +719,30 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
@Override
public CompletableFuture<Void> closeAsync() {
- synchronized (pendingAppendingTxnBufferTasks) {
- if (!checkIfClosed()) {
+ boolean closeStarted;
+ // Serialize closure with recovery entry handling and replay submission, which use the same monitor.
+ synchronized (this) {
+ closeStarted = !checkIfClosed();
+ changeToCloseState();
+ if (recoveryReplayTask != null) {
+ recoveryReplayTask.cancel(false);
+ recoveryReplayTask = null;
+ }
+ }
+ // Cancel snapshot recovery before completing futures whose callbacks may run inline.
+ CompletableFuture<Void> processorCloseFuture = this.snapshotAbortedTxnProcessor.closeAsync();
+ if (closeStarted) {
+ Throwable closeException =
+ new BrokerServiceException.ServiceUnitNotReadyException("Topic is closed");
+ getTransactionBufferFuture().completeExceptionally(closeException);
+ synchronized (pendingAppendingTxnBufferTasks) {
PendingAppendingTxnBufferTask pendingTask = null;
- Throwable t = new BrokerServiceException.ServiceUnitNotReadyException("Topic is closed");
while ((pendingTask = pendingAppendingTxnBufferTasks.poll()) != null) {
- pendingTask.fail(t);
+ pendingTask.fail(closeException);
}
}
- changeToCloseState();
}
- return this.snapshotAbortedTxnProcessor.closeAsync();
+ return processorCloseFuture;
}
@Override
@@ -793,8 +846,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private final TopicTransactionBufferRecoverCallBack callBack;
- private Position startReadCursorPosition = PositionFactory.EARLIEST;
-
private final SpscArrayQueue<Entry> entryQueue;
private final AtomicLong exceptionNumber = new AtomicLong();
@@ -815,7 +866,6 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
this.abortedTxnProcessor = abortedTxnProcessor;
}
- @SneakyThrows
@Override
public void run() {
if (!this.topicTransactionBuffer.changeToInitializingState()) {
@@ -823,30 +873,62 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
this, topic.getName());
return;
}
- abortedTxnProcessor.recoverFromSnapshot().thenAccept(startReadCursorPosition -> {
- //Transaction is not use for this topic, so just make maxReadPosition as LAC.
- if (startReadCursorPosition == null) {
- callBack.noNeedToRecover();
- return;
- } else {
- this.startReadCursorPosition = startReadCursorPosition;
- }
- ManagedCursor managedCursor;
+ abortedTxnProcessor.recoverFromSnapshot().thenAccept(this::submitReplay)
+ .exceptionally(this::handleRecoveryFailure);
+ }
+
+ private void submitReplay(Position recoveredPosition) {
+ topicTransactionBuffer.submitRecoveryReplay(() -> {
try {
- managedCursor = topic.getManagedLedger()
- .newNonDurableCursor(this.startReadCursorPosition, SUBSCRIPTION_NAME);
- } catch (ManagedLedgerException e) {
+ replayTransactionBuffer(recoveredPosition);
+ } catch (Throwable error) {
+ handleRecoveryFailure(error);
+ }
+ });
+ }
+
+ private Void handleRecoveryFailure(Throwable error) {
+ Throwable cause = FutureUtil.unwrapCompletionException(error);
+ if (!shouldStopRecovery()) {
+ log.error("[{}]Transaction buffer failed to recover snapshot!", topic.getName(), cause);
+ }
+ callBack.recoverExceptionally(cause);
+ return null;
+ }
+
+ private boolean shouldStopRecovery() {
+ // Topic deletion can be rolled back; only buffer closure is irreversible.
+ return topicTransactionBuffer.checkIfClosed();
+ }
+
+ private void replayTransactionBuffer(Position recoveredPosition) {
+ if (shouldStopRecovery()) {
+ return;
+ }
+ // Transaction is not used for this topic, so just make maxReadPosition as LAC.
+ if (recoveredPosition == null) {
+ callBack.noNeedToRecover();
+ return;
+ }
+ ManagedCursor managedCursor;
+ try {
+ managedCursor = topic.getManagedLedger()
+ .newNonDurableCursor(recoveredPosition, SUBSCRIPTION_NAME);
+ } catch (ManagedLedgerException e) {
+ if (!shouldStopRecovery()) {
callBack.recoverExceptionally(e);
log.error("[{}]Transaction buffer recover fail when open cursor!", topic.getName(), e);
- return;
}
- Position lastConfirmedEntry =
- topic.getManagedLedger().getLastConfirmedEntry();
- Position currentLoadPosition = this.startReadCursorPosition;
- FillEntryQueueCallback fillEntryQueueCallback = new FillEntryQueueCallback(entryQueue,
- managedCursor, TopicTransactionBufferRecover.this);
+ return;
+ }
+ Position lastConfirmedEntry = topic.getManagedLedger().getLastConfirmedEntry();
+ Position currentLoadPosition = recoveredPosition;
+ FillEntryQueueCallback fillEntryQueueCallback = new FillEntryQueueCallback(entryQueue,
+ managedCursor, TopicTransactionBufferRecover.this);
+ try {
if (lastConfirmedEntry.getEntryId() != -1) {
- while (lastConfirmedEntry.compareTo(currentLoadPosition) > 0
+ while (!shouldStopRecovery()
+ && lastConfirmedEntry.compareTo(currentLoadPosition) > 0
&& fillEntryQueueCallback.fillQueue()) {
Entry entry = entryQueue.poll();
if (entry != null) {
@@ -861,19 +943,19 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
try {
Thread.sleep(1);
} catch (InterruptedException e) {
- //no-op
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
}
}
}
}
-
+ } finally {
+ fillEntryQueueCallback.stopAndReleasePendingEntries();
closeCursor(SUBSCRIPTION_NAME);
+ }
+ if (!shouldStopRecovery()) {
callBack.recoverComplete();
- }).exceptionally(e -> {
- callBack.recoverExceptionally(e.getCause());
- log.error("[{}]Transaction buffer failed to recover snapshot!", topic.getName(), e);
- return null;
- });
+ }
}
private void closeCursor(String subscriptionName) {
@@ -929,6 +1011,8 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
private volatile boolean isReadable = true;
+ private boolean stopped;
+
private static final int NUMBER_OF_PER_READ_ENTRY = 100;
private FillEntryQueueCallback(SpscArrayQueue<Entry> entryQueue, ManagedCursor cursor,
@@ -954,7 +1038,12 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
@Override
- public void readEntriesComplete(List<Entry> entries, Object ctx) {
+ public synchronized void readEntriesComplete(List<Entry> entries, Object ctx) {
+ if (stopped) {
+ // An asynchronous read can complete after replay has stopped and no longer has a consumer.
+ entries.forEach(Entry::release);
+ return;
+ }
entryQueue.fill(new MessagePassingQueue.Supplier<Entry>() {
private int i = 0;
@Override
@@ -968,8 +1057,19 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
outstandingReadsRequests.decrementAndGet();
}
+ private synchronized void stopAndReleasePendingEntries() {
+ stopped = true;
+ Entry entry;
+ while ((entry = entryQueue.poll()) != null) {
+ entry.release();
+ }
+ }
+
@Override
- public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ public synchronized void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ if (stopped) {
+ return;
+ }
if (recover.topic.getManagedLedger().getConfig().isAutoSkipNonRecoverableData()
&& exception instanceof ManagedLedgerException.NonRecoverableLedgerException
|| exception instanceof ManagedLedgerException.ManagedLedgerFencedException
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
index 52786863f73e..5c9bd157d577 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
@@ -38,6 +38,7 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import io.netty.buffer.Unpooled;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
@@ -70,6 +71,7 @@ import lombok.Cleanup;
import lombok.Lombok;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.common.util.Bytes;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
@@ -1650,7 +1652,7 @@ public class TransactionTest extends TransactionTestBase {
public void testTBRecoverChangeStateError() throws InterruptedException, TimeoutException {
final AtomicReference<PersistentTopic> persistentTopic = new AtomicReference<>();
// Create Executor
- ScheduledExecutorService executorServiceRecover = mock(ScheduledExecutorService.class);
+ ListeningScheduledExecutorService executorServiceRecover = mock(ListeningScheduledExecutorService.class);
// Mock serviceConfiguration.
ServiceConfiguration serviceConfiguration = mock(ServiceConfiguration.class);
when(serviceConfiguration.isEnableReplicatedSubscriptions()).thenReturn(false);
@@ -1704,6 +1706,9 @@ public class TransactionTest extends TransactionTestBase {
when(pulsar.getConfiguration()).thenReturn(serviceConfiguration);
when(pulsar.getConfig()).thenReturn(serviceConfiguration);
when(pulsar.getTransactionExecutorProvider()).thenReturn(executorProvider);
+ OrderedScheduler snapshotRecoverExecutorProvider = mock(OrderedScheduler.class);
+ when(snapshotRecoverExecutorProvider.chooseThread(any(Object.class))).thenReturn(executorServiceRecover);
+ when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(snapshotRecoverExecutorProvider);
when(pulsar.getTransactionBufferSnapshotServiceFactory()).thenReturn(transactionBufferSnapshotServiceFactory);
TopicTransactionBufferProvider topicTransactionBufferProvider = new TopicTransactionBufferProvider();
when(pulsar.getTransactionBufferProvider()).thenReturn(topicTransactionBufferProvider);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java
new file mode 100644
index 000000000000..5310a1099614
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.policies.data.TransactionBufferStats;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class AbstractSnapshotAbortedTxnProcessorTest {
+
+ @Test(timeOut = 10_000)
+ public void testCloseCancelsQueuedRecovery() throws Exception {
+ try (RecoveryTestContext context = RecoveryTestContext.queued()) {
+ CompletableFuture<?> recoveryFuture = context.processor.recoverFromSnapshot();
+ CompletableFuture<Boolean> callbackHeldProcessorLock =
+ trackCallbackLock(recoveryFuture, context.processor);
+ CompletableFuture<Boolean> closeCompletedBeforeRecovery = recoveryFuture.handle(
+ (__, ___) -> context.processor.closeAsync().isDone());
+ CompletableFuture<?> repeatedRecoveryFuture = context.processor.recoverFromSnapshot();
+ context.awaitRecoveryQueued();
+
+ context.processor.closeAsync().get(5, TimeUnit.SECONDS);
+
+ assertTrue(context.submittedRecoveryTask().isCancelled());
+ context.verifyRecoveryFailedAfterClose(recoveryFuture);
+ context.verifyRecoveryFailedAfterClose(repeatedRecoveryFuture);
+ assertFalse(context.processor.recoveryStarted());
+ assertTrue(context.processor.resourcesClosed());
+ assertFalse(callbackHeldProcessorLock.get(1, TimeUnit.SECONDS),
+ "Recovery callbacks must run without holding the processor lock");
+ assertTrue(closeCompletedBeforeRecovery.get(1, TimeUnit.SECONDS),
+ "Close must finish before publishing a close-induced recovery failure");
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseWaitsForRunningRecovery() throws Exception {
+ try (RecoveryTestContext context = RecoveryTestContext.running()) {
+ CompletableFuture<?> recoveryFuture = context.processor.recoverFromSnapshot();
+ context.awaitRecoveryStarted();
+
+ Future<?> recoveryTask = context.submittedRecoveryTask();
+ CompletableFuture<Void> closeFuture = context.processor.closeAsync();
+
+ assertFalse(recoveryTask.isCancelled());
+ assertFalse(recoveryFuture.isDone(), "Running recovery must not complete until its task stops");
+ assertFalse(closeFuture.isDone(), "Closing must wait for running recovery before closing resources");
+ assertFalse(context.processor.resourcesClosed());
+
+ context.finishRecovery();
+ context.verifyRecoveryFailedAfterClose(recoveryFuture);
+ closeFuture.get(5, TimeUnit.SECONDS);
+ assertTrue(context.processor.resourcesClosed());
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseDoesNotWaitForRecoveryCallback() throws Exception {
+ CountDownLatch callbackStarted = new CountDownLatch(1);
+ CountDownLatch finishCallback = new CountDownLatch(1);
+ try (RecoveryTestContext context = RecoveryTestContext.running()) {
+ CompletableFuture<?> recoveryFuture = context.processor.recoverFromSnapshot();
+ CompletableFuture<Void> callbackFuture = recoveryFuture.thenRun(() -> {
+ assertFalse(Thread.holdsLock(context.processor),
+ "Recovery callbacks must run without holding the processor lock");
+ callbackStarted.countDown();
+ try {
+ assertTrue(finishCallback.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ });
+ context.awaitRecoveryStarted();
+
+ context.finishRecovery();
+ assertTrue(callbackStarted.await(5, TimeUnit.SECONDS));
+ CompletableFuture<Void> closeFuture = context.processor.closeAsync();
+
+ closeFuture.get(5, TimeUnit.SECONDS);
+ assertTrue(context.processor.resourcesClosed());
+ context.verifyRecoveryFailedAfterClose(context.processor.recoverFromSnapshot());
+
+ finishCallback.countDown();
+ callbackFuture.get(5, TimeUnit.SECONDS);
+ } finally {
+ finishCallback.countDown();
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testRecoverySubmissionRejected() throws Exception {
+ TrackingScheduledExecutor recoveryExecutor = new TrackingScheduledExecutor();
+ recoveryExecutor.shutdown();
+ TestSnapshotProcessor processor = new TestSnapshotProcessor(recoveryExecutor);
+
+ CompletableFuture<Position> recoveryFuture = processor.recoverFromSnapshot();
+
+ ExecutionException exception = expectThrows(ExecutionException.class, recoveryFuture::get);
+ assertTrue(exception.getCause() instanceof RejectedExecutionException);
+ processor.closeAsync().get(5, TimeUnit.SECONDS);
+ assertTrue(processor.resourcesClosed());
+ }
+
+ private static CompletableFuture<Boolean> trackCallbackLock(CompletableFuture<?> recoveryFuture,
+ Object processor) {
+ CompletableFuture<Boolean> callbackHeldProcessorLock = new CompletableFuture<>();
+ recoveryFuture.whenComplete((__, ___) ->
+ callbackHeldProcessorLock.complete(Thread.holdsLock(processor)));
+ return callbackHeldProcessorLock;
+ }
+
+ private static final class RecoveryTestContext implements AutoCloseable {
+
+ private final TrackingScheduledExecutor recoveryExecutor = new TrackingScheduledExecutor();
+ private final CountDownLatch releaseBlocker = new CountDownLatch(1);
+ private final TestSnapshotProcessor processor;
+
+ static RecoveryTestContext queued() throws Exception {
+ return new RecoveryTestContext(true);
+ }
+
+ static RecoveryTestContext running() throws Exception {
+ return new RecoveryTestContext(false);
+ }
+
+ private RecoveryTestContext(boolean queueRecovery) throws Exception {
+ if (queueRecovery) {
+ CountDownLatch blockerStarted = new CountDownLatch(1);
+ recoveryExecutor.execute(() -> {
+ blockerStarted.countDown();
+ try {
+ releaseBlocker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(blockerStarted.await(5, TimeUnit.SECONDS));
+ }
+
+ processor = new TestSnapshotProcessor(recoveryExecutor);
+ }
+
+ void awaitRecoveryQueued() {
+ assertEquals(recoveryExecutor.getQueue().size(), 1);
+ assertFalse(submittedRecoveryTask().isCancelled());
+ }
+
+ void awaitRecoveryStarted() throws InterruptedException {
+ assertTrue(processor.awaitRecoveryStarted());
+ }
+
+ Future<?> submittedRecoveryTask() {
+ Future<?> recoveryTask = recoveryExecutor.submittedTask();
+ assertNotNull(recoveryTask);
+ return recoveryTask;
+ }
+
+ void finishRecovery() {
+ processor.finishRecovery();
+ }
+
+ void verifyRecoveryFailedAfterClose(CompletableFuture<?> recoveryFuture) throws Exception {
+ ExecutionException exception = expectThrows(ExecutionException.class,
+ () -> recoveryFuture.get(5, TimeUnit.SECONDS));
+ assertTrue(exception.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException,
+ "Closing the processor must fail the recovery future");
+ }
+
+ @Override
+ public void close() throws Exception {
+ releaseBlocker.countDown();
+ processor.finishRecovery();
+ recoveryExecutor.shutdownNow();
+ assertTrue(recoveryExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ private static final class TrackingScheduledExecutor extends ScheduledThreadPoolExecutor {
+
+ private final AtomicReference<Future<?>> submittedTask = new AtomicReference<>();
+
+ private TrackingScheduledExecutor() {
+ super(1);
+ }
+
+ @Override
+ public Future<?> submit(Runnable task) {
+ Future<?> future = super.submit(task);
+ submittedTask.set(future);
+ return future;
+ }
+
+ private Future<?> submittedTask() {
+ return submittedTask.get();
+ }
+ }
+
+ private static final class TestSnapshotProcessor extends AbstractSnapshotAbortedTxnProcessor {
+
+ private final CountDownLatch recoveryStarted = new CountDownLatch(1);
+ private final CountDownLatch finishRecovery = new CountDownLatch(1);
+ private final AtomicBoolean resourcesClosed = new AtomicBoolean();
+
+ private TestSnapshotProcessor(ScheduledExecutorService recoveryExecutor) {
+ super(recoveryExecutor);
+ }
+
+ @Override
+ Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception {
+ recoveryStarted.countDown();
+ assertTrue(finishRecovery.await(5, TimeUnit.SECONDS));
+ return null;
+ }
+
+ @Override
+ CompletableFuture<Void> closeResources() {
+ resourcesClosed.set(true);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void putAbortedTxnAndPosition(TxnID txnID, Position position) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void trimExpiredAbortedTxns() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean checkAbortedTransaction(TxnID txnID) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<Void> clearAbortedTxnSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<Void> takeAbortedTxnsSnapshot(Position maxReadPosition) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) {
+ throw new UnsupportedOperationException();
+ }
+
+ boolean awaitRecoveryStarted() throws InterruptedException {
+ return recoveryStarted.await(5, TimeUnit.SECONDS);
+ }
+
+ boolean recoveryStarted() {
+ return recoveryStarted.getCount() == 0;
+ }
+
+ void finishRecovery() {
+ finishRecovery.countDown();
+ }
+
+ boolean resourcesClosed() {
+ return resourcesClosed.get();
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java
new file mode 100644
index 000000000000..5f2196a2f13d
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.Collections;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
+import org.apache.bookkeeper.mledger.ReadOnlyManagedLedger;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService;
+import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter;
+import org.apache.pulsar.broker.service.TransactionBufferSnapshotServiceFactory;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndex;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexesMetadata;
+import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotSegment;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class SnapshotSegmentAbortedTxnProcessorCloseTest {
+
+ @Test(timeOut = 10_000)
+ @SuppressWarnings("unchecked")
+ public void testCloseWaitsForRecoveryIndexUpdate() throws Exception {
+ ListeningScheduledExecutorService recoveryExecutor =
+ MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor());
+ CompletableFuture<MessageId> indexUpdateFuture = new CompletableFuture<>();
+ CountDownLatch indexUpdateStarted = new CountDownLatch(1);
+ SnapshotSegmentAbortedTxnProcessorImpl processor = null;
+ try {
+ String topicName = "persistent://public/default/test-close-during-index-update";
+ PersistentTopic topic = mock(PersistentTopic.class);
+ BrokerService brokerService = mock(BrokerService.class);
+ PulsarService pulsar = mock(PulsarService.class);
+ ServiceConfiguration configuration = mock(ServiceConfiguration.class);
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ ClientConfigurationData clientConfiguration = new ClientConfigurationData();
+ clientConfiguration.setOperationTimeoutMs(5_000);
+ OrderedScheduler recoveryScheduler = mock(OrderedScheduler.class);
+ TransactionBufferSnapshotServiceFactory serviceFactory =
+ mock(TransactionBufferSnapshotServiceFactory.class);
+
+ when(topic.getName()).thenReturn(topicName);
+ when(topic.getBrokerService()).thenReturn(brokerService);
+ when(brokerService.getPulsar()).thenReturn(pulsar);
+ when(pulsar.getConfiguration()).thenReturn(configuration);
+ when(pulsar.getClient()).thenReturn(client);
+ when(client.getConfiguration()).thenReturn(clientConfiguration);
+ when(configuration.getTransactionBufferSnapshotSegmentSize()).thenReturn(1024);
+ when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(recoveryScheduler);
+ when(recoveryScheduler.chooseThread(any(Object.class))).thenReturn(recoveryExecutor);
+ when(pulsar.getTransactionBufferSnapshotServiceFactory()).thenReturn(serviceFactory);
+
+ SystemTopicTxnBufferSnapshotService<TransactionBufferSnapshotSegment> segmentService =
+ mock(SystemTopicTxnBufferSnapshotService.class);
+ SystemTopicTxnBufferSnapshotService<TransactionBufferSnapshotIndexes> indexService =
+ mock(SystemTopicTxnBufferSnapshotService.class);
+ ReferenceCountedWriter<TransactionBufferSnapshotSegment> segmentWriter =
+ mock(ReferenceCountedWriter.class);
+ ReferenceCountedWriter<TransactionBufferSnapshotIndexes> indexWriter =
+ mock(ReferenceCountedWriter.class);
+ SystemTopicClient.Writer<TransactionBufferSnapshotSegment> segmentSystemWriter =
+ mock(SystemTopicClient.Writer.class);
+ SystemTopicClient.Writer<TransactionBufferSnapshotIndexes> indexSystemWriter =
+ mock(SystemTopicClient.Writer.class);
+
+ when(serviceFactory.getTxnBufferSnapshotSegmentService()).thenReturn(segmentService);
+ when(serviceFactory.getTxnBufferSnapshotIndexService()).thenReturn(indexService);
+ when(segmentService.getReferenceWriter(any())).thenReturn(segmentWriter);
+ when(indexService.getReferenceWriter(any())).thenReturn(indexWriter);
+ when(segmentWriter.getFuture()).thenReturn(CompletableFuture.completedFuture(segmentSystemWriter));
+ when(indexWriter.getFuture()).thenReturn(CompletableFuture.completedFuture(indexSystemWriter));
+ when(indexSystemWriter.writeAsync(anyString(), any())).thenAnswer(invocation -> {
+ indexUpdateStarted.countDown();
+ return indexUpdateFuture;
+ });
+
+ TransactionBufferSnapshotIndex invalidIndex = new TransactionBufferSnapshotIndex(0, 1, 1, 2, 2);
+ TransactionBufferSnapshotIndexesMetadata metadata =
+ new TransactionBufferSnapshotIndexesMetadata(3, 3, Collections.emptyList());
+ TransactionBufferSnapshotIndexes indexes =
+ new TransactionBufferSnapshotIndexes(topicName, Collections.singletonList(invalidIndex), metadata);
+ TableView<TransactionBufferSnapshotIndexes> tableView = mock(TableView.class);
+ when(indexService.getTableView(recoveryExecutor)).thenReturn(tableView);
+ when(tableView.readLatest(topicName)).thenReturn(indexes);
+
+ ManagedLedgerImpl managedLedger = mock(ManagedLedgerImpl.class);
+ when(topic.getManagedLedger()).thenReturn(managedLedger);
+ when(managedLedger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(managedLedger.getLedgersInfo()).thenReturn(new TreeMap<>());
+ ManagedLedgerFactory managedLedgerFactory = mock(ManagedLedgerFactory.class);
+ ReadOnlyManagedLedger readOnlyManagedLedger = mock(ReadOnlyManagedLedger.class);
+ when(brokerService.getManagedLedgerFactoryForTopic(any()))
+ .thenReturn(CompletableFuture.completedFuture(managedLedgerFactory));
+ doAnswer(invocation -> {
+ AsyncCallbacks.OpenReadOnlyManagedLedgerCallback callback = invocation.getArgument(1);
+ callback.openReadOnlyManagedLedgerComplete(readOnlyManagedLedger, invocation.getArgument(3));
+ return null;
+ }).when(managedLedgerFactory).asyncOpenReadOnlyManagedLedger(anyString(), any(), any(), isNull());
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntryCallback callback = invocation.getArgument(1);
+ callback.readEntryFailed(new ManagedLedgerException("missing segment"), invocation.getArgument(2));
+ return null;
+ }).when(readOnlyManagedLedger).asyncReadEntry(any(), any(), isNull());
+
+ processor = new SnapshotSegmentAbortedTxnProcessorImpl(topic);
+ processor.recoverFromSnapshot().get(5, TimeUnit.SECONDS);
+ assertTrue(indexUpdateStarted.await(5, TimeUnit.SECONDS));
+ // A later recovery must not replace the in-flight update in the close barrier.
+ processor.recoverFromSnapshot().get(5, TimeUnit.SECONDS);
+
+ CompletableFuture<Void> closeFuture = processor.closeAsync();
+
+ assertFalse(closeFuture.isDone(), "Closing must wait for the recovery index update");
+ verify(indexWriter, never()).release();
+
+ indexUpdateFuture.complete(MessageId.earliest);
+ closeFuture.get(5, TimeUnit.SECONDS);
+ verify(indexWriter).release();
+ } finally {
+ indexUpdateFuture.complete(MessageId.earliest);
+ if (processor != null) {
+ processor.closeAsync().get(5, TimeUnit.SECONDS);
+ }
+ recoveryExecutor.shutdownNow();
+ assertTrue(recoveryExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java
new file mode 100644
index 000000000000..9d766cef4640
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.ListeningScheduledExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.buffer.ByteBuf;
+import io.netty.util.HashedWheelTimer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+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.AtomicReference;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.api.proto.MarkerType;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class TopicTransactionBufferCloseTest {
+
+ @DataProvider
+ public Object[][] transactionCompletionModes() {
+ return new Object[][] { { true, false }, { true, true }, { false, false }, { false, true } };
+ }
+
+ @Test(dataProvider = "transactionCompletionModes", timeOut = 10_000)
+ public void testTxnCompletionUpdatesTopicOutsideBufferLock(boolean replay, boolean abort) throws Exception {
+ Position position = PositionFactory.create(1, 1);
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ try (TestContext context = new TestContext(recoveryFuture, position)) {
+ // Fenced publish failures close the buffer while holding the topic monitor.
+ doAnswer(__ -> {
+ assertThat(Thread.holdsLock(context.transactionBuffer))
+ .as("updating the topic must not acquire its monitor while holding the buffer monitor")
+ .isFalse();
+ return null;
+ }).when(context.topic).updateLastDispatchablePosition(null);
+
+ TxnID txnID = new TxnID(1, 1);
+ if (replay) {
+ Entry marker = mock(Entry.class);
+ when(marker.getLedgerId()).thenReturn(position.getLedgerId());
+ when(marker.getEntryId()).thenReturn(position.getEntryId());
+ when(marker.getMessageMetadata()).thenReturn(new MessageMetadata()
+ .setTxnidMostBits(txnID.getMostSigBits())
+ .setTxnidLeastBits(txnID.getLeastSigBits())
+ .setMarkerType((abort ? MarkerType.TXN_ABORT : MarkerType.TXN_COMMIT).getValue()));
+ when(context.managedCursor.hasMoreEntries()).thenReturn(true, false);
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(1);
+ callback.readEntriesComplete(List.of(marker), null);
+ return null;
+ }).when(context.managedCursor).asyncReadEntries(anyInt(), any(), anyLong(), any());
+
+ recoveryFuture.complete(PositionFactory.create(1, 0));
+ context.transactionBuffer.getTransactionBufferFuture().get(5, TimeUnit.SECONDS);
+ verify(marker).release();
+ } else {
+ recoveryFuture.complete(position);
+ context.transactionBuffer.getTransactionBufferFuture().get(5, TimeUnit.SECONDS);
+ doAnswer(invocation -> {
+ AsyncCallbacks.AddEntryCallback callback = invocation.getArgument(1);
+ callback.addComplete(position, invocation.getArgument(0), null);
+ return null;
+ }).when(context.managedLedger).asyncAddEntry(any(ByteBuf.class), any(), any());
+
+ CompletableFuture<Void> completion = abort
+ ? context.transactionBuffer.abortTxn(txnID, 0)
+ : context.transactionBuffer.commitTxn(txnID, 0);
+ completion.get(5, TimeUnit.SECONDS);
+ }
+ verify(context.topic).updateLastDispatchablePosition(null);
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseInducedRecoveryFailureDoesNotCloseTopicAgain() throws Exception {
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) {
+ context.awaitExecutorsIdle();
+ when(context.processor.closeAsync()).thenAnswer(__ -> {
+ recoveryFuture.completeExceptionally(
+ new BrokerServiceException.ServiceUnitNotReadyException("processor closed"));
+ return CompletableFuture.completedFuture(null);
+ });
+
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ context.awaitExecutorsIdle();
+
+ verify(context.topic, never()).close(true);
+ assertTrue(context.transactionBuffer.getTransactionBufferFuture().isCompletedExceptionally());
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testRecoveryContinuationDoesNotStartAfterClose() throws Exception {
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) {
+ context.awaitExecutorsIdle();
+
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ recoveryFuture.complete(PositionFactory.EARLIEST);
+ context.awaitExecutorsIdle();
+
+ verify(context.managedLedger, never()).newNonDurableCursor(any(), anyString());
+ assertTrue(context.transactionBuffer.getTransactionBufferFuture().isCompletedExceptionally());
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testCloseCancelsQueuedRecoveryReplay() throws Exception {
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ CountDownLatch blockerStarted = new CountDownLatch(1);
+ CountDownLatch releaseBlocker = new CountDownLatch(1);
+ try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) {
+ try {
+ context.awaitExecutorsIdle();
+ AtomicReference<Future<?>> submittedTask = new AtomicReference<>();
+ doAnswer(invocation -> {
+ Future<?> task = (Future<?>) invocation.callRealMethod();
+ submittedTask.set(task);
+ return task;
+ }).when(context.recoveryExecutor).submit(any(Runnable.class));
+ context.recoveryExecutor.execute(() -> {
+ blockerStarted.countDown();
+ try {
+ releaseBlocker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue(blockerStarted.await(5, TimeUnit.SECONDS));
+
+ recoveryFuture.complete(PositionFactory.EARLIEST);
+ Future<?> replayTask = submittedTask.get();
+ assertNotNull(replayTask);
+
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+
+ assertTrue(replayTask.isCancelled());
+ } finally {
+ releaseBlocker.countDown();
+ }
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ public void testLateRecoveryReadIsReleasedAfterClose() throws Exception {
+ Position startPosition = PositionFactory.create(1, 0);
+ Position lastPosition = PositionFactory.create(1, 3);
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ CountDownLatch readStarted = new CountDownLatch(1);
+ AtomicReference<AsyncCallbacks.ReadEntriesCallback> readCallback = new AtomicReference<>();
+ try (TestContext context = new TestContext(recoveryFuture, lastPosition)) {
+ when(context.managedCursor.hasMoreEntries()).thenReturn(true);
+ doAnswer(invocation -> {
+ readCallback.set(invocation.getArgument(1));
+ readStarted.countDown();
+ return null;
+ }).when(context.managedCursor).asyncReadEntries(anyInt(), any(), anyLong(), any());
+
+ recoveryFuture.complete(startPosition);
+ assertTrue(readStarted.await(5, TimeUnit.SECONDS));
+ // Live transaction work must still run while the recovery read is outstanding.
+ context.transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ context.awaitExecutorsIdle();
+
+ Entry lateEntry = mock(Entry.class);
+ readCallback.get().readEntriesComplete(List.of(lateEntry), null);
+
+ verify(lateEntry).release();
+ verify(context.managedCursor).asyncReadEntries(anyInt(), any(), anyLong(), any());
+ }
+ }
+
+ private static final class TestContext implements AutoCloseable {
+ private final ExecutorService transactionExecutor = Executors.newSingleThreadExecutor();
+ private final ListeningScheduledExecutorService recoveryExecutor =
+ spy(MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor()));
+ private final AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ private final PersistentTopic topic = mock(PersistentTopic.class);
+ private final ManagedLedgerImpl managedLedger = mock(ManagedLedgerImpl.class);
+ private final ManagedCursor managedCursor = mock(ManagedCursor.class);
+ private final TopicTransactionBuffer transactionBuffer;
+
+ private TestContext(CompletableFuture<Position> recoveryFuture, Position lastConfirmedEntry) throws Exception {
+ BrokerService brokerService = mock(BrokerService.class);
+ PulsarService pulsar = mock(PulsarService.class);
+ ServiceConfiguration configuration = mock(ServiceConfiguration.class);
+ ExecutorProvider executorProvider = mock(ExecutorProvider.class);
+ OrderedScheduler recoveryScheduler = mock(OrderedScheduler.class);
+
+ when(topic.getName()).thenReturn("persistent://public/default/test-close-during-recovery");
+ when(topic.getBrokerService()).thenReturn(brokerService);
+ when(topic.getManagedLedger()).thenReturn(managedLedger);
+ when(brokerService.getPulsar()).thenReturn(pulsar);
+ when(pulsar.getConfiguration()).thenReturn(configuration);
+ when(pulsar.getTransactionTimer()).thenReturn(mock(HashedWheelTimer.class));
+ when(pulsar.getTransactionExecutorProvider()).thenReturn(executorProvider);
+ when(executorProvider.getExecutor(any(Object.class))).thenReturn(transactionExecutor);
+ when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(recoveryScheduler);
+ when(recoveryScheduler.chooseThread(any(Object.class))).thenReturn(recoveryExecutor);
+ when(managedLedger.getLastConfirmedEntry()).thenReturn(lastConfirmedEntry);
+ when(managedLedger.getConfig()).thenReturn(new ManagedLedgerConfig());
+ when(managedLedger.newNonDurableCursor(any(), anyString())).thenReturn(managedCursor);
+ when(processor.recoverFromSnapshot()).thenReturn(recoveryFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+
+ transactionBuffer = new TopicTransactionBuffer(topic, processor, AbortedTxnProcessor.SnapshotType.Single);
+ }
+
+ private void awaitExecutorsIdle() throws Exception {
+ transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ recoveryExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public void close() throws Exception {
+ transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS);
+ transactionExecutor.shutdownNow();
+ recoveryExecutor.shutdownNow();
+ assertTrue(transactionExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ assertTrue(recoveryExecutor.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
index 93d154b8625d..16bdfacc6c05 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
@@ -18,8 +18,14 @@
*/
package org.apache.pulsar.broker.transaction.buffer.impl;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -28,17 +34,22 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -70,6 +81,83 @@ public class TopicTransactionBufferRecoveryTest extends ProducerConsumerBase {
return new Object[][] { { false }, { true } };
}
+ @Test(dataProvider = "snapshotExists", timeOut = 30_000)
+ public void testRecoveryNotifiesLastPositionQueryOutsideBufferLock(boolean snapshotExists) throws Exception {
+ String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tb-recovery-last-position");
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ when(processor.recoverFromSnapshot()).thenReturn(recoveryFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+ pulsar.setTransactionBufferProvider(topic -> new TopicTransactionBuffer(
+ (PersistentTopic) topic, processor, AbortedTxnProcessor.SnapshotType.Single));
+ try {
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopic(topicName, true).get(5, TimeUnit.SECONDS).orElseThrow();
+ Position lastConfirmedEntry = topic.getManagedLedger().getLastConfirmedEntry();
+ assertThat(lastConfirmedEntry.getEntryId()).isEqualTo(-1);
+
+ // GetLastMessageId registers this continuation; an empty ledger's lookup can take the topic lock inline.
+ CompletableFuture<Position> lastPositionFuture = topic.checkIfTransactionBufferRecoverCompletely()
+ .thenCompose(__ -> {
+ assertThat(Thread.holdsLock(topic.getTransactionBuffer()))
+ .as("recovery callbacks must run outside the buffer monitor")
+ .isFalse();
+ return topic.getLastDispatchablePosition();
+ });
+
+ recoveryFuture.complete(snapshotExists ? PositionFactory.EARLIEST : null);
+ assertThat(lastPositionFuture.get(5, TimeUnit.SECONDS)).isEqualTo(lastConfirmedEntry);
+ } finally {
+ recoveryFuture.complete(null);
+ pulsar.setTransactionBufferProvider(originalProvider);
+ }
+ }
+
+ @Test(dataProvider = "snapshotExists", timeOut = 30_000)
+ public void testRecoveryCompletesAfterDeletionFails(boolean snapshotExists) throws Exception {
+ String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tb-recovery-delete-failure");
+ TopicName parsedTopicName = TopicName.get(topicName);
+ CompletableFuture<Position> recoveryFuture = new CompletableFuture<>();
+ CompletableFuture<SchemaVersion> schemaDeleteFuture = new CompletableFuture<>();
+ SchemaRegistryService schemaRegistryService = spy(pulsar.getSchemaRegistryService());
+ doReturn(schemaDeleteFuture).when(schemaRegistryService).deleteSchemaStorage(parsedTopicName.getSchemaName());
+ AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ when(processor.recoverFromSnapshot()).thenReturn(recoveryFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+ pulsar.setTransactionBufferProvider(topic -> new TopicTransactionBuffer(
+ (PersistentTopic) topic, processor, AbortedTxnProcessor.SnapshotType.Single));
+ doReturn(schemaRegistryService).when(pulsar).getSchemaRegistryService();
+ try {
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopic(topicName, true).get(5, TimeUnit.SECONDS).orElseThrow();
+ ExecutorService transactionExecutor =
+ pulsar.getTransactionExecutorProvider().getExecutor(topic.getTransactionBuffer());
+ // Ensure the recovery continuation is registered before completing its future.
+ transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ CompletableFuture<Void> deleteFuture = topic.delete();
+ verify(schemaRegistryService, timeout(5_000)).deleteSchemaStorage(parsedTopicName.getSchemaName());
+ assertThat(topic.isClosingOrDeleting()).isTrue();
+
+ recoveryFuture.complete(snapshotExists ? PositionFactory.EARLIEST : null);
+ // Let replay run while deletion is pending, regardless of which executor dispatches it.
+ transactionExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ pulsar.getTransactionSnapshotRecoverExecutorProvider().chooseThread(parsedTopicName.getNamespace())
+ .submit(() -> { }).get(5, TimeUnit.SECONDS);
+
+ schemaDeleteFuture.completeExceptionally(new IllegalStateException("schema deletion failed"));
+ assertThatThrownBy(() -> deleteFuture.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(IllegalStateException.class);
+ assertThat(topic.isClosingOrDeleting()).isFalse();
+ topic.getTransactionBuffer().checkIfTBRecoverCompletely().get(5, TimeUnit.SECONDS);
+ } finally {
+ schemaDeleteFuture.completeExceptionally(new IllegalStateException("test finished"));
+ doCallRealMethod().when(pulsar).getSchemaRegistryService();
+ pulsar.setTransactionBufferProvider(originalProvider);
+ }
+ }
+
/**
* While the transaction buffer is recovering, normal publishes don't move the max read position, so they
* don't update the topic's lastMaxReadPositionMovedForwardTimestamp either. When recovery completes, the