You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/09/07 08:15:44 UTC
Re: [PR] [fix][broker] Cancel queued transaction snapshot recovery on topic close [pulsar]
lhotari commented on code in PR #26335:
URL: https://github.com/apache/pulsar/pull/26335#discussion_r3947773218
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.buffer.impl;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Coordinates snapshot recovery and resource closure so resources remain available until processor-owned recovery
+ * work finishes.
+ */
+abstract class AbstractSnapshotAbortedTxnProcessor implements AbortedTxnProcessor {
+
+ private enum State {
+ IDLE,
+ RECOVERY_QUEUED,
+ RECOVERY_RUNNING,
+ CLOSED
+ }
+
+ private final ScheduledExecutorService recoveryExecutor;
+
+ private volatile State state = State.IDLE;
+ private Future<?> recoveryTask;
+ private CompletableFuture<Position> recoveryFuture = CompletableFuture.completedFuture(null);
+ // Completes before the recovery result, so close waits only for processor-owned work.
+ private CompletableFuture<Void> recoveryWorkFinishedFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+ AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) {
+ this.recoveryExecutor = recoveryExecutor;
+ }
+
+ @Override
+ public CompletableFuture<Position> recoverFromSnapshot() {
Review Comment:
**[NIT] recoverFromSnapshot() and closeAsync() are the state machine but are not final**
This method and `closeAsync()` at :143 own the whole `IDLE`/`RECOVERY_QUEUED`/`RECOVERY_RUNNING`/`CLOSED` protocol; the subclass contract is `doRecoverFromSnapshot` and `closeResources`. `isClosed()` is already `final` at :122 — marking these two `final` as well would stop a future subclass from half-overriding the protocol and quietly losing the barrier. Non-blocking.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -697,17 +725,30 @@ public CompletableFuture<Void> clearSnapshotAndClose() {
@Override
public CompletableFuture<Void> closeAsync() {
- synchronized (pendingAppendingTxnBufferTasks) {
- if (!checkIfClosed()) {
+ boolean closeStarted;
+ // Serialize closure with recovery entry handling and replay submission, which use the same monitor.
+ synchronized (this) {
Review Comment:
**[BUG] taking the transaction buffer monitor in closeAsync() inverts a lock order the buffer already relies on, so a recovery replay racing a fenced publish deadlocks the snapshot-recover thread against a managed-ledger callback thread**
The buffer already acquires the `PersistentTopic` monitor **while holding its own**. `handleTxnEntry` takes `synchronized (TopicTransactionBuffer.this)` at :208 and, for a transaction marker, calls `removeTxnAndUpdateMaxReadPosition` at :216 — which ends in `updateLastDispatchablePosition(null)`:
https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L641-L651
That delegates to `topic.updateLastDispatchablePosition(...)` ([`TopicTransactionBuffer.java:473-475`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L473-L475)), which is `synchronized` on the topic — [`PersistentTopic.java:4793`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L4793). So the existing order is **buffer monitor → topic monitor**, and it is exercised on all three callers of `removeTxnAndUpdateMaxReadPosition` ([`:216`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L216), [`:497`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broke
r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L497), [`:547`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L547)). It is also why `syncMaxReadPositionForNormalPublish` deliberately calls `updateLastDispatchablePosition` *after* leaving the monitor at [`:783-786`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L783-L786).
This line adds the opposite order. `PersistentTopic.addFailed` is `synchronized` ([`PersistentTopic.java:793`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L793)) and, on `ManagedLedgerFencedException`, calls `close()` inline while still holding the topic monitor ([`PersistentTopic.java:809-811`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L809-L811)). `close()` → `close(true, false)` reaches [`PersistentTopic.java:1852`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L1852) synchronously, on that same thread, before anything async — and `transactionBuffer.closeAsync()` now blocks on the buffer monitor.
That is **topic monitor → buffer monitor**.
On the base branch `closeAsync()` synchronized on `pendingAppendingTxnBufferTasks` and never on `this`, so this edge is new here.
**Failure scenario:** a transactional topic is replaying its buffer while producers are publishing. The managed ledger is fenced (ownership change, or a BK-side fence during the repeated reload this PR is about).
- Thread A, `pulsar-transaction-snapshot-recover-…`: inside `handleTxnEntry` for a commit/abort marker, holds the buffer monitor, calls `topic.updateLastDispatchablePosition(null)`, blocks on the topic monitor.
- Thread B, the managed ledger's callback thread: inside `addFailed`, holds the topic monitor, calls `close()` → `transactionBuffer.closeAsync()`, blocks on the buffer monitor.
Neither can proceed. Both are shared ordered-executor threads — the recover thread is chosen per namespace (`chooseThread(namespace)` at :255-256), so every transaction buffer recovery hashing to it stalls too, and the managed-ledger callback thread serves many ledgers. Close cannot make progress either, so nothing reaches the state change that would stop the replay.
The window is narrow, but it is a genuine lock cycle rather than a timing window that eventually resolves.
The cheapest way out looks like removing the buffer→topic edge instead of the new one: hoist `updateLastDispatchablePosition(null)` out of `removeTxnAndUpdateMaxReadPosition` and have each caller invoke it after leaving the monitor — exactly the shape `syncMaxReadPositionForNormalPublish` already uses. That leaves the serialisation this line needs intact. (The `// ThreadSafe` comment at [`:472`](https://github.com/apache/pulsar/blob/29d214151988ec74357996c348624df8f3ae45e4/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java#L472) is worth revisiting either way.)
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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<Position> recoveryFuture = CompletableFuture.completedFuture(null);
+ private CompletableFuture<Void> recoveryStoppedFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+ protected AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) {
+ this.recoveryExecutor = recoveryExecutor;
+ }
+
+ @Override
+ public final synchronized CompletableFuture<Position> 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<Position> future = new CompletableFuture<>();
+ CompletableFuture<Void> 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);
+ }
+ // Do not let callers complete the internal recovery result.
+ return future.copy();
+ }
+
+ private void runRecovery(CompletableFuture<Position> future, CompletableFuture<Void> stoppedFuture) {
+ 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());
+ } finally {
+ stoppedFuture.complete(null);
+ }
+ }
+
+ /**
+ * 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<Void> closeAsync() {
+ boolean recoveryOwnsFutureCompletion;
+ CompletableFuture<Position> currentRecoveryFuture;
+ CompletableFuture<Void> currentRecoveryStoppedFuture;
+ 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;
+ currentRecoveryStoppedFuture = this.recoveryStoppedFuture;
+ if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) {
+ this.recoveryTask.cancel(false);
+ }
+ }
+ if (!recoveryOwnsFutureCompletion) {
+ currentRecoveryFuture.completeExceptionally(closedException());
Review Comment:
Addressed, and in two independent places, which is the right belt-and-braces here.
`failRecoveryAfterClose` defers the close-induced failure until `closeFuture` completes rather than publishing it inline during `closeAsync()`, and `recoverExceptionally` returns before `topic.close(true)` once the buffer is closed. Since `TopicTransactionBuffer.closeAsync()` sets the close state at `:730-737` before calling `snapshotAbortedTxnProcessor.closeAsync()` at `:739`, and that is the only production call site of `AbortedTxnProcessor.closeAsync()`, the guard cannot be bypassed on that path. `testCloseInducedRecoveryFailureDoesNotCloseTopicAgain` pins it directly, and `testCloseCancelsQueuedRecovery` pins the deferral.
The transfer-specific second-order effect I described — the orphaned `closeFutures.transferring` — goes away with the re-entrancy itself.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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<Position> recoveryFuture = CompletableFuture.completedFuture(null);
+ private CompletableFuture<Void> recoveryStoppedFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+ protected AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) {
+ this.recoveryExecutor = recoveryExecutor;
+ }
+
+ @Override
+ public final synchronized CompletableFuture<Position> 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<Position> future = new CompletableFuture<>();
+ CompletableFuture<Void> 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);
+ }
+ // Do not let callers complete the internal recovery result.
+ return future.copy();
+ }
+
+ private void runRecovery(CompletableFuture<Position> future, CompletableFuture<Void> stoppedFuture) {
+ 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());
+ } finally {
+ stoppedFuture.complete(null);
+ }
+ }
+
+ /**
+ * 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<Void> closeAsync() {
+ boolean recoveryOwnsFutureCompletion;
+ CompletableFuture<Position> currentRecoveryFuture;
+ CompletableFuture<Void> currentRecoveryStoppedFuture;
+ 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;
+ currentRecoveryStoppedFuture = this.recoveryStoppedFuture;
+ if (this.recoveryTask != null && previousState == State.RECOVERY_QUEUED) {
+ this.recoveryTask.cancel(false);
+ }
+ }
+ if (!recoveryOwnsFutureCompletion) {
+ currentRecoveryFuture.completeExceptionally(closedException());
+ currentRecoveryStoppedFuture.complete(null);
+ }
+ currentRecoveryStoppedFuture.thenCompose(v -> closeResources())
Review Comment:
Addressed, and by the route I was hoping for rather than the one I asked about.
The barrier is now narrowed to processor-owned work — `recoveryWorkFinished` completes before the recovery result, and `closeResources()` hangs off it alone — and the replay is a separate cancellable task that re-checks `shouldStopRecovery()` on every iteration. So close no longer chains behind the replay, and a replay stuck reissuing a transiently-failing read is cut short by close rather than holding it open. `testCloseDoesNotWaitForRecoveryCallback` inverts the old assertion, and `testCloseCancelsQueuedRecoveryReplay` pins the cancellation.
What close does still wait for is the snapshot read itself, which is what the description says and is bounded by `getSystemClientOperationTimeoutMs()` in the segment processor's `wait()` helper. That is a reasonable trade.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/AbstractSnapshotAbortedTxnProcessor.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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<Position> recoveryFuture = CompletableFuture.completedFuture(null);
+ private CompletableFuture<Void> recoveryStoppedFuture = CompletableFuture.completedFuture(null);
+ private final CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+ protected AbstractSnapshotAbortedTxnProcessor(ScheduledExecutorService recoveryExecutor) {
+ this.recoveryExecutor = recoveryExecutor;
+ }
+
+ @Override
+ public final synchronized CompletableFuture<Position> 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<Position> future = new CompletableFuture<>();
+ CompletableFuture<Void> 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);
+ }
+ // Do not let callers complete the internal recovery result.
+ return future.copy();
+ }
+
+ private void runRecovery(CompletableFuture<Position> future, CompletableFuture<Void> stoppedFuture) {
+ 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());
+ } finally {
+ stoppedFuture.complete(null);
Review Comment:
Addressed structurally, which is better than the fix I suggested.
The barrier no longer tries to cover the caller's continuation — `recoverFromSnapshot()` returns a copy and the replay is submitted through `submitRecoveryReplay`, which is synchronized on the buffer and refuses once it is closed. Since `closeAsync()` marks the buffer closed under that same monitor before releasing any resource, a continuation that attaches late finds a closed buffer instead of starting a replay behind close, and `handleTxnEntry` re-checks the same state under the same monitor. When the continuation attaches no longer matters.
`testRecoveryContinuationDoesNotStartAfterClose` pins it by completing the recovery future after close and asserting no cursor is ever opened.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org