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..409696d6953f1 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java @@ -0,0 +1,200 @@ +/* + * 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 processor-owned recovery + * work finishes. + */ +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); + // 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<>(); + + AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) { + this.recoveryExecutor = recoveryExecutor; + } + + @Override + 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; + } + } + if (submissionFailure != null) { + newRecoveryWorkFinishedFuture.complete(null); + newRecoveryFuture.completeExceptionally(submissionFailure); + } + // Do not let callers complete the internal recovery result. + return newRecoveryFuture.copy(); + } + + private void runRecovery(CompletableFuture recoveryResult, + CompletableFuture recoveryWorkFinished) { + Position recoveredPosition = null; + Throwable recoveryFailure = null; + boolean closeWon = false; + try { + if (!tryStartRecovery()) { + closeWon = true; + } else { + recoveredPosition = doRecoverFromSnapshot(recoveryExecutor); + if (!tryMarkRecoveryFinished()) { + closeWon = true; + } + } + } catch (Throwable throwable) { + if (tryMarkRecoveryFinished()) { + recoveryFailure = throwable; + } else { + closeWon = true; + } + } + 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 + */ + abstract Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception; + + final boolean isClosed() { + return this.state == State.CLOSED; + } + + private synchronized boolean tryStartRecovery() { + if (this.state != State.RECOVERY_QUEUED) { + return false; + } + this.state = State.RECOVERY_RUNNING; + return true; + } + + private synchronized boolean tryMarkRecoveryFinished() { + if (this.state != State.RECOVERY_RUNNING) { + return false; + } + this.state = State.RECOVERY_FINISHED; + return true; + } + + @Override + public CompletableFuture closeAsync() { + State previousState; + CompletableFuture currentRecoveryFuture; + CompletableFuture currentRecoveryWorkFinishedFuture; + synchronized (this) { + if (this.state == State.CLOSED) { + return closeFuture; + } + previousState = this.state; + this.state = State.CLOSED; + currentRecoveryFuture = this.recoveryFuture; + currentRecoveryWorkFinishedFuture = this.recoveryWorkFinishedFuture; + if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) { + this.recoveryTask.cancel(false); + } + } + currentRecoveryWorkFinishedFuture.thenCompose(v -> closeResources()) + .whenComplete((v, throwable) -> { + if (throwable != null) { + closeFuture.completeExceptionally(throwable); + } else { + 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 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. + */ + 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..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 @@ -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(); + 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(); - } + 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..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 @@ -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,7 +70,7 @@ 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; @@ -140,10 +139,14 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess *

Clear all snapshot segment.

*/ private final PersistentWorker persistentWorker; + // 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-"; 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 +234,48 @@ public CompletableFuture takeAbortedTxnsSnapshot(Position maxReadPosition) } @Override - public CompletableFuture recoverFromSnapshot() { + 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 { + if (isClosed()) { + return; + } final var managedLedger = openReadOnlyManagedLedger(topicName); boolean hasInvalidIndex = false; for (var index : indexes.getIndexList()) { + if (isClosed()) { + return; + } final var position = PositionFactory.create(index.getSegmentLedgerID(), index.getSegmentEntryID()); final var abortedPosition = PositionFactory.create(index.abortedMarkLedgerID, index.abortedMarkEntryID); try { @@ -297,10 +300,11 @@ private void readSegmentEntries(TopicName topicName, TransactionBufferSnapshotIn } } } - if (hasInvalidIndex) { + if (hasInvalidIndex && !isClosed()) { // Update the snapshot segment index if there exist invalid indexes. - persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, - () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot())); + recoveryIndexUpdatesFuture = CompletableFuture.allOf(recoveryIndexUpdatesFuture, + persistentWorker.appendTask(PersistentWorker.OperationType.UpdateIndex, + () -> persistentWorker.updateSnapshotIndex(indexes.getSnapshot()))); } } @@ -350,22 +354,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,8 +419,9 @@ public TransactionBufferStats generateSnapshotStats(boolean segmentStats) { } @Override - public CompletableFuture closeAsync() { - return persistentWorker.closeAsync(); + CompletableFuture closeResources() { + return recoveryIndexUpdatesFuture.handle((__, throwable) -> null) + .thenCompose(__ -> persistentWorker.closeAsync()); } private void handleSnapshotSegmentEntry(Entry entry) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java index c51e8e57a7d39..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 @@ -27,12 +27,14 @@ 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.ExecutorService; +import java.util.concurrent.Future; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; -import lombok.SneakyThrows; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -106,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<>(); @@ -143,17 +147,21 @@ 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()); - this.topic.getBrokerService().getPulsar().getTransactionExecutorProvider().getExecutor(this) - .execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() { + transactionExecutor.execute(new TopicTransactionBufferRecover(new TopicTransactionBufferRecoverCallBack() { @Override public void recoverComplete() { synchronized (TopicTransactionBuffer.this) { + if (checkIfClosed()) { + return; + } if (ongoingTxns.isEmpty()) { updateMaxReadPositionAfterRecovery(); } @@ -175,6 +183,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 +208,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 +225,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,11 +237,25 @@ 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)); } + private synchronized void submitRecoveryReplay(Runnable replay) { + if (checkIfClosed()) { + return; + } + recoveryReplayTask = transactionExecutor.submit(replay); + } + @Override public CompletableFuture getTransactionMeta(TxnID txnID) { return CompletableFuture.completedFuture(null); @@ -314,6 +337,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 +712,30 @@ public CompletableFuture clearSnapshotAndClose() { @Override public CompletableFuture closeAsync() { - synchronized (pendingAppendingTxnBufferTasks) { - if (!checkIfClosed()) { + boolean closeStarted; + // Serialize closure with recovery entry handling and replay submission, which use the same monitor. + synchronized (this) { + closeStarted = !checkIfClosed(); + changeToCloseState(); + if (recoveryReplayTask != null) { + recoveryReplayTask.cancel(false); + recoveryReplayTask = null; + } + } + // Cancel snapshot recovery before completing futures whose callbacks may run inline. + CompletableFuture 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 +839,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(); @@ -820,7 +859,6 @@ private TopicTransactionBufferRecover(TopicTransactionBufferRecoverCallBack call this.abortedTxnProcessor = abortedTxnProcessor; } - @SneakyThrows @Override public void run() { if (!this.topicTransactionBuffer.changeToInitializingState()) { @@ -829,32 +867,66 @@ 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; + // 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 { - managedCursor = topic.getManagedLedger() - .newNonDurableCursor(this.startReadCursorPosition, SUBSCRIPTION_NAME); - } catch (ManagedLedgerException e) { + replayTransactionBuffer(recoveredPosition); + } catch (Throwable error) { + handleRecoveryFailure(error); + } + }); + } + + private Void handleRecoveryFailure(Throwable error) { + Throwable cause = FutureUtil.unwrapCompletionException(error); + if (!shouldStopRecovery()) { + 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 +941,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 +1010,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 +1021,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 +1040,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 +1060,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/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); 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..0ed02607a5e52 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessorTest.java @@ -0,0 +1,364 @@ +/* + * 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.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.bookkeeper.mledger.Position; +import org.apache.pulsar.broker.service.BrokerServiceException; +import org.apache.pulsar.client.api.transaction.TxnID; +import org.apache.pulsar.common.policies.data.TransactionBufferStats; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class AbstractSnapshotAbortedTxnProcessorTest { + + @Test(timeOut = 10_000) + public void testCloseCancelsQueuedRecovery() throws Exception { + try (RecoveryTestContext context = RecoveryTestContext.queued()) { + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); + CompletableFuture callbackHeldProcessorLock = + trackCallbackLock(recoveryFuture, context.processor); + CompletableFuture closeCompletedBeforeRecovery = recoveryFuture.handle( + (__, ___) -> context.processor.closeAsync().isDone()); + CompletableFuture repeatedRecoveryFuture = context.processor.recoverFromSnapshot(); + context.awaitRecoveryQueued(); + + context.processor.closeAsync().get(5, TimeUnit.SECONDS); + + assertTrue(context.submittedRecoveryTask().isCancelled()); + context.verifyRecoveryFailedAfterClose(recoveryFuture); + context.verifyRecoveryFailedAfterClose(repeatedRecoveryFuture); + assertFalse(context.processor.recoveryStarted()); + assertTrue(context.processor.resourcesClosed()); + assertFalse(callbackHeldProcessorLock.get(1, TimeUnit.SECONDS), + "Recovery callbacks must run without holding the processor lock"); + assertTrue(closeCompletedBeforeRecovery.get(1, TimeUnit.SECONDS), + "Close must finish before publishing a close-induced recovery failure"); + } + } + + @Test(timeOut = 10_000) + public void testCloseWaitsForRunningRecovery() throws Exception { + try (RecoveryTestContext context = RecoveryTestContext.running()) { + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); + context.awaitRecoveryStarted(); + + Future recoveryTask = context.submittedRecoveryTask(); + CompletableFuture closeFuture = context.processor.closeAsync(); + + assertFalse(recoveryTask.isCancelled()); + assertFalse(recoveryFuture.isDone(), "Running recovery must not complete until its task stops"); + assertFalse(closeFuture.isDone(), "Closing must wait for running recovery before closing resources"); + assertFalse(context.processor.resourcesClosed()); + + context.finishRecovery(); + context.verifyRecoveryFailedAfterClose(recoveryFuture); + closeFuture.get(5, TimeUnit.SECONDS); + assertTrue(context.processor.resourcesClosed()); + } + } + + @Test(timeOut = 10_000) + public void testCloseDoesNotWaitForRecoveryCallback() throws Exception { + CountDownLatch callbackStarted = new CountDownLatch(1); + CountDownLatch finishCallback = new CountDownLatch(1); + try (RecoveryTestContext context = RecoveryTestContext.running()) { + CompletableFuture recoveryFuture = context.processor.recoverFromSnapshot(); + CompletableFuture 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(); + + closeFuture.get(5, TimeUnit.SECONDS); + assertTrue(context.processor.resourcesClosed()); + + finishCallback.countDown(); + callbackFuture.get(5, TimeUnit.SECONDS); + } finally { + finishCallback.countDown(); + } + } + + @Test(timeOut = 10_000) + public void testCloseAfterRecoveryCompleted() throws Exception { + try (RecoveryTestContext context = RecoveryTestContext.running()) { + 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.processor.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.processor.recoverFromSnapshot(); + context.awaitRecoveryStarted(); + + context.finishRecovery(); + context.verifyRecoveryFailed(recoveryFuture, failure); + context.verifyRecoverySucceeded(context.processor.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 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; + + static RecoveryTestContext queued() throws Exception { + return new RecoveryTestContext(true); + } + + static RecoveryTestContext running() throws Exception { + return new RecoveryTestContext(false); + } + + private RecoveryTestContext(boolean queueRecovery) throws Exception { + if (queueRecovery) { + CountDownLatch blockerStarted = new CountDownLatch(1); + recoveryExecutor.execute(() -> { + blockerStarted.countDown(); + try { + releaseBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + assertTrue(blockerStarted.await(5, TimeUnit.SECONDS)); + } + + processor = new TestSnapshotProcessor(recoveryExecutor); + } + + void awaitRecoveryQueued() { + assertEquals(recoveryExecutor.getQueue().size(), 1); + assertFalse(submittedRecoveryTask().isCancelled()); + } + + void awaitRecoveryStarted() throws InterruptedException { + assertTrue(processor.awaitRecoveryStarted()); + } + + Future submittedRecoveryTask() { + Future recoveryTask = recoveryExecutor.submittedTask(); + assertNotNull(recoveryTask); + return recoveryTask; + } + + void finishRecovery() { + processor.finishRecovery(); + } + + void verifyRecoveryFailedAfterClose(CompletableFuture recoveryFuture) throws Exception { + ExecutionException exception = expectThrows(ExecutionException.class, + () -> recoveryFuture.get(5, TimeUnit.SECONDS)); + assertTrue(exception.getCause() instanceof BrokerServiceException.ServiceUnitNotReadyException, + "Closing the processor must fail the recovery future"); + } + + void verifyRecoverySucceeded(CompletableFuture recoveryFuture) throws Exception { + assertNull(recoveryFuture.get(5, TimeUnit.SECONDS)); + } + + void verifyRecoveryFailed(CompletableFuture recoveryFuture, Throwable expected) { + ExecutionException exception = expectThrows(ExecutionException.class, + () -> recoveryFuture.get(5, TimeUnit.SECONDS)); + assertSame(exception.getCause(), expected); + } + + @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 final AtomicInteger recoveryAttempts = new AtomicInteger(); + + private TestSnapshotProcessor(ScheduledExecutorService recoveryExecutor) { + super(recoveryExecutor); + } + + @Override + Position doRecoverFromSnapshot(ScheduledExecutorService executor) throws Exception { + recoveryAttempts.incrementAndGet(); + recoveryStarted.countDown(); + assertTrue(finishRecovery.await(5, TimeUnit.SECONDS)); + RuntimeException failure = nextRecoveryFailure.getAndSet(null); + if (failure != null) { + throw failure; + } + return null; + } + + private int recoveryAttempts() { + return recoveryAttempts.get(); + } + + @Override + 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(); + } + } +} 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..5f2196a2f13de --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorCloseTest.java @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.transaction.buffer.impl; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import com.google.common.util.concurrent.ListeningScheduledExecutorService; +import com.google.common.util.concurrent.MoreExecutors; +import java.util.Collections; +import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.common.util.OrderedScheduler; +import org.apache.bookkeeper.mledger.AsyncCallbacks; +import org.apache.bookkeeper.mledger.ManagedLedgerConfig; +import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.ManagedLedgerFactory; +import org.apache.bookkeeper.mledger.ReadOnlyManagedLedger; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService; +import org.apache.pulsar.broker.service.SystemTopicTxnBufferSnapshotService.ReferenceCountedWriter; +import org.apache.pulsar.broker.service.TransactionBufferSnapshotServiceFactory; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.broker.systopic.SystemTopicClient; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndex; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexes; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotIndexesMetadata; +import org.apache.pulsar.broker.transaction.buffer.metadata.v2.TransactionBufferSnapshotSegment; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.client.impl.conf.ClientConfigurationData; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class SnapshotSegmentAbortedTxnProcessorCloseTest { + + @Test(timeOut = 10_000) + @SuppressWarnings("unchecked") + public void testCloseWaitsForRecoveryIndexUpdate() throws Exception { + ListeningScheduledExecutorService recoveryExecutor = + MoreExecutors.listeningDecorator(Executors.newSingleThreadScheduledExecutor()); + CompletableFuture indexUpdateFuture = new CompletableFuture<>(); + CountDownLatch indexUpdateStarted = new CountDownLatch(1); + SnapshotSegmentAbortedTxnProcessorImpl processor = null; + try { + String topicName = "persistent://public/default/test-close-during-index-update"; + PersistentTopic topic = mock(PersistentTopic.class); + BrokerService brokerService = mock(BrokerService.class); + PulsarService pulsar = mock(PulsarService.class); + ServiceConfiguration configuration = mock(ServiceConfiguration.class); + PulsarClientImpl client = mock(PulsarClientImpl.class); + ClientConfigurationData clientConfiguration = new ClientConfigurationData(); + clientConfiguration.setOperationTimeoutMs(5_000); + OrderedScheduler recoveryScheduler = mock(OrderedScheduler.class); + TransactionBufferSnapshotServiceFactory serviceFactory = + mock(TransactionBufferSnapshotServiceFactory.class); + + when(topic.getName()).thenReturn(topicName); + when(topic.getBrokerService()).thenReturn(brokerService); + when(brokerService.getPulsar()).thenReturn(pulsar); + when(pulsar.getConfiguration()).thenReturn(configuration); + when(pulsar.getClient()).thenReturn(client); + when(client.getConfiguration()).thenReturn(clientConfiguration); + when(configuration.getTransactionBufferSnapshotSegmentSize()).thenReturn(1024); + when(pulsar.getTransactionSnapshotRecoverExecutorProvider()).thenReturn(recoveryScheduler); + when(recoveryScheduler.chooseThread(any(Object.class))).thenReturn(recoveryExecutor); + when(pulsar.getTransactionBufferSnapshotServiceFactory()).thenReturn(serviceFactory); + + SystemTopicTxnBufferSnapshotService segmentService = + mock(SystemTopicTxnBufferSnapshotService.class); + SystemTopicTxnBufferSnapshotService indexService = + mock(SystemTopicTxnBufferSnapshotService.class); + ReferenceCountedWriter segmentWriter = + mock(ReferenceCountedWriter.class); + ReferenceCountedWriter indexWriter = + mock(ReferenceCountedWriter.class); + SystemTopicClient.Writer segmentSystemWriter = + mock(SystemTopicClient.Writer.class); + 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); + 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)); + // A later recovery must not replace the in-flight update in the close barrier. + processor.recoverFromSnapshot().get(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)); + } + } +} 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..2ab109c28db35 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferCloseTest.java @@ -0,0 +1,258 @@ +/* + * 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.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.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; +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 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); + 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 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); + 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)); + } + } + + 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); + 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(); + } + } +}