From fbd31ae360b5b847cb26540afbd3ea8b8aa008c8 Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sat, 15 Aug 2026 14:13:24 +0800 Subject: [PATCH 1/7] [fix][broker] Cancel queued transaction snapshot recovery on topic close Cancel queued snapshot recovery tasks when the transaction buffer closes and wait for running recovery before releasing snapshot resources. This prevents the recovery executor queue from retaining closed topics and managed ledgers. --- .../AbstractSnapshotAbortedTxnProcessor.java | 164 +++++++++ ...SingleSnapshotAbortedTxnProcessorImpl.java | 49 +-- ...napshotSegmentAbortedTxnProcessorImpl.java | 93 +++-- ...stractSnapshotAbortedTxnProcessorTest.java | 321 ++++++++++++++++++ 4 files changed, 548 insertions(+), 79 deletions(-) create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java 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 0000000000000..214fb8c9e2e62 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java @@ -0,0 +1,164 @@ +/* + * 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; + +/** + * Coordinates snapshot recovery and resource closure so resources remain available until recovery stops. + */ +abstract class AbstractSnapshotAbortedTxnProcessor implements AbortedTxnProcessor { + + private enum State { + OPEN, + RECOVERY_QUEUED, + RECOVERY_RUNNING, + RECOVERY_FINISHED, + CLOSED + } + + private final ScheduledExecutorService recoveryExecutor; + + private volatile State state = State.OPEN; + private Future recoveryTask; + private CompletableFuture recoveryFuture = CompletableFuture.completedFuture(null); + private final CompletableFuture closeFuture = new CompletableFuture<>(); + + protected AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) { + this.recoveryExecutor = recoveryExecutor; + } + + @Override + public final synchronized CompletableFuture recoverFromSnapshot() { + 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. + CompletableFuture future = new CompletableFuture<>(); + this.recoveryFuture = future; + this.state = State.RECOVERY_QUEUED; + try { + this.recoveryTask = recoveryExecutor.submit(() -> runRecovery(future)); + } catch (RejectedExecutionException e) { + this.state = State.RECOVERY_FINISHED; + future.completeExceptionally(e); + } + // Do not expose the internal future used to determine when recovery has stopped. + return future.copy(); + } + + private void runRecovery(CompletableFuture future) { + try { + if (!tryStartRecovery()) { + future.completeExceptionally(closedException()); + return; + } + Position recoveredPosition = doRecoverFromSnapshot(recoveryExecutor); + if (tryMarkRecoveryFinished()) { + future.complete(recoveredPosition); + } else { + future.completeExceptionally(closedException()); + } + } catch (Throwable throwable) { + future.completeExceptionally(tryMarkRecoveryFinished() ? throwable : closedException()); + } + } + + /** + * Recovers the processor state synchronously on the recovery executor thread. + * + * @param executor executor used by recovery-related operations + * @return the recovery position, or {@code null} when no snapshot exists + */ + protected abstract Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception; + + protected final boolean isClosed() { + return this.state == State.CLOSED; + } + + private synchronized boolean tryStartRecovery() { + if (this.state == State.CLOSED) { + return false; + } + this.state = State.RECOVERY_RUNNING; + return true; + } + + private synchronized boolean tryMarkRecoveryFinished() { + if (this.state == State.CLOSED) { + return false; + } + this.state = State.RECOVERY_FINISHED; + return true; + } + + @Override + public final CompletableFuture closeAsync() { + boolean recoveryOwnsFutureCompletion; + CompletableFuture currentRecoveryFuture; + synchronized (this) { + if (this.state == State.CLOSED) { + return closeFuture; + } + State previousState = this.state; + this.state = State.CLOSED; + recoveryOwnsFutureCompletion = previousState == State.RECOVERY_RUNNING + || previousState == State.RECOVERY_FINISHED; + currentRecoveryFuture = this.recoveryFuture; + if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) { + this.recoveryTask.cancel(false); + } + } + if (!recoveryOwnsFutureCompletion) { + currentRecoveryFuture.completeExceptionally(closedException()); + } + currentRecoveryFuture.handle((v, throwable) -> null) + .thenCompose(v -> closeResources()) + .whenComplete((v, throwable) -> { + if (throwable != null) { + closeFuture.completeExceptionally(throwable); + } else { + closeFuture.complete(null); + } + }); + return closeFuture; + } + + /** + * Closes resources after recovery has stopped. + * + *

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. + */ + protected abstract CompletableFuture 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 4c4833cd87898..85b1895f65d3b 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.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.policies.data.TransactionBufferStats; @CustomLog -public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcessor { +public class SingleSnapshotAbortedTxnProcessorImpl extends AbstractSnapshotAbortedTxnProcessor { private final PersistentTopic topic; private final ReferenceCountedWriter 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) -> { @@ -90,29 +89,18 @@ public boolean checkAbortedTransaction(TxnID txnID) { } @Override - public CompletableFuture recoverFromSnapshot() { - final var future = new CompletableFuture(); + protected 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()) { + return null; + } + if (snapshot == null) { + return null; + } + handleSnapshot(snapshot); + return PositionFactory.create(snapshot.getMaxReadPositionLedgerId(), snapshot.getMaxReadPositionEntryId()); } @Override @@ -177,11 +165,8 @@ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) { } @Override - public synchronized CompletableFuture closeAsync() { - if (!isClosed) { - isClosed = true; - takeSnapshotWriter.release(); - } + protected CompletableFuture 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 8c28115c9d405..29c96b06ace4c 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.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; @@ -71,10 +70,11 @@ import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.util.FutureUtil; -public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcessor { +public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbortedTxnProcessor { private static final Logger LOG = Logger.get(SnapshotSegmentAbortedTxnProcessorImpl.class); private final Logger log; + private final PersistentTopic topic; /** * Stored the unsealed aborted transaction IDs Whose size is always less than the snapshotSegmentCapacity. @@ -120,8 +120,6 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess */ private final LinkedMap indexes = new LinkedMap<>(); - private final PersistentTopic topic; - private volatile long lastSnapshotTimestamps; private volatile long lastTakedSnapshotSegmentTimestamp; @@ -144,6 +142,8 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess 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.log = LOG.with().attr("topic", topic.getName()).build(); this.persistentWorker = new PersistentWorker(topic); @@ -231,48 +231,45 @@ public CompletableFuture takeAbortedTxnsSnapshot(Position maxReadPosition) } @Override - public CompletableFuture recoverFromSnapshot() { + protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { final var pulsar = topic.getBrokerService().getPulsar(); - final var future = new CompletableFuture(); - 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 { 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 { @@ -350,22 +347,24 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) { } // 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); @@ -413,7 +412,7 @@ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) { } @Override - public CompletableFuture closeAsync() { + protected CompletableFuture closeResources() { return persistentWorker.closeAsync(); } 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 0000000000000..92cf3d0046d67 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java @@ -0,0 +1,321 @@ +/* + * 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.assertNull; +import static org.testng.Assert.assertSame; +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.recoverFromSnapshot(); + 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()); + } + } + + @Test(timeOut = 10_000) + public void testCloseWaitsForRunningRecovery() throws Exception { + try (RecoveryTestContext context = RecoveryTestContext.running()) { + CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + context.awaitRecoveryStarted(); + + Future recoveryTask = context.submittedRecoveryTask(); + CompletableFuture 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 testCloseAfterRecoveryCompleted() throws Exception { + try (RecoveryTestContext context = RecoveryTestContext.running()) { + CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + context.awaitRecoveryStarted(); + + context.finishRecovery(); + context.verifyRecoverySucceeded(recoveryFuture); + context.processor.closeAsync().get(5, TimeUnit.SECONDS); + + assertTrue(context.processor.resourcesClosed()); + context.verifyRecoveryFailedAfterClose(context.recoverFromSnapshot()); + } + } + + @Test(timeOut = 10_000) + public void testRetryAfterRecoveryFailed() throws Exception { + try (RecoveryTestContext context = RecoveryTestContext.running()) { + RuntimeException failure = new RuntimeException("recovery failed"); + context.processor.failNextRecovery(failure); + CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + context.awaitRecoveryStarted(); + + context.finishRecovery(); + context.verifyRecoveryFailed(recoveryFuture, failure); + context.verifyRecoverySucceeded(context.recoverFromSnapshot()); + + context.processor.closeAsync().get(5, TimeUnit.SECONDS); + } + } + + @Test(timeOut = 10_000) + public void testRecoverySubmissionRejected() throws Exception { + TrackingScheduledExecutor recoveryExecutor = new TrackingScheduledExecutor(); + recoveryExecutor.shutdown(); + TestSnapshotProcessor processor = new TestSnapshotProcessor(recoveryExecutor); + + CompletableFuture 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 final class RecoveryTestContext implements AutoCloseable { + + private final TrackingScheduledExecutor recoveryExecutor = new TrackingScheduledExecutor(); + private final CountDownLatch releaseBlocker = new CountDownLatch(1); + private final TestSnapshotProcessor processor; + + private CompletableFuture recoveryCallbackHeldProcessorLock; + + 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); + } + + CompletableFuture recoverFromSnapshot() { + CompletableFuture recoveryFuture = processor.recoverFromSnapshot(); + recoveryCallbackHeldProcessorLock = new CompletableFuture<>(); + recoveryFuture.whenComplete((__, ___) -> + recoveryCallbackHeldProcessorLock.complete(Thread.holdsLock(processor))); + return recoveryFuture; + } + + 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"); + verifyRecoveryCallbackDidNotHoldProcessorLock(); + } + + void verifyRecoverySucceeded(CompletableFuture recoveryFuture) throws Exception { + assertNull(recoveryFuture.get(5, TimeUnit.SECONDS)); + verifyRecoveryCallbackDidNotHoldProcessorLock(); + } + + void verifyRecoveryFailed(CompletableFuture recoveryFuture, Throwable expected) { + ExecutionException exception = expectThrows(ExecutionException.class, + () -> recoveryFuture.get(5, TimeUnit.SECONDS)); + assertSame(exception.getCause(), expected); + } + + private void verifyRecoveryCallbackDidNotHoldProcessorLock() throws Exception { + assertFalse(recoveryCallbackHeldProcessorLock.get(1, TimeUnit.SECONDS), + "Recovery callbacks must run without holding the processor lock"); + } + + @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> 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 final AtomicReference nextRecoveryFailure = new AtomicReference<>(); + + private TestSnapshotProcessor(ScheduledExecutorService recoveryExecutor) { + super(recoveryExecutor); + } + + @Override + protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { + recoveryStarted.countDown(); + assertTrue(finishRecovery.await(5, TimeUnit.SECONDS)); + RuntimeException failure = nextRecoveryFailure.getAndSet(null); + if (failure != null) { + throw failure; + } + return null; + } + + @Override + protected CompletableFuture 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 clearAbortedTxnSnapshot() { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture 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(); + } + + void failNextRecovery(RuntimeException failure) { + nextRecoveryFailure.set(failure); + } + + boolean resourcesClosed() { + return resourcesClosed.get(); + } + } +} From 01bd2d6db7735d44c68ed7b48d61a5aa5f828327 Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sat, 15 Aug 2026 15:11:05 +0800 Subject: [PATCH 2/7] Replace ScheduledExecutorService with ListeningScheduledExecutorService in TransactionTest --- .../apache/pulsar/broker/transaction/TransactionTest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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 51e4f6c8d1bbe..b887eb318e460 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.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.CustomLog; import lombok.Lombok; 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.ManagedLedgerConfig; @@ -1620,7 +1622,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable { public void testTBRecoverChangeStateError() throws InterruptedException, TimeoutException { final AtomicReference 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); @@ -1674,6 +1676,9 @@ public void testTBRecoverChangeStateError() throws InterruptedException, Timeout 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); From 0eaf1b0068f4bfa9f26f265aaae4e41f3ca1558d Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sat, 15 Aug 2026 20:59:06 +0800 Subject: [PATCH 3/7] Ensure recovery completion is awaited during close operation --- .../AbstractSnapshotAbortedTxnProcessor.java | 18 +++++++--- ...stractSnapshotAbortedTxnProcessorTest.java | 33 +++++++++++++++++++ 2 files changed, 46 insertions(+), 5 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 index 214fb8c9e2e62..79328dae50bc3 100644 --- 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 @@ -44,6 +44,7 @@ private enum State { private volatile State state = State.OPEN; private Future recoveryTask; private CompletableFuture recoveryFuture = CompletableFuture.completedFuture(null); + private CompletableFuture recoveryStoppedFuture = CompletableFuture.completedFuture(null); private final CompletableFuture closeFuture = new CompletableFuture<>(); protected AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) { @@ -60,19 +61,22 @@ public final synchronized CompletableFuture recoverFromSnapshot() { } // Bind the task to this recovery attempt so a later retry cannot complete the wrong future. CompletableFuture future = new CompletableFuture<>(); + CompletableFuture stoppedFuture = new CompletableFuture<>(); this.recoveryFuture = future; + this.recoveryStoppedFuture = stoppedFuture; this.state = State.RECOVERY_QUEUED; try { - this.recoveryTask = recoveryExecutor.submit(() -> runRecovery(future)); + this.recoveryTask = recoveryExecutor.submit(() -> runRecovery(future, stoppedFuture)); } catch (RejectedExecutionException e) { this.state = State.RECOVERY_FINISHED; future.completeExceptionally(e); + stoppedFuture.complete(null); } - // Do not expose the internal future used to determine when recovery has stopped. + // Do not let callers complete the internal recovery result. return future.copy(); } - private void runRecovery(CompletableFuture future) { + private void runRecovery(CompletableFuture future, CompletableFuture stoppedFuture) { try { if (!tryStartRecovery()) { future.completeExceptionally(closedException()); @@ -86,6 +90,8 @@ private void runRecovery(CompletableFuture future) { } } catch (Throwable throwable) { future.completeExceptionally(tryMarkRecoveryFinished() ? throwable : closedException()); + } finally { + stoppedFuture.complete(null); } } @@ -121,6 +127,7 @@ private synchronized boolean tryMarkRecoveryFinished() { public final CompletableFuture closeAsync() { boolean recoveryOwnsFutureCompletion; CompletableFuture currentRecoveryFuture; + CompletableFuture currentRecoveryStoppedFuture; synchronized (this) { if (this.state == State.CLOSED) { return closeFuture; @@ -130,15 +137,16 @@ public final CompletableFuture closeAsync() { recoveryOwnsFutureCompletion = previousState == State.RECOVERY_RUNNING || previousState == State.RECOVERY_FINISHED; currentRecoveryFuture = this.recoveryFuture; + currentRecoveryStoppedFuture = this.recoveryStoppedFuture; if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) { this.recoveryTask.cancel(false); } } if (!recoveryOwnsFutureCompletion) { currentRecoveryFuture.completeExceptionally(closedException()); + currentRecoveryStoppedFuture.complete(null); } - currentRecoveryFuture.handle((v, throwable) -> null) - .thenCompose(v -> closeResources()) + currentRecoveryStoppedFuture.thenCompose(v -> closeResources()) .whenComplete((v, throwable) -> { if (throwable != null) { closeFuture.completeExceptionally(throwable); 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 index 92cf3d0046d67..bb0d4ebd3cd4f 100644 --- 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 @@ -82,6 +82,39 @@ public void testCloseWaitsForRunningRecovery() throws Exception { } } + @Test(timeOut = 10_000) + public void testCloseWaitsForSynchronousRecoveryCallback() throws Exception { + CountDownLatch callbackStarted = new CountDownLatch(1); + CountDownLatch finishCallback = new CountDownLatch(1); + try (RecoveryTestContext context = RecoveryTestContext.running()) { + CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + CompletableFuture callbackFuture = recoveryFuture.thenRun(() -> { + 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 closeFuture = context.processor.closeAsync(); + + assertFalse(closeFuture.isDone(), "Closing must wait for the recovery callback to return"); + assertFalse(context.processor.resourcesClosed()); + + finishCallback.countDown(); + callbackFuture.get(5, TimeUnit.SECONDS); + closeFuture.get(5, TimeUnit.SECONDS); + assertTrue(context.processor.resourcesClosed()); + } finally { + finishCallback.countDown(); + } + } + @Test(timeOut = 10_000) public void testCloseAfterRecoveryCompleted() throws Exception { try (RecoveryTestContext context = RecoveryTestContext.running()) { From b486f7907bffaa3efc43e25c81e34a3960187e1f Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sat, 15 Aug 2026 21:01:01 +0800 Subject: [PATCH 4/7] Prevent reading segment entries if the topic is closed --- .../buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java | 3 +++ 1 file changed, 3 insertions(+) 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 29c96b06ace4c..4ca38b58bb5e7 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 @@ -264,6 +264,9 @@ protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) thro } 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()) { From 1b70c10c173b76cc2872fd39bcf65f0867e4bac1 Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sun, 16 Aug 2026 09:51:05 +0800 Subject: [PATCH 5/7] Ensure recovery index update is awaited during resource closure --- ...napshotSegmentAbortedTxnProcessorImpl.java | 6 +- ...otSegmentAbortedTxnProcessorCloseTest.java | 177 ++++++++++++++++++ 2 files changed, 181 insertions(+), 2 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java 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 4ca38b58bb5e7..0d785167f1591 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 @@ -138,6 +138,7 @@ public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbor *

Clear all snapshot segment.

*/ private final PersistentWorker persistentWorker; + private CompletableFuture recoveryIndexUpdateFuture = CompletableFuture.completedFuture(null); private static final String SNAPSHOT_PREFIX = "multiple-"; @@ -299,7 +300,7 @@ private void readSegmentEntries(TopicName topicName, TransactionBufferSnapshotIn } if (hasInvalidIndex) { // Update the snapshot segment index if there exist invalid indexes. - persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, + recoveryIndexUpdateFuture = persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot())); } } @@ -416,7 +417,8 @@ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) { @Override protected CompletableFuture closeResources() { - return persistentWorker.closeAsync(); + return recoveryIndexUpdateFuture.handle((__, throwable) -> null) + .thenCompose(__ -> persistentWorker.closeAsync()); } private void handleSnapshotSegmentEntry(Entry entry) { 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 0000000000000..147bc6411758b --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java @@ -0,0 +1,177 @@ +/* + * 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) + public void testCloseWaitsForRecoveryIndexUpdate() throws Exception { + ListeningScheduledExecutorService recoveryExecutor = + MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor()); + CompletableFuture 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); + + @SuppressWarnings("unchecked") + SystemTopicTxnBufferSnapshotService segmentService = + mock(SystemTopicTxnBufferSnapshotService.class); + @SuppressWarnings("unchecked") + SystemTopicTxnBufferSnapshotService indexService = + mock(SystemTopicTxnBufferSnapshotService.class); + @SuppressWarnings("unchecked") + ReferenceCountedWriter segmentWriter = + mock(ReferenceCountedWriter.class); + @SuppressWarnings("unchecked") + ReferenceCountedWriter indexWriter = + mock(ReferenceCountedWriter.class); + @SuppressWarnings("unchecked") + SystemTopicClient.Writer segmentSystemWriter = + mock(SystemTopicClient.Writer.class); + @SuppressWarnings("unchecked") + SystemTopicClient.Writer 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); + @SuppressWarnings("unchecked") + TableView 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)); + + CompletableFuture 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)); + } + } +} From 9c10dde5c4697f1c46a62b05a789b08313807733 Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sat, 29 Aug 2026 09:53:14 +0800 Subject: [PATCH 6/7] Refactor recovery handling to ensure closure waits for processor-owned recovery work to finish --- .../AbstractSnapshotAbortedTxnProcessor.java | 128 ++++++----- ...SingleSnapshotAbortedTxnProcessorImpl.java | 4 +- ...napshotSegmentAbortedTxnProcessorImpl.java | 19 +- .../buffer/impl/TopicTransactionBuffer.java | 173 ++++++++++----- ...stractSnapshotAbortedTxnProcessorTest.java | 72 ++++--- ...otSegmentAbortedTxnProcessorCloseTest.java | 10 +- .../impl/TopicTransactionBufferCloseTest.java | 201 ++++++++++++++++++ 7 files changed, 459 insertions(+), 148 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java 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 index 79328dae50bc3..409696d6953f1 100644 --- 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 @@ -27,7 +27,8 @@ import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor; /** - * Coordinates snapshot recovery and resource closure so resources remain available until recovery stops. + * Coordinates snapshot recovery and resource closure so resources remain available until processor-owned recovery + * work finishes. */ abstract class AbstractSnapshotAbortedTxnProcessor implements AbortedTxnProcessor { @@ -44,71 +45,94 @@ private enum State { private volatile State state = State.OPEN; private Future recoveryTask; private CompletableFuture recoveryFuture = CompletableFuture.completedFuture(null); - private CompletableFuture recoveryStoppedFuture = CompletableFuture.completedFuture(null); + // Completes before the recovery result, so close waits only for processor-owned work. + private CompletableFuture recoveryWorkFinishedFuture = CompletableFuture.completedFuture(null); private final CompletableFuture closeFuture = new CompletableFuture<>(); - protected AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) { + AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) { this.recoveryExecutor = recoveryExecutor; } @Override - public final synchronized CompletableFuture recoverFromSnapshot() { - if (this.state == State.CLOSED) { - return CompletableFuture.failedFuture(closedException()); - } - if (!recoveryFuture.isDone()) { - return recoveryFuture.copy(); + public CompletableFuture recoverFromSnapshot() { + CompletableFuture newRecoveryFuture; + CompletableFuture 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.RECOVERY_FINISHED; + submissionFailure = e; + } } - // Bind the task to this recovery attempt so a later retry cannot complete the wrong future. - CompletableFuture future = new CompletableFuture<>(); - CompletableFuture stoppedFuture = new CompletableFuture<>(); - this.recoveryFuture = future; - this.recoveryStoppedFuture = stoppedFuture; - this.state = State.RECOVERY_QUEUED; - try { - this.recoveryTask = recoveryExecutor.submit(() -> runRecovery(future, stoppedFuture)); - } catch (RejectedExecutionException e) { - this.state = State.RECOVERY_FINISHED; - future.completeExceptionally(e); - stoppedFuture.complete(null); + if (submissionFailure != null) { + newRecoveryWorkFinishedFuture.complete(null); + newRecoveryFuture.completeExceptionally(submissionFailure); } // Do not let callers complete the internal recovery result. - return future.copy(); + return newRecoveryFuture.copy(); } - private void runRecovery(CompletableFuture future, CompletableFuture stoppedFuture) { + private void runRecovery(CompletableFuture recoveryResult, + CompletableFuture recoveryWorkFinished) { + Position recoveredPosition = null; + Throwable recoveryFailure = null; + boolean closeWon = false; try { if (!tryStartRecovery()) { - future.completeExceptionally(closedException()); - return; + closeWon = true; + } else { + recoveredPosition = doRecoverFromSnapshot(recoveryExecutor); + if (!tryMarkRecoveryFinished()) { + closeWon = true; + } } - Position recoveredPosition = doRecoverFromSnapshot(recoveryExecutor); + } catch (Throwable throwable) { if (tryMarkRecoveryFinished()) { - future.complete(recoveredPosition); + recoveryFailure = throwable; } else { - future.completeExceptionally(closedException()); + closeWon = true; } - } catch (Throwable throwable) { - future.completeExceptionally(tryMarkRecoveryFinished() ? throwable : closedException()); - } finally { - stoppedFuture.complete(null); + } + recoveryWorkFinished.complete(null); + if (closeWon) { + 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 */ - protected abstract Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception; + abstract Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception; - protected final boolean isClosed() { + final boolean isClosed() { return this.state == State.CLOSED; } private synchronized boolean tryStartRecovery() { - if (this.state == State.CLOSED) { + if (this.state != State.RECOVERY_QUEUED) { return false; } this.state = State.RECOVERY_RUNNING; @@ -116,7 +140,7 @@ private synchronized boolean tryStartRecovery() { } private synchronized boolean tryMarkRecoveryFinished() { - if (this.state == State.CLOSED) { + if (this.state != State.RECOVERY_RUNNING) { return false; } this.state = State.RECOVERY_FINISHED; @@ -124,29 +148,23 @@ private synchronized boolean tryMarkRecoveryFinished() { } @Override - public final CompletableFuture closeAsync() { - boolean recoveryOwnsFutureCompletion; + public CompletableFuture closeAsync() { + State previousState; CompletableFuture currentRecoveryFuture; - CompletableFuture currentRecoveryStoppedFuture; + CompletableFuture currentRecoveryWorkFinishedFuture; synchronized (this) { if (this.state == State.CLOSED) { return closeFuture; } - State previousState = this.state; + previousState = this.state; this.state = State.CLOSED; - recoveryOwnsFutureCompletion = previousState == State.RECOVERY_RUNNING - || previousState == State.RECOVERY_FINISHED; currentRecoveryFuture = this.recoveryFuture; - currentRecoveryStoppedFuture = this.recoveryStoppedFuture; + currentRecoveryWorkFinishedFuture = this.recoveryWorkFinishedFuture; if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) { this.recoveryTask.cancel(false); } } - if (!recoveryOwnsFutureCompletion) { - currentRecoveryFuture.completeExceptionally(closedException()); - currentRecoveryStoppedFuture.complete(null); - } - currentRecoveryStoppedFuture.thenCompose(v -> closeResources()) + currentRecoveryWorkFinishedFuture.thenCompose(v -> closeResources()) .whenComplete((v, throwable) -> { if (throwable != null) { closeFuture.completeExceptionally(throwable); @@ -154,16 +172,26 @@ public final CompletableFuture closeAsync() { closeFuture.complete(null); } }); + if (previousState == State.RECOVERY_QUEUED) { + failRecoveryAfterClose(currentRecoveryFuture); + } + if (previousState != State.RECOVERY_RUNNING) { + currentRecoveryWorkFinishedFuture.complete(null); + } return closeFuture; } + private void failRecoveryAfterClose(CompletableFuture recoveryResult) { + closeFuture.whenComplete((__, ___) -> recoveryResult.completeExceptionally(closedException())); + } + /** - * Closes resources after recovery has stopped. + * Closes resources after processor-owned recovery work has finished. * *

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. */ - protected abstract CompletableFuture closeResources(); + abstract CompletableFuture closeResources(); private static BrokerServiceException closedException() { return new BrokerServiceException.ServiceUnitNotReadyException( 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 85b1895f65d3b..800e5019be0e8 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 @@ -89,7 +89,7 @@ public boolean checkAbortedTransaction(TxnID txnID) { } @Override - protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { + Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { final var pulsar = topic.getBrokerService().getPulsar(); final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService() .getTableView(executor).readLatest(topic.getName()); @@ -165,7 +165,7 @@ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) { } @Override - protected CompletableFuture closeResources() { + CompletableFuture 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 0d785167f1591..480cd5f71575c 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 @@ -74,7 +74,6 @@ public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbor private static final Logger LOG = Logger.get(SnapshotSegmentAbortedTxnProcessorImpl.class); private final Logger log; - private final PersistentTopic topic; /** * Stored the unsealed aborted transaction IDs Whose size is always less than the snapshotSegmentCapacity. @@ -120,6 +119,8 @@ public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbor */ private final LinkedMap indexes = new LinkedMap<>(); + private final PersistentTopic topic; + private volatile long lastSnapshotTimestamps; private volatile long lastTakedSnapshotSegmentTimestamp; @@ -138,7 +139,8 @@ public class SnapshotSegmentAbortedTxnProcessorImpl extends AbstractSnapshotAbor *

Clear all snapshot segment.

*/ private final PersistentWorker persistentWorker; - private CompletableFuture recoveryIndexUpdateFuture = CompletableFuture.completedFuture(null); + // A failed recovery can be retried, so close must retain updates started by every attempt. + private CompletableFuture recoveryIndexUpdatesFuture = CompletableFuture.completedFuture(null); private static final String SNAPSHOT_PREFIX = "multiple-"; @@ -232,7 +234,7 @@ public CompletableFuture takeAbortedTxnsSnapshot(Position maxReadPosition) } @Override - protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { + Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { final var pulsar = topic.getBrokerService().getPulsar(); final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory() .getTxnBufferSnapshotIndexService().getTableView(executor) @@ -298,10 +300,11 @@ private void readSegmentEntries(TopicName topicName, TransactionBufferSnapshotIn } } } - if (hasInvalidIndex) { + if (hasInvalidIndex && !isClosed()) { // Update the snapshot segment index if there exist invalid indexes. - recoveryIndexUpdateFuture = persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, - () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot())); + recoveryIndexUpdatesFuture = CompletableFuture.allOf(recoveryIndexUpdatesFuture, + persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, + () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot()))); } } @@ -416,8 +419,8 @@ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) { } @Override - protected CompletableFuture closeResources() { - return recoveryIndexUpdateFuture.handle((__, throwable) -> null) + CompletableFuture closeResources() { + return recoveryIndexUpdatesFuture.handle((__, throwable) -> null) .thenCompose(__ -> persistentWorker.closeAsync()); } 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 c51e8e57a7d39..0a52fa37d5ade 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 @@ -27,12 +27,13 @@ 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.Executor; 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 org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -149,11 +150,15 @@ public TopicTransactionBuffer(PersistentTopic topic) { private void recover() { recoverTime.setRecoverStartTime(System.currentTimeMillis()); - this.topic.getBrokerService().getPulsar().getTransactionExecutorProvider().getExecutor(this) - .execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() { + Executor transactionExecutor = this.topic.getBrokerService().getPulsar() + .getTransactionExecutorProvider().getExecutor(this); + transactionExecutor.execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() { @Override public void recoverComplete() { synchronized (TopicTransactionBuffer.this) { + if (checkIfClosed()) { + return; + } if (ongoingTxns.isEmpty()) { updateMaxReadPositionAfterRecovery(); } @@ -175,6 +180,9 @@ public void recoverComplete() { @Override public void noNeedToRecover() { synchronized (TopicTransactionBuffer.this) { + if (checkIfClosed()) { + return; + } updateMaxReadPositionAfterRecovery(); if (!changeToNoSnapshotState()) { log.error().log("Transaction buffer recover fail"); @@ -197,6 +205,9 @@ public void handleTxnEntry(Entry entry) { TxnID txnID = new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits()); Position position = PositionFactory.create(entry.getLedgerId(), entry.getEntryId()); synchronized (TopicTransactionBuffer.this) { + if (checkIfClosed()) { + return; + } if (Markers.isTxnMarker(msgMetadata)) { if (Markers.isTxnAbortMarker(msgMetadata)) { snapshotAbortedTxnProcessor.putAbortedTxnAndPosition(txnID, position); @@ -211,11 +222,6 @@ public void handleTxnEntry(Entry entry) { @Override public void recoverExceptionally(Throwable e) { - - log.warn() - .exception(e) - .log("Closing topic due to read transaction buffer snapshot while recovering the" - + " transaction buffer throw exception"); // when create reader or writer fail throw PulsarClientException, // should close this topic and then reinit this topic if (e instanceof PulsarClientException) { @@ -228,9 +234,16 @@ public void recoverExceptionally(Throwable e) { getTransactionBufferFuture().completeExceptionally(e); } recoverTime.setRecoverEndTime(System.currentTimeMillis()); + if (checkIfClosed() || topic.isClosingOrDeleting()) { + return; + } + log.warn() + .exception(e) + .log("Closing topic due to read transaction buffer snapshot while recovering the" + + " transaction buffer throw exception"); topic.close(true); } - }, this.topic, this, snapshotAbortedTxnProcessor)); + }, this.topic, this, snapshotAbortedTxnProcessor, transactionExecutor)); } @Override @@ -314,6 +327,11 @@ public CompletableFuture appendBufferToTxn(TxnID txnId, long sequenceI 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 pendingFuture = @@ -684,17 +702,26 @@ public CompletableFuture clearSnapshotAndClose() { @Override public CompletableFuture closeAsync() { - synchronized (pendingAppendingTxnBufferTasks) { - if (!checkIfClosed()) { + boolean closeStarted; + // Serialize closure with recovery entry handling, which uses the same monitor. + synchronized (this) { + closeStarted = !checkIfClosed(); + changeToCloseState(); + } + // Cancel queued recovery before completing futures whose callbacks may run inline. + CompletableFuture 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 @@ -798,8 +825,6 @@ public static class TopicTransactionBufferRecover implements Runnable { private final TopicTransactionBufferRecoverCallBack callBack; - private Position startReadCursorPosition = PositionFactory.EARLIEST; - private final SpscArrayQueue entryQueue; private final AtomicLong exceptionNumber = new AtomicLong(); @@ -810,17 +835,19 @@ public static class TopicTransactionBufferRecover implements Runnable { private final AbortedTxnProcessor abortedTxnProcessor; + private final Executor replayExecutor; + private TopicTransactionBufferRecover(TopicTransactionBufferRecoverCallBack callBack, PersistentTopic topic, TopicTransactionBuffer transactionBuffer, - AbortedTxnProcessor abortedTxnProcessor) { + AbortedTxnProcessor abortedTxnProcessor, Executor replayExecutor) { this.topic = topic; this.callBack = callBack; this.entryQueue = new SpscArrayQueue<>(2000); this.topicTransactionBuffer = transactionBuffer; this.abortedTxnProcessor = abortedTxnProcessor; + this.replayExecutor = replayExecutor; } - @SneakyThrows @Override public void run() { if (!this.topicTransactionBuffer.changeToInitializingState()) { @@ -829,32 +856,56 @@ public void run() { .log("TransactionBuffer of topic can not change state to Initializing"); 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; - try { - managedCursor = topic.getManagedLedger() - .newNonDurableCursor(this.startReadCursorPosition, SUBSCRIPTION_NAME); - } catch (ManagedLedgerException e) { + // Transaction-buffer replay must not extend the snapshot processor's close barrier. + abortedTxnProcessor.recoverFromSnapshot().thenAcceptAsync(this::replayTransactionBuffer, replayExecutor) + .exceptionally(this::handleRecoveryFailure); + } + + private Void handleRecoveryFailure(Throwable error) { + Throwable cause = FutureUtil.unwrapCompletionException(error); + if (!shouldStopRecovery()) { + topicTransactionBuffer.log.error() + .exception(cause) + .log("Transaction buffer failed to recover snapshot"); + } + callBack.recoverExceptionally(cause); + return null; + } + + private boolean shouldStopRecovery() { + return topicTransactionBuffer.checkIfClosed() || topic.isClosingOrDeleting(); + } + + 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); topicTransactionBuffer.log.error() .exception(e) .log("Transaction buffer recover fail when open cursor!"); - 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) { @@ -869,21 +920,19 @@ public void run() { 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()); - topicTransactionBuffer.log.error() - .exception(e) - .log("Transaction buffer failed to recover snapshot"); - return null; - }); + } } private void closeCursor(String subscriptionName) { @@ -940,6 +989,8 @@ static class FillEntryQueueCallback implements AsyncCallbacks.ReadEntriesCallbac private volatile boolean isReadable = true; + private volatile boolean stopped; + private static final int NUMBER_OF_PER_READ_ENTRY = 100; private FillEntryQueueCallback(SpscArrayQueue entryQueue, ManagedCursor cursor, @@ -949,6 +1000,9 @@ private FillEntryQueueCallback(SpscArrayQueue entryQueue, ManagedCursor c this.recover = recover; } boolean fillQueue() { + if (stopped) { + return false; + } if (entryQueue.size() + NUMBER_OF_PER_READ_ENTRY < entryQueue.capacity() && outstandingReadsRequests.get() == 0) { if (cursor.hasMoreEntries()) { @@ -965,7 +1019,13 @@ boolean fillQueue() { } @Override - public void readEntriesComplete(List entries, Object ctx) { + public synchronized void readEntriesComplete(List entries, Object ctx) { + if (stopped) { + // An asynchronous read can complete after replay has stopped and no longer has a consumer. + entries.forEach(Entry::release); + outstandingReadsRequests.decrementAndGet(); + return; + } entryQueue.fill(new MessagePassingQueue.Supplier() { private int i = 0; @Override @@ -979,8 +1039,21 @@ public Entry get() { outstandingReadsRequests.decrementAndGet(); } + private synchronized void stopAndReleasePendingEntries() { + stopped = true; + isReadable = false; + 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) { + outstandingReadsRequests.decrementAndGet(); + 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/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java index bb0d4ebd3cd4f..0ed02607a5e52 100644 --- 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 @@ -34,6 +34,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; 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.bookkeeper.mledger.Position; import org.apache.pulsar.broker.service.BrokerServiceException; @@ -47,7 +48,11 @@ public class AbstractSnapshotAbortedTxnProcessorTest { @Test(timeOut = 10_000) public void testCloseCancelsQueuedRecovery() throws Exception { try (RecoveryTestContext context = RecoveryTestContext.queued()) { - CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); + CompletableFuture callbackHeldProcessorLock = + trackCallbackLock(recoveryFuture, context.processor); + CompletableFuture closeCompletedBeforeRecovery = recoveryFuture.handle( + (__, ___) -> context.processor.closeAsync().isDone()); CompletableFuture repeatedRecoveryFuture = context.processor.recoverFromSnapshot(); context.awaitRecoveryQueued(); @@ -58,13 +63,17 @@ public void testCloseCancelsQueuedRecovery() throws Exception { 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.recoverFromSnapshot(); + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); context.awaitRecoveryStarted(); Future recoveryTask = context.submittedRecoveryTask(); @@ -83,11 +92,11 @@ public void testCloseWaitsForRunningRecovery() throws Exception { } @Test(timeOut = 10_000) - public void testCloseWaitsForSynchronousRecoveryCallback() throws Exception { + public void testCloseDoesNotWaitForRecoveryCallback() throws Exception { CountDownLatch callbackStarted = new CountDownLatch(1); CountDownLatch finishCallback = new CountDownLatch(1); try (RecoveryTestContext context = RecoveryTestContext.running()) { - CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); CompletableFuture callbackFuture = recoveryFuture.thenRun(() -> { callbackStarted.countDown(); try { @@ -103,13 +112,11 @@ public void testCloseWaitsForSynchronousRecoveryCallback() throws Exception { assertTrue(callbackStarted.await(5, TimeUnit.SECONDS)); CompletableFuture closeFuture = context.processor.closeAsync(); - assertFalse(closeFuture.isDone(), "Closing must wait for the recovery callback to return"); - assertFalse(context.processor.resourcesClosed()); + closeFuture.get(5, TimeUnit.SECONDS); + assertTrue(context.processor.resourcesClosed()); finishCallback.countDown(); callbackFuture.get(5, TimeUnit.SECONDS); - closeFuture.get(5, TimeUnit.SECONDS); - assertTrue(context.processor.resourcesClosed()); } finally { finishCallback.countDown(); } @@ -118,15 +125,21 @@ public void testCloseWaitsForSynchronousRecoveryCallback() throws Exception { @Test(timeOut = 10_000) public void testCloseAfterRecoveryCompleted() throws Exception { try (RecoveryTestContext context = RecoveryTestContext.running()) { - CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); + CompletableFuture callbackHeldProcessorLock = + trackCallbackLock(recoveryFuture, context.processor); context.awaitRecoveryStarted(); context.finishRecovery(); context.verifyRecoverySucceeded(recoveryFuture); + context.verifyRecoverySucceeded(context.processor.recoverFromSnapshot()); + assertEquals(context.processor.recoveryAttempts(), 2); + assertFalse(callbackHeldProcessorLock.get(1, TimeUnit.SECONDS), + "Recovery callbacks must run without holding the processor lock"); context.processor.closeAsync().get(5, TimeUnit.SECONDS); assertTrue(context.processor.resourcesClosed()); - context.verifyRecoveryFailedAfterClose(context.recoverFromSnapshot()); + context.verifyRecoveryFailedAfterClose(context.processor.recoverFromSnapshot()); } } @@ -135,12 +148,12 @@ public void testRetryAfterRecoveryFailed() throws Exception { try (RecoveryTestContext context = RecoveryTestContext.running()) { RuntimeException failure = new RuntimeException("recovery failed"); context.processor.failNextRecovery(failure); - CompletableFuture recoveryFuture = context.recoverFromSnapshot(); + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); context.awaitRecoveryStarted(); context.finishRecovery(); context.verifyRecoveryFailed(recoveryFuture, failure); - context.verifyRecoverySucceeded(context.recoverFromSnapshot()); + context.verifyRecoverySucceeded(context.processor.recoverFromSnapshot()); context.processor.closeAsync().get(5, TimeUnit.SECONDS); } @@ -160,14 +173,20 @@ public void testRecoverySubmissionRejected() throws Exception { assertTrue(processor.resourcesClosed()); } + private static CompletableFuture trackCallbackLock(CompletableFuture recoveryFuture, + Object processor) { + CompletableFuture 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; - private CompletableFuture recoveryCallbackHeldProcessorLock; - static RecoveryTestContext queued() throws Exception { return new RecoveryTestContext(true); } @@ -193,14 +212,6 @@ private RecoveryTestContext(boolean queueRecovery) throws Exception { processor = new TestSnapshotProcessor(recoveryExecutor); } - CompletableFuture recoverFromSnapshot() { - CompletableFuture recoveryFuture = processor.recoverFromSnapshot(); - recoveryCallbackHeldProcessorLock = new CompletableFuture<>(); - recoveryFuture.whenComplete((__, ___) -> - recoveryCallbackHeldProcessorLock.complete(Thread.holdsLock(processor))); - return recoveryFuture; - } - void awaitRecoveryQueued() { assertEquals(recoveryExecutor.getQueue().size(), 1); assertFalse(submittedRecoveryTask().isCancelled()); @@ -225,12 +236,10 @@ void verifyRecoveryFailedAfterClose(CompletableFuture recoveryFuture) throws () -> recoveryFuture.get(5, TimeUnit.SECONDS)); assertTrue(exception.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException, "Closing the processor must fail the recovery future"); - verifyRecoveryCallbackDidNotHoldProcessorLock(); } void verifyRecoverySucceeded(CompletableFuture recoveryFuture) throws Exception { assertNull(recoveryFuture.get(5, TimeUnit.SECONDS)); - verifyRecoveryCallbackDidNotHoldProcessorLock(); } void verifyRecoveryFailed(CompletableFuture recoveryFuture, Throwable expected) { @@ -239,11 +248,6 @@ void verifyRecoveryFailed(CompletableFuture recoveryFuture, Throwable expecte assertSame(exception.getCause(), expected); } - private void verifyRecoveryCallbackDidNotHoldProcessorLock() throws Exception { - assertFalse(recoveryCallbackHeldProcessorLock.get(1, TimeUnit.SECONDS), - "Recovery callbacks must run without holding the processor lock"); - } - @Override public void close() throws Exception { releaseBlocker.countDown(); @@ -279,13 +283,15 @@ private static final class TestSnapshotProcessor extends AbstractSnapshotAborted private final CountDownLatch finishRecovery = new CountDownLatch(1); private final AtomicBoolean resourcesClosed = new AtomicBoolean(); private final AtomicReference nextRecoveryFailure = new AtomicReference<>(); + private final AtomicInteger recoveryAttempts = new AtomicInteger(); private TestSnapshotProcessor(ScheduledExecutorService recoveryExecutor) { super(recoveryExecutor); } @Override - protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { + Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { + recoveryAttempts.incrementAndGet(); recoveryStarted.countDown(); assertTrue(finishRecovery.await(5, TimeUnit.SECONDS)); RuntimeException failure = nextRecoveryFailure.getAndSet(null); @@ -295,8 +301,12 @@ protected Position doRecoverFromSnapshot(ScheduledExecutorService executor) thro return null; } + private int recoveryAttempts() { + return recoveryAttempts.get(); + } + @Override - protected CompletableFuture closeResources() { + CompletableFuture closeResources() { resourcesClosed.set(true); return CompletableFuture.completedFuture(null); } 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 index 147bc6411758b..5f2196a2f13de 100644 --- 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 @@ -64,6 +64,7 @@ public class SnapshotSegmentAbortedTxnProcessorCloseTest { @Test(timeOut = 10_000) + @SuppressWarnings("unchecked") public void testCloseWaitsForRecoveryIndexUpdate() throws Exception { ListeningScheduledExecutorService recoveryExecutor = MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor()); @@ -94,22 +95,16 @@ public void testCloseWaitsForRecoveryIndexUpdate() throws Exception { when(recoveryScheduler.chooseThread(any(Object.class))).thenReturn(recoveryExecutor); when(pulsar.getTransactionBufferSnapshotServiceFactory()).thenReturn(serviceFactory); - @SuppressWarnings("unchecked") SystemTopicTxnBufferSnapshotService segmentService = mock(SystemTopicTxnBufferSnapshotService.class); - @SuppressWarnings("unchecked") SystemTopicTxnBufferSnapshotService indexService = mock(SystemTopicTxnBufferSnapshotService.class); - @SuppressWarnings("unchecked") ReferenceCountedWriter segmentWriter = mock(ReferenceCountedWriter.class); - @SuppressWarnings("unchecked") ReferenceCountedWriter indexWriter = mock(ReferenceCountedWriter.class); - @SuppressWarnings("unchecked") SystemTopicClient.Writer segmentSystemWriter = mock(SystemTopicClient.Writer.class); - @SuppressWarnings("unchecked") SystemTopicClient.Writer indexSystemWriter = mock(SystemTopicClient.Writer.class); @@ -129,7 +124,6 @@ public void testCloseWaitsForRecoveryIndexUpdate() throws Exception { new TransactionBufferSnapshotIndexesMetadata(3, 3, Collections.emptyList()); TransactionBufferSnapshotIndexes indexes = new TransactionBufferSnapshotIndexes(topicName, Collections.singletonList(invalidIndex), metadata); - @SuppressWarnings("unchecked") TableView tableView = mock(TableView.class); when(indexService.getTableView(recoveryExecutor)).thenReturn(tableView); when(tableView.readLatest(topicName)).thenReturn(indexes); @@ -156,6 +150,8 @@ public void testCloseWaitsForRecoveryIndexUpdate() throws Exception { 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 closeFuture = processor.closeAsync(); 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 0000000000000..df7da49c0ef2c --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java @@ -0,0 +1,201 @@ +/* + * 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.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.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertTrue; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +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.util.ExecutorProvider; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class TopicTransactionBufferCloseTest { + + @Test(timeOut = 10_000) + public void testCloseInducedRecoveryFailureDoesNotCloseTopicAgain() throws Exception { + ContinuationBlockingRecoveryFuture recoveryFuture = new ContinuationBlockingRecoveryFuture(); + recoveryFuture.allowContinuationRegistration(); + try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) { + recoveryFuture.awaitContinuationRegistrationStarted(); + when(context.processor.closeAsync()).thenAnswer(__ -> { + recoveryFuture.completeExceptionally( + new BrokerServiceException.ServiceUnitNotReadyException("processor closed")); + return CompletableFuture.completedFuture(null); + }); + + context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS); + context.awaitExecutorIdle(); + + verify(context.topic, never()).close(true); + assertTrue(context.transactionBuffer.getTransactionBufferFuture().isCompletedExceptionally()); + } + } + + @Test(timeOut = 10_000) + public void testRecoveryContinuationDoesNotStartAfterClose() throws Exception { + ContinuationBlockingRecoveryFuture recoveryFuture = new ContinuationBlockingRecoveryFuture(); + try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) { + recoveryFuture.awaitContinuationRegistrationStarted(); + recoveryFuture.complete(PositionFactory.EARLIEST); + + context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS); + recoveryFuture.allowContinuationRegistration(); + context.awaitExecutorIdle(); + + verify(context.managedLedger, never()).newNonDurableCursor(any(), anyString()); + assertTrue(context.transactionBuffer.getTransactionBufferFuture().isCompletedExceptionally()); + } finally { + recoveryFuture.allowContinuationRegistration(); + } + } + + @Test(timeOut = 10_000) + public void testLateRecoveryReadIsReleasedAfterClose() throws Exception { + Position startPosition = PositionFactory.create(1, 0); + Position lastPosition = PositionFactory.create(1, 3); + CompletableFuture recoveryFuture = CompletableFuture.completedFuture(startPosition); + CountDownLatch readStarted = new CountDownLatch(1); + AtomicReference 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()); + + assertTrue(readStarted.await(5, TimeUnit.SECONDS)); + context.transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS); + context.awaitExecutorIdle(); + + 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 executor = Executors.newSingleThreadExecutor(); + 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 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); + + 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.getTransactionExecutorProvider()).thenReturn(executorProvider); + when(executorProvider.getExecutor(any(Object.class))).thenReturn(executor); + 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 awaitExecutorIdle() throws Exception { + executor.submit(() -> { }).get(5, TimeUnit.SECONDS); + } + + @Override + public void close() throws Exception { + transactionBuffer.closeAsync().get(5, TimeUnit.SECONDS); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + /** Blocks both continuation APIs so tests can deterministically control the registration race. */ + private static final class ContinuationBlockingRecoveryFuture extends CompletableFuture { + private final CountDownLatch registrationStarted = new CountDownLatch(1); + private final CountDownLatch allowRegistration = new CountDownLatch(1); + + @Override + public CompletableFuture thenAccept(Consumer action) { + awaitRegistration(); + return super.thenAccept(action); + } + + @Override + public CompletableFuture thenAcceptAsync(Consumer action, Executor executor) { + awaitRegistration(); + return super.thenAcceptAsync(action, executor); + } + + private void awaitRegistration() { + registrationStarted.countDown(); + try { + assertTrue(allowRegistration.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + + private void awaitContinuationRegistrationStarted() throws Exception { + assertTrue(registrationStarted.await(5, TimeUnit.SECONDS)); + } + + private void allowContinuationRegistration() { + allowRegistration.countDown(); + } + } +} From b8aae5067a912b867fbd960b69332d57f1f12832 Mon Sep 17 00:00:00 2001 From: Kai Wang Date: Sun, 30 Aug 2026 17:30:37 +0800 Subject: [PATCH 7/7] Cancel queued recovery replay on topic close --- .../buffer/impl/TopicTransactionBuffer.java | 45 +++++++++---- .../impl/TopicTransactionBufferCloseTest.java | 63 ++++++++++++++++++- 2 files changed, 93 insertions(+), 15 deletions(-) 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 0a52fa37d5ade..d2832f24ff3a5 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 @@ -29,7 +29,8 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; @@ -107,7 +108,9 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen private final AbortedTxnProcessor snapshotAbortedTxnProcessor; private final AbortedTxnProcessor.SnapshotType snapshotType; + private final ExecutorService transactionExecutor; private final MaxReadPositionCallBack maxReadPositionCallBack; + private Future recoveryReplayTask; /** if the first snapshot is in progress, it will pending following publishing tasks. **/ private final LinkedList pendingAppendingTxnBufferTasks = new LinkedList<>(); @@ -144,14 +147,14 @@ public TopicTransactionBuffer(PersistentTopic topic) { this.maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry(); this.snapshotAbortedTxnProcessor = snapshotAbortedTxnProcessor; this.snapshotType = snapshotType; + this.transactionExecutor = topic.getBrokerService().getPulsar() + .getTransactionExecutorProvider().getExecutor(this); this.maxReadPositionCallBack = topic.getMaxReadPositionCallBack(); this.recover(); } private void recover() { recoverTime.setRecoverStartTime(System.currentTimeMillis()); - Executor transactionExecutor = this.topic.getBrokerService().getPulsar() - .getTransactionExecutorProvider().getExecutor(this); transactionExecutor.execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() { @Override public void recoverComplete() { @@ -243,7 +246,14 @@ public void recoverExceptionally(Throwable e) { + " transaction buffer throw exception"); topic.close(true); } - }, this.topic, this, snapshotAbortedTxnProcessor, transactionExecutor)); + }, this.topic, this, snapshotAbortedTxnProcessor)); + } + + private synchronized void submitRecoveryReplay(Runnable replay) { + if (checkIfClosed()) { + return; + } + recoveryReplayTask = transactionExecutor.submit(replay); } @Override @@ -703,12 +713,16 @@ public CompletableFuture clearSnapshotAndClose() { @Override public CompletableFuture closeAsync() { boolean closeStarted; - // Serialize closure with recovery entry handling, which uses the same monitor. + // 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 queued recovery before completing futures whose callbacks may run inline. + // Cancel snapshot recovery before completing futures whose callbacks may run inline. CompletableFuture processorCloseFuture = this.snapshotAbortedTxnProcessor.closeAsync(); if (closeStarted) { Throwable closeException = @@ -835,17 +849,14 @@ public static class TopicTransactionBufferRecover implements Runnable { private final AbortedTxnProcessor abortedTxnProcessor; - private final Executor replayExecutor; - private TopicTransactionBufferRecover(TopicTransactionBufferRecoverCallBack callBack, PersistentTopic topic, TopicTransactionBuffer transactionBuffer, - AbortedTxnProcessor abortedTxnProcessor, Executor replayExecutor) { + AbortedTxnProcessor abortedTxnProcessor) { this.topic = topic; this.callBack = callBack; this.entryQueue = new SpscArrayQueue<>(2000); this.topicTransactionBuffer = transactionBuffer; this.abortedTxnProcessor = abortedTxnProcessor; - this.replayExecutor = replayExecutor; } @Override @@ -856,11 +867,21 @@ public void run() { .log("TransactionBuffer of topic can not change state to Initializing"); return; } - // Transaction-buffer replay must not extend the snapshot processor's close barrier. - abortedTxnProcessor.recoverFromSnapshot().thenAcceptAsync(this::replayTransactionBuffer, replayExecutor) + // Keep replay on the transaction executor while retaining a task handle that close can cancel. + abortedTxnProcessor.recoverFromSnapshot().thenAccept(this::submitReplay) .exceptionally(this::handleRecoveryFailure); } + private void submitReplay(Position recoveredPosition) { + topicTransactionBuffer.submitRecoveryReplay(() -> { + try { + replayTransactionBuffer(recoveredPosition); + } catch (Throwable error) { + handleRecoveryFailure(error); + } + }); + } + private Void handleRecoveryFailure(Throwable error) { Throwable cause = FutureUtil.unwrapCompletionException(error); if (!shouldStopRecovery()) { 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 index df7da49c0ef2c..2ab109c28db35 100644 --- 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 @@ -27,13 +27,15 @@ import static org.mockito.Mockito.never; 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 java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; @@ -94,6 +96,38 @@ public void testRecoveryContinuationDoesNotStartAfterClose() throws Exception { } } + @Test(timeOut = 10_000) + public void testCloseCancelsQueuedRecoveryReplay() throws Exception { + CompletableFuture recoveryFuture = new CompletableFuture<>(); + CountDownLatch blockerStarted = new CountDownLatch(1); + CountDownLatch releaseBlocker = new CountDownLatch(1); + try (TestContext context = new TestContext(recoveryFuture, PositionFactory.EARLIEST)) { + try { + context.awaitExecutorIdle(); + context.executor.clearSubmittedTask(); + context.executor.execute(() -> { + blockerStarted.countDown(); + try { + releaseBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + assertTrue(blockerStarted.await(5, TimeUnit.SECONDS)); + + recoveryFuture.complete(PositionFactory.EARLIEST); + Future replayTask = context.executor.submittedTask(); + 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); @@ -122,7 +156,7 @@ public void testLateRecoveryReadIsReleasedAfterClose() throws Exception { } private static final class TestContext implements AutoCloseable { - private final ExecutorService executor = Executors.newSingleThreadExecutor(); + private final TrackingExecutor executor = new TrackingExecutor(); private final AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class); private final PersistentTopic topic = mock(PersistentTopic.class); private final ManagedLedgerImpl managedLedger = mock(ManagedLedgerImpl.class); @@ -163,6 +197,29 @@ public void close() throws Exception { } } + private static final class TrackingExecutor extends ThreadPoolExecutor { + private final AtomicReference> submittedTask = new AtomicReference<>(); + + private TrackingExecutor() { + super(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()); + } + + @Override + public Future submit(Runnable task) { + Future future = super.submit(task); + submittedTask.set(future); + return future; + } + + private Future submittedTask() { + return submittedTask.get(); + } + + private void clearSubmittedTask() { + submittedTask.set(null); + } + } + /** Blocks both continuation APIs so tests can deterministically control the registration race. */ private static final class ContinuationBlockingRecoveryFuture extends CompletableFuture { private final CountDownLatch registrationStarted = new CountDownLatch(1);