From 8f682f0032170e723390e39f784db379326cbaee Mon Sep 17 00:00:00 2001 From: Alexandre Burgoni Date: Fri, 10 Jul 2026 17:07:40 +0200 Subject: [PATCH 1/2] [fix][broker] Prevent stale read completions from stranding Failover subscriptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On a single-active-consumer (Failover/Exclusive) subscription, internalRedeliverUnacknowledgedMessages clears havePendingRead and re-arms a fresh tail-wait read without cancelling a read that is already in flight (an in-flight read cannot be cancelled). When the disowned read later completes, readEntriesComplete/readEntriesFailed unconditionally clear havePendingRead again while the fresh op stays armed on the cursor: the dispatcher now believes no read is pending while the cursor holds an armed waiting read. The desync is benign while the cursor remains in waitingCursors, but on the next last-consumer disconnect cancelPendingRead() short-circuits on the false flag (the armed op survives) while PersistentSubscription#removeConsumer removes the cursor from waitingCursors. From then on the subscription is permanently stuck: every re-arm CAS-fails on the leftover op (ConcurrentWaitCallbackException is returned without rescheduling) and every publish's notifyCursors() misses the cursor. Only a topic unload recovers it. The cursor side is already generation-guarded ((op, opReadId)); the dispatcher's boolean mirror is not — that asymmetry is the defect. Fixes #26164 Signed-off-by: Alexandre Burgoni --- ...sistentDispatcherSingleActiveConsumer.java | 37 +- .../broker/service/PersistentTopicTest.java | 2 +- ...cherSingleActiveConsumerStuckReadTest.java | 583 ++++++++++++++++++ ...entDispatcherSingleActiveConsumerTest.java | 2 +- 4 files changed, 618 insertions(+), 6 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java index 08356f378335d..b08860d7e2000 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java @@ -74,6 +74,11 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher private Optional dispatchRateLimiter = Optional.empty(); protected volatile boolean havePendingRead = false; + // Monotonic identity of the read that havePendingRead currently refers to. Bumped under this monitor + // whenever readMoreEntries issues a read; a completion mutates dispatcher state only when its captured + // epoch still matches, so a completion whose read was disowned by a redeliver-driven rewind + re-arm + // is ignored instead of clearing havePendingRead for the newer outstanding read. Guarded by "this". + private long readOpEpoch = 0L; protected volatile int readBatchSize; protected final Backoff readFailureBackoff; @@ -159,7 +164,20 @@ protected void cancelPendingRead() { } } - private synchronized void readEntriesComplete(List entries, Consumer readConsumer, long epoch) { + private synchronized void readEntriesComplete(List entries, Consumer readConsumer, long epoch, + long readOpEpoch) { + if (readOpEpoch != this.readOpEpoch) { + // Stale completion: this read was disowned by a redeliver-driven cursor rewind + re-arm + // (internalRedeliverUnacknowledgedMessages) that already issued a newer read. Clearing + // havePendingRead or dispatching here would strand the newer armed read and double-deliver; + // release the entries (the newer read re-delivers from the rewound position) and do nothing else. + log.debug() + .attr("readConsumer", readConsumer) + .attr("size", entries.size()) + .log("Discarding stale read completion"); + entries.forEach(Entry::release); + return; + } log.debug() .attr("readConsumer", readConsumer) .attr("size", entries.size()) @@ -363,6 +381,9 @@ void readMoreEntries(Consumer consumer) { .attr("messagesToRead", messagesToRead) .log("Schedule read of messages"); havePendingRead = true; + // Tag this read so a completion that has since been superseded by a redeliver-driven re-arm + // can be detected as stale (see readEntriesComplete / readEntriesFailed). + final long readOpEpoch = ++this.readOpEpoch; // TODO: should we pass the consumer epoch for compacted read path? See // https://github.com/apache/pulsar/issues/13690 final var epoch = consumer.readCompacted() ? DEFAULT_CONSUMER_EPOCH : consumer.getConsumerEpoch(); @@ -379,9 +400,9 @@ void readMoreEntries(Consumer consumer) { } entriesFuture.whenCompleteAsync((entries, e) -> { if (e == null) { - readEntriesComplete(entries, consumer, epoch); + readEntriesComplete(entries, consumer, epoch, readOpEpoch); } else { - readEntriesFailed(e, consumer); + readEntriesFailed(e, consumer, readOpEpoch); } }, executor); } @@ -452,7 +473,15 @@ protected Pair calculateToRead(Consumer consumer) { } @VisibleForTesting - public synchronized void readEntriesFailed(Throwable throwable, Consumer consumer) { + public synchronized void readEntriesFailed(Throwable throwable, Consumer consumer, long readOpEpoch) { + if (readOpEpoch != this.readOpEpoch) { + // Stale failure for a read already disowned by a redeliver-driven re-arm; clearing havePendingRead + // or rescheduling here would disturb the newer outstanding read. Ignore it. + log.debug() + .attr("consumer", consumer) + .log("Ignoring stale read failure"); + return; + } havePendingRead = false; final var exception = FutureUtil.unwrapCompletionException(throwable); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java index a4e9958b7b794..9a447a3282f86 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java @@ -385,7 +385,7 @@ public void testDispatcherSingleConsumerReadFailed() { PersistentDispatcherSingleActiveConsumer dispatcher = new PersistentDispatcherSingleActiveConsumer(cursor, SubType.Exclusive, 1, topic, null); Consumer consumer = mock(Consumer.class); - dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), consumer); + dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), consumer, 0L); verify(topic, atLeast(1)).getBrokerService(); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java new file mode 100644 index 0000000000000..48dd478229ff5 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java @@ -0,0 +1,583 @@ +/* + * 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.service.persistent; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import lombok.CustomLog; +import org.apache.bookkeeper.common.util.OrderedExecutor; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedLedgerConfig; +import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.test.MockedBookKeeperTestCase; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.testcontext.PulsarTestContext; +import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; +import org.awaitility.Awaitility; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.Test; + +/** + * Regression test for a single-active-consumer (Failover) subscription that becomes permanently stuck + * when a client redeliver command races the completion of an already-dispatched managed-ledger read. + * + *

Four coupled state variables. A caught-up Failover subscription is described by four pieces + * of state that must stay consistent: + *

    + *
  1. {@code cursor.waitingReadOp} — the armed tail-wait read op + * ({@link ManagedCursorImpl#hasPendingReadRequest()});
    2. + *
  2. the cursor's membership in {@link ManagedLedgerImpl}'s {@code waitingCursors} queue — the + * set a publish walks to wake parked readers;
    3. + *
  3. {@code cursor.pendingReadOps} — reads currently in flight + * ({@link ManagedCursorImpl#getPendingReadOpsCount()});
    4. + *
  4. {@code dispatcher.havePendingRead} — the dispatcher's belief that a read is outstanding.
    5. + *
+ * In healthy operation an armed wait op implies {@code havePendingRead}, and implies the cursor is + * either queued in {@code waitingCursors} or has an in-flight read. + * + *

The race. {@code internalRedeliverUnacknowledgedMessages} + * (PersistentDispatcherSingleActiveConsumer:275-306) unconditionally clears {@code havePendingRead} + * (:300) and arms a fresh read via {@code readMoreEntries} (:305) without draining a read that + * {@code notifyEntriesAvailable} has already dispatched. The stale read's completion + * ({@code readEntriesComplete} :168 / {@code readEntriesFailed} :456) then clears {@code havePendingRead} + * again: there is no read-generation guard on {@code havePendingRead}, and the only staleness check is + * the consumer-identity comparison at :200, which runs after the flag was already cleared at :168. + * When the redeliver's re-arm and the stale completion interleave in the order [redeliver, then stale + * completion], the result is the seed state — an armed wait op with {@code havePendingRead == false}. + * This is still benign while the cursor remains in {@code waitingCursors}. + * + *

The absorbing conversion. When the last consumer disconnects, the durable-cursor path + * ({@code AbstractDispatcherSingleActiveConsumer.removeConsumer} → {@code cancelPendingRead} :157, + * then {@code ManagedLedger.removeWaitingCursor}) deliberately does not cancel the pending read for a + * durable cursor — see the comment at PersistentSubscription:381-388. Because {@code havePendingRead} + * is already {@code false}, {@code cancelPendingRead} short-circuits and leaves the wait op armed, while + * {@code removeWaitingCursor} strips the cursor from the queue. The subscription is now in the absorbing + * orphan state: an armed wait op, absent from {@code waitingCursors}, with no in-flight read and + * {@code havePendingRead == false}. Every subsequent re-arm CAS-fails at ManagedCursorImpl:1110 with + * {@code ConcurrentWaitCallbackException} (which {@code readEntriesFailed} :465-468 does not reschedule), + * and every publish's {@code notifyCursors} poll-misses. The subscription never reads again. + * + *

The two racing events genuinely arrive on different threads: the read completion is posted to the + * dispatcher's ordered executor by the managed-ledger/BookKeeper completion chain + * ({@code whenCompleteAsync}, PersistentDispatcherSingleActiveConsumer:380-386), while the redeliver is + * posted by the client-command thread (:272). Their arrival order is a real race. + * + *

Fidelity notes / test seams. + *

    + *
  • The cursor, ledger, dispatcher, {@code waitingReadOp} compare-and-set, {@code checkForNewEntries}, + * {@code notifyEntriesAvailable}, {@code notifyCursors}, {@code cancelPendingReadRequest}, + * {@code addWaitingCursor}/{@code removeWaitingCursor} and {@code havePendingRead} management are all + * real production code.
    • + *
  • The dispatcher's message-delivery tail ({@code filterEntriesForConsumer} + + * {@code dispatchEntriesToConsumer}) is stubbed on a Mockito spy to release the entries and post the + * same {@code readMoreEntries} continuation the real code posts at {@code dispatchEntriesToConsumer} + * :239. It touches none of the four state variables above.
    • + *
  • The dispatcher's ordered executor is replaced with a {@link ManualExecutor} (injected through a + * spied {@code getTopicOrderedExecutor()} scoped to dispatcher construction) so every dispatcher + * async hop becomes an explicit task the test drains in a chosen order.
    • + *
  • The consumer's ack of {@code m1} is telescoped: it is applied before the dispatcher delivery of + * {@code m1}'s in-flight read completes. This is what keeps the schedule deterministic — after + * the ack the redeliver's {@code rewind} lands at the tail with no backlog, so the re-arm is a pure + * in-memory tail-wait and no BookKeeper read escapes the {@link ManualExecutor} to race the drain. + * Without the ack the {@code rewind} re-exposes the unacked entry and {@code readMoreEntries} takes + * the immediate-read branch (ManagedCursorImpl:1099-1103), issuing a real async read whose completion + * is off-board. The staged state (mark-delete at the tail plus an already-dispatched read whose + * completion is still pending) is production-reachable via a redeliver→ack→redeliver + * sequence.
    • + *
  • {@code newEntriesCheckDelayInMillis} is pinned to {@code 0} so {@code checkForNewEntries} runs + * inline at arm time rather than on the ledger scheduler. This is a determinism pin only; the bug + * itself is not an artifact of the zero delay. With the telescoped ack in place, every + * dispatcher-visible async hop is then an explicit task on the {@link ManualExecutor} board that the + * test drains in a chosen order.
    • + *
+ * + *

Reproduces apache/pulsar#26164. + */ +@CustomLog +public class PersistentDispatcherSingleActiveConsumerStuckReadTest extends MockedBookKeeperTestCase { + + private static final String TOPIC = "persistent://prop/ns/sac-stuck-read"; + private static final int BOARD_WAIT_SECONDS = 5; + private static final long QUIESCE_TIMEOUT_SECONDS = 10; + private static final int QUIESCE_IDLE_ROUNDS = 4; + + private PulsarTestContext pulsarTestContext; + private BrokerService brokerService; + + // Real managed-ledger objects under test. + private ManagedLedgerImpl ledger; + private ManagedCursorImpl cursor; + + // Real dispatcher (Mockito spy so we can stub the delivery tail + pin the active consumer). + private PersistentTopic topic; + private PersistentSubscription subscription; + private PersistentDispatcherSingleActiveConsumer dispatcher; + private Consumer consumer; + + // The dispatcher's ordered executor: a manual task board we drain in a chosen order. + private ManualExecutor dispatcherExecutor; + + private final AtomicLong msgCounter = new AtomicLong(); + + @Override + protected ManagedLedgerConfig initManagedLedgerConfig(ManagedLedgerConfig config) { + super.initManagedLedgerConfig(config); + // Inline checkForNewEntries: the +10ms task runs synchronously at arm time -> deterministic. + config.setNewEntriesCheckDelayInMillis(0); + config.setMaxEntriesPerLedger(1_000_000); + config.setRetentionTime(1, TimeUnit.HOURS); + config.setRetentionSizeInMB(-1); + return config; + } + + // ----------------------------------------------------------------------------------------------- + // Fixture + // ----------------------------------------------------------------------------------------------- + + /** Build a fresh topic/subscription/dispatcher on top of a real ledger + cursor. */ + private void buildFixture(String ledgerName) throws Exception { + ServiceConfiguration svcConfig = new ServiceConfiguration(); + svcConfig.setBrokerShutdownTimeoutMs(0L); + svcConfig.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d)); + svcConfig.setClusterName("test"); + svcConfig.setActiveConsumerFailoverDelayTimeMillis(0); + svcConfig.setSystemTopicEnabled(false); + svcConfig.setTopicLevelPoliciesEnabled(false); + + // Real BrokerService via PulsarTestContext, but backed by the real MockedBookKeeper-based + // ManagedLedgerFactory so we open a REAL ledger + cursor. + pulsarTestContext = PulsarTestContext.builderForNonStartableContext() + .config(svcConfig) + .spyByDefault() + .managedLedgerClients(bkc, factory) + .build(); + brokerService = pulsarTestContext.getBrokerService(); + + ledger = (ManagedLedgerImpl) factory.open(ledgerName, initManagedLedgerConfig(new ManagedLedgerConfig())); + topic = new PersistentTopic(TOPIC, ledger, brokerService); + // Open the cursor AFTER topic construction so the topic does not auto-create a second + // subscription sharing this cursor. + cursor = (ManagedCursorImpl) ledger.openCursor("sub"); + subscription = new PersistentSubscription(topic, "sub", cursor, false); + + // Inject the manual dispatcher executor via getTopicOrderedExecutor().chooseThread(), scoped to + // just the dispatcher construction so unrelated broker work keeps using the real executor. + OrderedExecutor realTopicOrdered = brokerService.getTopicOrderedExecutor(); + dispatcherExecutor = new ManualExecutor(); + OrderedExecutor topicOrdered = mock(OrderedExecutor.class); + doReturn(dispatcherExecutor).when(topicOrdered).chooseThread(); + doReturn(dispatcherExecutor).when(topicOrdered).chooseThread(any()); + doReturn(topicOrdered).when(brokerService).getTopicOrderedExecutor(); + dispatcher = spy(new PersistentDispatcherSingleActiveConsumer(cursor, SubType.Failover, -1, topic, + subscription)); + doReturn(realTopicOrdered).when(brokerService).getTopicOrderedExecutor(); + + consumer = mock(Consumer.class); + doReturn(1000).when(consumer).getAvailablePermits(); + doReturn(1).when(consumer).getAvgMessagesPerEntry(); + doReturn(true).when(consumer).isWritable(); + doReturn(false).when(consumer).readCompacted(); + doReturn(false).when(consumer).isPreciseDispatcherFlowControl(); + doReturn(false).when(consumer).isBlocked(); + doReturn("c1").when(consumer).consumerName(); + doReturn(0).when(consumer).getPriorityLevel(); + doReturn(0L).when(consumer).getConsumerEpoch(); + + // Stub only the message-delivery tail; preserve the readEntriesComplete invariant logic + re-arm post. + doAnswer(inv -> { + List entries = inv.getArgument(1); + entries.forEach(Entry::release); + dispatcherExecutor.execute(() -> dispatcher.readMoreEntries(dispatcher.getActiveConsumer())); + return null; + }).when(dispatcher).dispatchEntriesToConsumer(any(), any(), any(), any(), any(), anyLong()); + doReturn(0).when(dispatcher).filterEntriesForConsumer(any(), any(), any(), any(), any(), + anyBoolean(), any()); + } + + private void tearDownFixture() { + try { + if (cursor != null && !cursor.isClosed()) { + cursor.close(); + } + } catch (Exception ignore) { + // best-effort cleanup + } + try { + if (ledger != null) { + ledger.close(); + } + } catch (Exception ignore) { + // best-effort cleanup + } + try { + if (pulsarTestContext != null) { + pulsarTestContext.close(); + } + } catch (Exception ignore) { + // best-effort cleanup + } + cursor = null; + ledger = null; + brokerService = null; + pulsarTestContext = null; + } + + @AfterMethod(alwaysRun = true) + public void afterMethod() { + tearDownFixture(); + } + + // ----------------------------------------------------------------------------------------------- + // Manual executor task board + // ----------------------------------------------------------------------------------------------- + + /** A deterministic executor: submitted Runnables queue up and run only when the test drains them. */ + static final class ManualExecutor extends AbstractExecutorService { + private final ArrayDeque queue = new ArrayDeque<>(); + + @Override + public synchronized void execute(Runnable command) { + queue.add(command); + } + + synchronized int size() { + return queue.size(); + } + + /** Remove and return all currently-queued tasks (tasks added later are NOT included). */ + synchronized List takeAll() { + List snapshot = new ArrayList<>(queue); + queue.clear(); + return snapshot; + } + + /** Run one task FIFO; returns false if empty. */ + boolean runOne() { + Runnable r; + synchronized (this) { + r = queue.poll(); + } + if (r == null) { + return false; + } + r.run(); + return true; + } + + @Override + public void shutdown() { + } + + @Override + public List shutdownNow() { + return takeAll(); + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return true; + } + } + + // ----------------------------------------------------------------------------------------------- + // Operation alphabet (each drives REAL code) + // ----------------------------------------------------------------------------------------------- + + /** Connect the (only) consumer via the REAL addConsumer path; for Failover+delay=0 this arms the + * tail-wait read synchronously through scheduleReadOnActiveConsumer -> readMoreEntries. */ + private void arm() throws Exception { + dispatcher.addConsumer(consumer).get(); + } + + private Position publish() throws Exception { + return ledger.addEntry(("m" + msgCounter.incrementAndGet()).getBytes()); + } + + private void ackUpTo(Position p) throws Exception { + cursor.markDelete(p); + } + + /** Block (with an explicit timeout) until the dispatcher board holds at least {@code n} tasks. */ + private void awaitBoardHasAtLeast(int n) { + Awaitility.await("dispatcher board should hold at least " + n + " task(s)") + .atMost(Duration.ofSeconds(BOARD_WAIT_SECONDS)) + .pollInterval(Duration.ofMillis(2)) + .until(() -> dispatcherExecutor.size() >= n); + } + + /** + * Drain the dispatcher board until nothing new is produced, with a hard deadline so a regression can + * never hang CI. Async read completions posted by the (real) managed-ledger worker pool land on the + * board between idle rounds. + */ + private void quiesce() { + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(QUIESCE_TIMEOUT_SECONDS); + for (int idleRounds = 0; idleRounds < QUIESCE_IDLE_ROUNDS;) { + if (System.nanoTime() > deadlineNanos) { + throw new IllegalStateException("quiesce() did not settle within " + QUIESCE_TIMEOUT_SECONDS + + "s; board=" + dispatcherExecutor.size() + " pendingReadOps=" + + cursor.getPendingReadOpsCount()); + } + boolean did = false; + while (dispatcherExecutor.runOne()) { + did = true; + } + if (did) { + idleRounds = 0; + continue; + } + if (dispatcherExecutor.size() == 0 && cursor.getPendingReadOpsCount() == 0) { + idleRounds++; + } + sleepQuietly(); + } + } + + private static void sleepQuietly() { + try { + Thread.sleep(3); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * The last-consumer disconnect conversion for a durable cursor: the two operative production steps that + * {@code PersistentSubscription.removeConsumer} performs. {@code dispatcher.removeConsumer} calls + * {@code cancelPendingRead()} (which short-circuits when {@code havePendingRead} is already false, + * leaving the op armed), then the subscription strips the cursor from the waiting queue — see the + * comment at PersistentSubscription:381-388 explaining why the pending read is intentionally not + * cancelled for a durable cursor. + */ + private void applyLastConsumerDisconnect() throws Exception { + dispatcher.removeConsumer(consumer); + ledger.removeWaitingCursor(cursor); + } + + // ----------------------------------------------------------------------------------------------- + // State snapshot + predicates + // ----------------------------------------------------------------------------------------------- + + /** + * A snapshot of the four coupled state variables. {@code inQueue} is derived from + * {@code ledger.getWaitingCursorsCount()}; the single-cursor fixture makes that ledger-wide count a + * per-cursor membership proxy (0 ⇔ this cursor is not queued). + */ + private record StuckState(boolean armed, boolean inQueue, int pendingReadOps, boolean havePendingRead) { + + /** The absorbing orphan state: armed, not queued, no in-flight read, and havePendingRead cleared. */ + boolean stranded() { + return armed && !inQueue && pendingReadOps == 0 && !havePendingRead; + } + + /** The benign seed: armed with havePendingRead cleared, but still queued (so not yet stranded). */ + boolean seed() { + return armed && inQueue && !havePendingRead; + } + + @Override + public String toString() { + return "armed=" + armed + " inQueue=" + inQueue + " pendingReadOps=" + pendingReadOps + + " havePendingRead=" + havePendingRead; + } + } + + private StuckState snapshot() { + return new StuckState(cursor.hasPendingReadRequest(), ledger.getWaitingCursorsCount() >= 1, + cursor.getPendingReadOpsCount(), dispatcher.havePendingRead); + } + + /** + * Tail for the reproduction schedule: disconnect the last consumer, publish once more, and assert the + * five-leg failure at quiescence. FAILS deterministically on buggy master. + */ + private void assertCursorStrandedAfterDisconnectAndWake(StuckState seed) throws Exception { + applyLastConsumerDisconnect(); + quiesce(); + StuckState postDisconnect = snapshot(); + log.info().attr("state", postDisconnect).log("post-disconnect quiescent state"); + + // A subsequent publish MUST wake the subscription. + publish(); + quiesce(); + StuckState afterWake = snapshot(); + log.warn().attr("state", afterWake).log("post-wake-publish state"); + + // Five legs at quiescence: (1) armed, (2) absent from waitingCursors (count==0), (3) no in-flight + // read, (4) havePendingRead==false, and (5) the same tuple STILL holds after the wake publish. + Assert.assertFalse(afterWake.stranded(), + "BUG REPRODUCED: after the last consumer disconnected, the cursor holds an armed waitingReadOp " + + "(hasPendingReadRequest=true) but is absent from ManagedLedger.waitingCursors " + + "(waitingCursorsCount=0) with no in-flight read (pendingReadOps=0) while the dispatcher " + + "believes havePendingRead=false; a subsequent publish did not wake it (subscription " + + "permanently stuck). seed=[" + seed + "] post-disconnect=[" + postDisconnect + + "] after-wake-publish=[" + afterWake + "]"); + } + + // =============================================================================================== + // (a) PRIMARY reproduction: a redeliver races the completion of an already-dispatched read. m1's ack is + // telescoped (applied before that read completes) so the redeliver's rewind lands at the tail and the + // re-arm is a pure in-memory tail-wait -- fully deterministic. See the class Javadoc fidelity notes. + // =============================================================================================== + + @Test(groups = "broker") + public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() throws Exception { + buildFixture("sac-stuck-read-race"); + + // Arm the tail-wait read: addConsumer -> scheduleReadOnActiveConsumer -> readMoreEntries. + arm(); + Assert.assertTrue(cursor.hasPendingReadRequest(), "the tail-wait read op should be armed after addConsumer"); + Assert.assertEquals(ledger.getWaitingCursorsCount(), 1, "the cursor should be registered in waitingCursors"); + Assert.assertTrue(dispatcher.havePendingRead, "havePendingRead should be true while the tail read is armed"); + + // Publish m1: notifyCursors -> notifyEntriesAvailable dispatches the armed op's read, advancing the + // read position past m1 and posting readEntriesComplete onto the board. The wait op is now consumed, + // so hasPendingReadRequest() is false, but havePendingRead is still true. + Position m1 = publish(); + awaitBoardHasAtLeast(1); + Assert.assertFalse(cursor.hasPendingReadRequest(), + "the armed op should have been consumed by notifyEntriesAvailable"); + + // Telescoped ack: mark-delete m1 before its already-dispatched read completes. This lands the + // redeliver's subsequent rewind at the tail with no backlog, so the re-arm is a pure in-memory + // tail-wait and no BookKeeper read escapes the ManualExecutor to race quiesce(). The staged state + // (mark-delete at tail + an already-dispatched read whose completion is still pending) is + // production-reachable via a redeliver->ack->redeliver sequence (see the class Javadoc). + ackUpTo(m1); + + // Redeliver: posts internalRedeliver onto the board, behind the still-pending read completion. + dispatcher.redeliverUnacknowledgedMessages(consumer, 1L); + awaitBoardHasAtLeast(2); + + // The board now holds exactly two tasks from two different producers: + // [0] readEntriesComplete - posted by the managed-ledger/BookKeeper completion chain (whenCompleteAsync) + // [1] internalRedeliver - posted by redeliverUnacknowledgedMessages (the client-command thread) + List board = dispatcherExecutor.takeAll(); + Assert.assertEquals(board.size(), 2, + "the board must hold exactly two tasks from two producers: the stale read completion " + + "(posted by the managed-ledger/BookKeeper completion chain) and internalRedeliver " + + "(posted by the client-command thread)"); + + // Critical interleaving: run the redeliver's re-arm BEFORE the stale read completion. internalRedeliver + // clears havePendingRead and arms a fresh read; the stale completion then clears havePendingRead again + // while the fresh op stays armed -> the seed state. + board.get(1).run(); // internalRedeliver + board.get(0).run(); // stale readEntriesComplete + quiesce(); + + // Observation only (NOT asserted): the race mints the seed (armed op + havePendingRead=false, still + // queued). The seed is deliberately not asserted -- a completion-side fix (a read-generation guard on + // havePendingRead) would prevent the seed from ever forming, so binding on it would couple this test to + // one fix strategy. The single binding assertion is assertFalse(stranded) after the disconnect + wake. + StuckState seed = snapshot(); + log.info().attr("state", seed).attr("seedPreconditionHeld", seed.seed()) + .log("post-redeliver-race quiescent state (observation only)"); + + assertCursorStrandedAfterDisconnectAndWake(seed); + } + + // =============================================================================================== + // (b) NEGATIVE CONTROL: the same setup but the benign FIFO order [stale completion, then redeliver]. + // The invariant holds end to end; this PASSES on master. + // =============================================================================================== + + @Test(groups = "broker") + public void testRedeliverAfterReadCompletionDoesNotStrandCursor() throws Exception { + buildFixture("sac-stuck-read-fifo"); + + arm(); + Assert.assertTrue(cursor.hasPendingReadRequest(), "the tail-wait read op should be armed after addConsumer"); + Assert.assertEquals(ledger.getWaitingCursorsCount(), 1, "the cursor should be registered in waitingCursors"); + Assert.assertTrue(dispatcher.havePendingRead, "havePendingRead should be true while the tail read is armed"); + + Position m1 = publish(); + awaitBoardHasAtLeast(1); + Assert.assertFalse(cursor.hasPendingReadRequest(), + "the armed op should have been consumed by notifyEntriesAvailable"); + + // Same telescoped-ack staging as the primary test, so the benign schedule is equally deterministic + // (no off-board async read). The ONLY difference from the primary test is the board drain order below. + ackUpTo(m1); + + dispatcher.redeliverUnacknowledgedMessages(consumer, 1L); + awaitBoardHasAtLeast(2); + + List board = dispatcherExecutor.takeAll(); + Assert.assertEquals(board.size(), 2, + "the board must hold exactly two tasks from two producers: the read completion " + + "(posted by the managed-ledger/BookKeeper completion chain) and internalRedeliver " + + "(posted by the client-command thread)"); + + // Benign FIFO order: the stale completion runs FIRST (clearing havePendingRead), then the redeliver + // re-arms and sets havePendingRead=true. The armed op and havePendingRead stay coupled, so no seed + // forms and the later disconnect can cancel the op cleanly. + board.get(0).run(); // stale readEntriesComplete + board.get(1).run(); // internalRedeliver + quiesce(); + + StuckState healthy = snapshot(); + log.info().attr("state", healthy).log("post-FIFO-race quiescent state"); + Assert.assertFalse(healthy.stranded(), "the benign FIFO order must not strand the cursor: " + healthy); + + // The disconnect now finds havePendingRead=true, so cancelPendingRead actually cancels the armed op: + // no orphan is created, and the invariant holds end to end. + applyLastConsumerDisconnect(); + quiesce(); + StuckState postDisconnect = snapshot(); + log.info().attr("state", postDisconnect).log("post-disconnect quiescent state"); + + publish(); + quiesce(); + StuckState afterWake = snapshot(); + log.info().attr("state", afterWake).log("post-wake-publish state"); + + Assert.assertFalse(afterWake.stranded(), + "NEGATIVE CONTROL: with the benign FIFO order the cursor must never be stranded. " + + "healthy=[" + healthy + "] post-disconnect=[" + postDisconnect + "] after-wake-publish=[" + + afterWake + "]"); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index d7ef063a34d1d..232f127d47472 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -102,7 +102,7 @@ public void testSkipReadEntriesFromCloseCursor() throws Exception { Mockito.doAnswer(inv -> { callReadEntriesFailed.getAndIncrement(); return inv.callRealMethod(); - }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any()); + }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any(), Mockito.anyLong()); Mockito.doReturn(false).when(cursor).isClosed(); From 453de80d163d3373550f6aedb4b036e7efc1476b Mon Sep 17 00:00:00 2001 From: Alexandre Burgoni Date: Thu, 23 Jul 2026 18:58:58 +0200 Subject: [PATCH 2/2] Address review feedback on the stale read completion fix - Assert the externally observable contract of #26164: both tests now reconnect a fresh consumer after the last-consumer disconnect and require the next published message to actually reach Consumer#sendMessages and advance the cursor read position. The internal state tuple is kept only as a secondary diagnostic, so the regression test is no longer coupled to one fix strategy. - Drive the consumer lifecycle through the real PersistentSubscription#addConsumer and removeConsumer(Consumer, boolean) instead of replaying two selected steps by hand, so dispatcher removal, deactivateCursor() and removeWaitingCursor() run in the production order. The spied dispatcher is injected through the existing PersistentSubscription#reuseOrCreateDispatcher test seam; no production code was added for the test. - Replace the hard-coded line-number references in the test Javadoc with durable method-name anchors, which cannot silently rot as the referenced files change. - Record both the stale and the current readOpEpoch on the two stale-read debug log paths, so a suspected recurrence in production shows which epochs disagreed. --- ...sistentDispatcherSingleActiveConsumer.java | 4 + ...cherSingleActiveConsumerStuckReadTest.java | 374 +++++++++++------- 2 files changed, 236 insertions(+), 142 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java index b08860d7e2000..8c2b1d820520b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java @@ -174,6 +174,8 @@ private synchronized void readEntriesComplete(List entries, Consumer read log.debug() .attr("readConsumer", readConsumer) .attr("size", entries.size()) + .attr("staleReadOpEpoch", readOpEpoch) + .attr("currentReadOpEpoch", this.readOpEpoch) .log("Discarding stale read completion"); entries.forEach(Entry::release); return; @@ -479,6 +481,8 @@ public synchronized void readEntriesFailed(Throwable throwable, Consumer consume // or rescheduling here would disturb the newer outstanding read. Ignore it. log.debug() .attr("consumer", consumer) + .attr("staleReadOpEpoch", readOpEpoch) + .attr("currentReadOpEpoch", this.readOpEpoch) .log("Ignoring stale read failure"); return; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java index 48dd478229ff5..c6a207699b5ee 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java @@ -20,14 +20,17 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; +import io.netty.util.concurrent.ImmediateEventExecutor; import java.time.Duration; import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Optional; import java.util.concurrent.AbstractExecutorService; @@ -44,8 +47,12 @@ import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.Dispatcher; +import org.apache.pulsar.broker.service.EntryBatchIndexesAcks; +import org.apache.pulsar.broker.service.EntryBatchSizes; import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; +import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl; import org.awaitility.Awaitility; import org.testng.Assert; import org.testng.annotations.AfterMethod; @@ -69,43 +76,56 @@ * In healthy operation an armed wait op implies {@code havePendingRead}, and implies the cursor is * either queued in {@code waitingCursors} or has an in-flight read. * - *

The race. {@code internalRedeliverUnacknowledgedMessages} - * (PersistentDispatcherSingleActiveConsumer:275-306) unconditionally clears {@code havePendingRead} - * (:300) and arms a fresh read via {@code readMoreEntries} (:305) without draining a read that - * {@code notifyEntriesAvailable} has already dispatched. The stale read's completion - * ({@code readEntriesComplete} :168 / {@code readEntriesFailed} :456) then clears {@code havePendingRead} - * again: there is no read-generation guard on {@code havePendingRead}, and the only staleness check is - * the consumer-identity comparison at :200, which runs after the flag was already cleared at :168. - * When the redeliver's re-arm and the stale completion interleave in the order [redeliver, then stale - * completion], the result is the seed state — an armed wait op with {@code havePendingRead == false}. - * This is still benign while the cursor remains in {@code waitingCursors}. + *

The race. + * {@code PersistentDispatcherSingleActiveConsumer#internalRedeliverUnacknowledgedMessages} unconditionally + * clears {@code havePendingRead} and arms a fresh read via {@code readMoreEntries} without draining + * a read that {@code notifyEntriesAvailable} has already dispatched. The stale read's completion + * ({@code readEntriesComplete} / {@code readEntriesFailed}) then clears {@code havePendingRead} again: + * there is no read-generation guard on {@code havePendingRead}, and the only staleness check is the + * {@code readConsumer != currentConsumer} identity comparison in {@code readEntriesComplete}, which runs + * after that same method has already cleared the flag at its top. When the redeliver's re-arm and + * the stale completion interleave in the order [redeliver, then stale completion], the result is the seed + * state — an armed wait op with {@code havePendingRead == false}. This is still benign while the + * cursor remains in {@code waitingCursors}. * *

The absorbing conversion. When the last consumer disconnects, the durable-cursor path - * ({@code AbstractDispatcherSingleActiveConsumer.removeConsumer} → {@code cancelPendingRead} :157, - * then {@code ManagedLedger.removeWaitingCursor}) deliberately does not cancel the pending read for a - * durable cursor — see the comment at PersistentSubscription:381-388. Because {@code havePendingRead} - * is already {@code false}, {@code cancelPendingRead} short-circuits and leaves the wait op armed, while - * {@code removeWaitingCursor} strips the cursor from the queue. The subscription is now in the absorbing - * orphan state: an armed wait op, absent from {@code waitingCursors}, with no in-flight read and - * {@code havePendingRead == false}. Every subsequent re-arm CAS-fails at ManagedCursorImpl:1110 with - * {@code ConcurrentWaitCallbackException} (which {@code readEntriesFailed} :465-468 does not reschedule), - * and every publish's {@code notifyCursors} poll-misses. The subscription never reads again. + * ({@code AbstractDispatcherSingleActiveConsumer#removeConsumer} → {@code cancelPendingRead}, then + * {@code ManagedLedger#removeWaitingCursor}) deliberately does not cancel the pending read for a durable + * cursor — see the durable-cursor comment in {@code PersistentSubscription#removeConsumer}. Because + * {@code havePendingRead} is already {@code false}, {@code cancelPendingRead} short-circuits on its + * {@code havePendingRead} guard and leaves the wait op armed, while {@code removeWaitingCursor} strips the + * cursor from the queue. The subscription is now in the absorbing orphan state: an armed wait op, absent + * from {@code waitingCursors}, with no in-flight read and {@code havePendingRead == false}. Every + * subsequent re-arm CAS-fails on the {@code WAITING_READ_OP_UPDATER} compare-and-set in + * {@code ManagedCursorImpl#asyncReadEntriesWithSkipOrWait}, raising {@code ConcurrentWaitCallbackException} + * — which {@code readEntriesFailed} returns early on instead of rescheduling — and every + * publish's {@code notifyCursors} poll-misses. The subscription never reads again — which is exactly + * what both tests assert behaviourally: after the disconnect a fresh consumer reconnects, and a message + * published afterwards must actually reach that consumer's {@code sendMessages}. * *

The two racing events genuinely arrive on different threads: the read completion is posted to the - * dispatcher's ordered executor by the managed-ledger/BookKeeper completion chain - * ({@code whenCompleteAsync}, PersistentDispatcherSingleActiveConsumer:380-386), while the redeliver is - * posted by the client-command thread (:272). Their arrival order is a real race. + * dispatcher's ordered executor by the managed-ledger/BookKeeper completion chain (the + * {@code entriesFuture.whenCompleteAsync(..., executor)} hand-off at the end of {@code readMoreEntries}), + * while the redeliver is posted by {@code redeliverUnacknowledgedMessages} from the client-command thread. + * Their arrival order is a real race. * *

Fidelity notes / test seams. *

    - *
  • The cursor, ledger, dispatcher, {@code waitingReadOp} compare-and-set, {@code checkForNewEntries}, - * {@code notifyEntriesAvailable}, {@code notifyCursors}, {@code cancelPendingReadRequest}, - * {@code addWaitingCursor}/{@code removeWaitingCursor} and {@code havePendingRead} management are all - * real production code.
    • - *
  • The dispatcher's message-delivery tail ({@code filterEntriesForConsumer} + - * {@code dispatchEntriesToConsumer}) is stubbed on a Mockito spy to release the entries and post the - * same {@code readMoreEntries} continuation the real code posts at {@code dispatchEntriesToConsumer} - * :239. It touches none of the four state variables above.
    • + *
  • The cursor, ledger, subscription, dispatcher, {@code waitingReadOp} compare-and-set, + * {@code checkForNewEntries}, {@code notifyEntriesAvailable}, {@code notifyCursors}, + * {@code cancelPendingReadRequest}, {@code addWaitingCursor}/{@code removeWaitingCursor} and + * {@code havePendingRead} management are all real production code.
    • + *
  • The consumer lifecycle runs through the real {@code PersistentSubscription#addConsumer} and + * {@code PersistentSubscription#removeConsumer(Consumer, boolean)}, so dispatcher removal, + * {@code deactivateCursor()} and {@code removeWaitingCursor()} happen in the production order. The + * spied dispatcher is handed to the subscription through the existing production test seam + * {@code PersistentSubscription#reuseOrCreateDispatcher}.
    • + *
  • Only the metadata-parsing filter step ({@code filterEntriesForConsumer}) is stubbed on a Mockito + * spy, because the entries this test publishes are raw bytes rather than serialized Pulsar messages. + * The delivery tail itself is real: {@code dispatchEntriesToConsumer} runs and calls + * {@code Consumer#sendMessages}, which the mock consumer records before releasing the entries and + * completing immediately — so the production {@code readMoreEntries} continuation is posted by + * the real listener. Neither seam touches the four state variables above.
    • *
  • The dispatcher's ordered executor is replaced with a {@link ManualExecutor} (injected through a * spied {@code getTopicOrderedExecutor()} scoped to dispatcher construction) so every dispatcher * async hop becomes an explicit task the test drains in a chosen order.
    • @@ -114,10 +134,11 @@ * the ack the redeliver's {@code rewind} lands at the tail with no backlog, so the re-arm is a pure * in-memory tail-wait and no BookKeeper read escapes the {@link ManualExecutor} to race the drain. * Without the ack the {@code rewind} re-exposes the unacked entry and {@code readMoreEntries} takes - * the immediate-read branch (ManagedCursorImpl:1099-1103), issuing a real async read whose completion - * is off-board. The staged state (mark-delete at the tail plus an already-dispatched read whose - * completion is still pending) is production-reachable via a redeliver→ack→redeliver - * sequence. + * the immediate-read branch of {@code ManagedCursorImpl#asyncReadEntriesWithSkipOrWait} (the + * {@code hasMoreEntries()} fast path that delegates straight to {@code asyncReadEntriesWithSkip}), + * issuing a real async read whose completion is off-board. The staged state (mark-delete at the + * tail plus an already-dispatched read whose completion is still pending) is production-reachable + * via a redeliver→ack→redeliver sequence. *
  • {@code newEntriesCheckDelayInMillis} is pinned to {@code 0} so {@code checkForNewEntries} runs * inline at arm time rather than on the ledger scheduler. This is a determinism pin only; the bug * itself is not an artifact of the zero delay. With the telescoped ack in place, every @@ -134,6 +155,7 @@ public class PersistentDispatcherSingleActiveConsumerStuckReadTest extends Mocke private static final int BOARD_WAIT_SECONDS = 5; private static final long QUIESCE_TIMEOUT_SECONDS = 10; private static final int QUIESCE_IDLE_ROUNDS = 4; + private static final long DELIVERY_TIMEOUT_SECONDS = 5; private PulsarTestContext pulsarTestContext; private BrokerService brokerService; @@ -142,7 +164,7 @@ public class PersistentDispatcherSingleActiveConsumerStuckReadTest extends Mocke private ManagedLedgerImpl ledger; private ManagedCursorImpl cursor; - // Real dispatcher (Mockito spy so we can stub the delivery tail + pin the active consumer). + // Real topic + subscription; the dispatcher is a Mockito spy so we can stub the entry filter step. private PersistentTopic topic; private PersistentSubscription subscription; private PersistentDispatcherSingleActiveConsumer dispatcher; @@ -151,6 +173,9 @@ public class PersistentDispatcherSingleActiveConsumerStuckReadTest extends Mocke // The dispatcher's ordered executor: a manual task board we drain in a chosen order. private ManualExecutor dispatcherExecutor; + // Everything the dispatcher handed to a Consumer#sendMessages tail, in delivery order. + private final List deliveries = Collections.synchronizedList(new ArrayList<>()); + private final AtomicLong msgCounter = new AtomicLong(); @Override @@ -192,7 +217,15 @@ private void buildFixture(String ledgerName) throws Exception { // Open the cursor AFTER topic construction so the topic does not auto-create a second // subscription sharing this cursor. cursor = (ManagedCursorImpl) ledger.openCursor("sub"); - subscription = new PersistentSubscription(topic, "sub", cursor, false); + // A real subscription that adopts the spied dispatcher through the production test seam, so + // addConsumer/removeConsumer drive the real consumer lifecycle. The returned reference must be + // qualified: the inherited PersistentSubscription.dispatcher field would otherwise shadow ours. + subscription = new PersistentSubscription(topic, "sub", cursor, false) { + @Override + protected Dispatcher reuseOrCreateDispatcher(Dispatcher existingDispatcher, Consumer newConsumer) { + return PersistentDispatcherSingleActiveConsumerStuckReadTest.this.dispatcher; + } + }; // Inject the manual dispatcher executor via getTopicOrderedExecutor().chooseThread(), scoped to // just the dispatcher construction so unrelated broker work keeps using the real executor. @@ -206,26 +239,53 @@ private void buildFixture(String ledgerName) throws Exception { subscription)); doReturn(realTopicOrdered).when(brokerService).getTopicOrderedExecutor(); - consumer = mock(Consumer.class); - doReturn(1000).when(consumer).getAvailablePermits(); - doReturn(1).when(consumer).getAvgMessagesPerEntry(); - doReturn(true).when(consumer).isWritable(); - doReturn(false).when(consumer).readCompacted(); - doReturn(false).when(consumer).isPreciseDispatcherFlowControl(); - doReturn(false).when(consumer).isBlocked(); - doReturn("c1").when(consumer).consumerName(); - doReturn(0).when(consumer).getPriorityLevel(); - doReturn(0L).when(consumer).getConsumerEpoch(); - - // Stub only the message-delivery tail; preserve the readEntriesComplete invariant logic + re-arm post. - doAnswer(inv -> { - List entries = inv.getArgument(1); - entries.forEach(Entry::release); - dispatcherExecutor.execute(() -> dispatcher.readMoreEntries(dispatcher.getActiveConsumer())); - return null; - }).when(dispatcher).dispatchEntriesToConsumer(any(), any(), any(), any(), any(), anyLong()); + // The published payloads are raw bytes, not serialized Pulsar messages, so the metadata-parsing + // filter step is stubbed out. The rest of the delivery tail (dispatchEntriesToConsumer -> + // Consumer#sendMessages) is real production code. doReturn(0).when(dispatcher).filterEntriesForConsumer(any(), any(), any(), any(), any(), anyBoolean(), any()); + + deliveries.clear(); + consumer = newMockConsumer("c1"); + } + + /** + * A mock consumer whose {@code sendMessages} tail records what the dispatcher delivered, releases the + * entries and completes immediately. The succeeded future notifies its listener inline, so the + * production {@code readMoreEntries} continuation lands on the manual board exactly as it does in + * production. + */ + private Consumer newMockConsumer(String consumerName) { + Consumer mockConsumer = mock(Consumer.class); + doReturn(1000).when(mockConsumer).getAvailablePermits(); + doReturn(1).when(mockConsumer).getAvgMessagesPerEntry(); + doReturn(true).when(mockConsumer).isWritable(); + doReturn(false).when(mockConsumer).readCompacted(); + doReturn(false).when(mockConsumer).isPreciseDispatcherFlowControl(); + doReturn(false).when(mockConsumer).isBlocked(); + doReturn(consumerName).when(mockConsumer).consumerName(); + doReturn(0).when(mockConsumer).getPriorityLevel(); + doReturn(0L).when(mockConsumer).getConsumerEpoch(); + doReturn(SubType.Failover).when(mockConsumer).subType(); + // PersistentSubscription.removeConsumer folds the removed consumer's counters into the subscription. + doReturn(new ConsumerStatsImpl()).when(mockConsumer).getStats(); + doAnswer(inv -> { + List entries = inv.getArgument(0); + for (Entry entry : entries) { + deliveries.add(Delivery.of(consumerName, entry.getPosition())); + entry.release(); + } + EntryBatchSizes batchSizes = inv.getArgument(1); + if (batchSizes != null) { + batchSizes.recyle(); + } + EntryBatchIndexesAcks batchIndexesAcks = inv.getArgument(2); + if (batchIndexesAcks != null) { + batchIndexesAcks.recycle(); + } + return ImmediateEventExecutor.INSTANCE.newSucceededFuture(null); + }).when(mockConsumer).sendMessages(any(), any(), any(), anyInt(), anyLong(), anyLong(), any(), anyLong()); + return mockConsumer; } private void tearDownFixture() { @@ -327,10 +387,22 @@ public boolean awaitTermination(long timeout, TimeUnit unit) { // Operation alphabet (each drives REAL code) // ----------------------------------------------------------------------------------------------- - /** Connect the (only) consumer via the REAL addConsumer path; for Failover+delay=0 this arms the - * tail-wait read synchronously through scheduleReadOnActiveConsumer -> readMoreEntries. */ - private void arm() throws Exception { - dispatcher.addConsumer(consumer).get(); + /** Connect a consumer via the REAL PersistentSubscription.addConsumer path; for Failover with a zero + * failover delay this arms the tail-wait read synchronously through the dispatcher's + * scheduleReadOnActiveConsumer -> readMoreEntries. */ + private void connect(Consumer consumerToAdd) throws Exception { + subscription.addConsumer(consumerToAdd).get(); + } + + /** + * Disconnect a consumer via the REAL {@code PersistentSubscription.removeConsumer(consumer, false)} path, + * so dispatcher removal (which calls {@code cancelPendingRead()}), {@code deactivateCursor()} and + * {@code removeWaitingCursor()} all run in the production order. For a durable cursor the subscription + * deliberately does not cancel the pending read — see the durable-cursor comment in + * {@code PersistentSubscription#removeConsumer}. + */ + private void disconnect(Consumer consumerToRemove) throws Exception { + subscription.removeConsumer(consumerToRemove, false); } private Position publish() throws Exception { @@ -377,6 +449,25 @@ private void quiesce() { } } + /** + * Keep draining the dispatcher board until {@code position} has been handed to {@code target}'s + * {@code sendMessages}, with a hard deadline so a regression fails instead of hanging CI. + */ + private boolean pumpUntilDelivered(Consumer target, Position position) { + Delivery expected = Delivery.of(target.consumerName(), position); + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(DELIVERY_TIMEOUT_SECONDS); + do { + while (dispatcherExecutor.runOne()) { + // A delivery posts follow-up work onto the board; keep draining before re-checking. + } + if (deliveries.contains(expected)) { + return true; + } + sleepQuietly(); + } while (System.nanoTime() < deadlineNanos); + return false; + } + private static void sleepQuietly() { try { Thread.sleep(3); @@ -385,23 +476,23 @@ private static void sleepQuietly() { } } - /** - * The last-consumer disconnect conversion for a durable cursor: the two operative production steps that - * {@code PersistentSubscription.removeConsumer} performs. {@code dispatcher.removeConsumer} calls - * {@code cancelPendingRead()} (which short-circuits when {@code havePendingRead} is already false, - * leaving the op armed), then the subscription strips the cursor from the waiting queue — see the - * comment at PersistentSubscription:381-388 explaining why the pending read is intentionally not - * cancelled for a durable cursor. - */ - private void applyLastConsumerDisconnect() throws Exception { - dispatcher.removeConsumer(consumer); - ledger.removeWaitingCursor(cursor); - } - // ----------------------------------------------------------------------------------------------- // State snapshot + predicates // ----------------------------------------------------------------------------------------------- + /** One entry handed to a consumer by the dispatcher's real delivery tail. */ + private record Delivery(String consumerName, long ledgerId, long entryId) { + + static Delivery of(String consumerName, Position position) { + return new Delivery(consumerName, position.getLedgerId(), position.getEntryId()); + } + + @Override + public String toString() { + return consumerName + "@" + ledgerId + ":" + entryId; + } + } + /** * A snapshot of the four coupled state variables. {@code inQueue} is derived from * {@code ledger.getWaitingCursorsCount()}; the single-cursor fixture makes that ledger-wide count a @@ -432,44 +523,55 @@ private StuckState snapshot() { } /** - * Tail for the reproduction schedule: disconnect the last consumer, publish once more, and assert the - * five-leg failure at quiescence. FAILS deterministically on buggy master. + * Tail shared by both schedules, asserting the externally observable contract of #26164: after the last + * consumer disconnects and a fresh consumer reconnects to the same durable subscription, a message + * published afterwards must actually be delivered to that consumer. On buggy master the primary + * schedule leaves the cursor in the absorbing orphan state, the reconnect's re-arm CAS-fails and the + * message never reaches {@code sendMessages}. */ - private void assertCursorStrandedAfterDisconnectAndWake(StuckState seed) throws Exception { - applyLastConsumerDisconnect(); + private void assertSubscriptionDeliversAfterReconnect(String schedule, StuckState preDisconnect) + throws Exception { + disconnect(consumer); quiesce(); StuckState postDisconnect = snapshot(); log.info().attr("state", postDisconnect).log("post-disconnect quiescent state"); - // A subsequent publish MUST wake the subscription. - publish(); + // A client reconnect: a brand new consumer takes over the (still durable) subscription. + Consumer reconnected = newMockConsumer("c2"); + connect(reconnected); quiesce(); - StuckState afterWake = snapshot(); - log.warn().attr("state", afterWake).log("post-wake-publish state"); + StuckState postReconnect = snapshot(); + log.info().attr("state", postReconnect).log("post-reconnect quiescent state"); - // Five legs at quiescence: (1) armed, (2) absent from waitingCursors (count==0), (3) no in-flight - // read, (4) havePendingRead==false, and (5) the same tuple STILL holds after the wake publish. + Position m2 = publish(); + boolean deliveredM2 = pumpUntilDelivered(reconnected, m2); + quiesce(); + StuckState afterWake = snapshot(); + log.info().attr("state", afterWake).attr("delivered", deliveredM2).log("post-wake-publish state"); + + String states = "schedule=[" + schedule + "] pre-disconnect=[" + preDisconnect + "] post-disconnect=[" + + postDisconnect + "] post-reconnect=[" + postReconnect + "] after-wake-publish=[" + afterWake + + "] deliveries=" + deliveries; + + // Primary, behavioural: the reconnected consumer must actually receive the new message. + Assert.assertTrue(deliveredM2, + "BUG REPRODUCED: after the last consumer disconnected and a new one reconnected, the message " + + "published next (" + m2 + ") never reached the reconnected consumer's sendMessages within " + + DELIVERY_TIMEOUT_SECONDS + "s; the Failover subscription is permanently stuck. " + states); + Assert.assertTrue(cursor.getReadPosition().compareTo(m2) > 0, + "the cursor read position must have advanced past the delivered message " + m2 + ", but it is " + + cursor.getReadPosition() + ". " + states); + + // Secondary, diagnostic: name the internal tuple that causes the stall, so a failure points at it. Assert.assertFalse(afterWake.stranded(), - "BUG REPRODUCED: after the last consumer disconnected, the cursor holds an armed waitingReadOp " - + "(hasPendingReadRequest=true) but is absent from ManagedLedger.waitingCursors " - + "(waitingCursorsCount=0) with no in-flight read (pendingReadOps=0) while the dispatcher " - + "believes havePendingRead=false; a subsequent publish did not wake it (subscription " - + "permanently stuck). seed=[" + seed + "] post-disconnect=[" + postDisconnect - + "] after-wake-publish=[" + afterWake + "]"); + "the cursor must not be left in the absorbing orphan state (armed waitingReadOp, absent from " + + "ManagedLedger.waitingCursors, no in-flight read, havePendingRead=false). " + states); } - // =============================================================================================== - // (a) PRIMARY reproduction: a redeliver races the completion of an already-dispatched read. m1's ack is - // telescoped (applied before that read completes) so the redeliver's rewind lands at the tail and the - // re-arm is a pure in-memory tail-wait -- fully deterministic. See the class Javadoc fidelity notes. - // =============================================================================================== - - @Test(groups = "broker") - public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() throws Exception { - buildFixture("sac-stuck-read-race"); - + /** Stage the shared precondition: an armed tail-wait read plus a read completion pending on the board. */ + private void stageInFlightReadCompletion() throws Exception { // Arm the tail-wait read: addConsumer -> scheduleReadOnActiveConsumer -> readMoreEntries. - arm(); + connect(consumer); Assert.assertTrue(cursor.hasPendingReadRequest(), "the tail-wait read op should be armed after addConsumer"); Assert.assertEquals(ledger.getWaitingCursorsCount(), 1, "the cursor should be registered in waitingCursors"); Assert.assertTrue(dispatcher.havePendingRead, "havePendingRead should be true while the tail read is armed"); @@ -492,15 +594,37 @@ public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() // Redeliver: posts internalRedeliver onto the board, behind the still-pending read completion. dispatcher.redeliverUnacknowledgedMessages(consumer, 1L); awaitBoardHasAtLeast(2); + } - // The board now holds exactly two tasks from two different producers: - // [0] readEntriesComplete - posted by the managed-ledger/BookKeeper completion chain (whenCompleteAsync) - // [1] internalRedeliver - posted by redeliverUnacknowledgedMessages (the client-command thread) + /** + * The board now holds exactly two tasks from two different producers: + *
      + *
    1. {@code readEntriesComplete} — posted by the managed-ledger/BookKeeper completion chain + * ({@code whenCompleteAsync});
      2. + *
    2. {@code internalRedeliver} — posted by {@code redeliverUnacknowledgedMessages} (the + * client-command thread).
      3. + *
    + */ + private List takeTwoRacingTasks() { List board = dispatcherExecutor.takeAll(); Assert.assertEquals(board.size(), 2, - "the board must hold exactly two tasks from two producers: the stale read completion " + "the board must hold exactly two tasks from two producers: the read completion " + "(posted by the managed-ledger/BookKeeper completion chain) and internalRedeliver " + "(posted by the client-command thread)"); + return board; + } + + // =============================================================================================== + // (a) PRIMARY reproduction: a redeliver races the completion of an already-dispatched read. m1's ack is + // telescoped (applied before that read completes) so the redeliver's rewind lands at the tail and the + // re-arm is a pure in-memory tail-wait -- fully deterministic. See the class Javadoc fidelity notes. + // =============================================================================================== + + @Test(groups = "broker") + public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() throws Exception { + buildFixture("sac-stuck-read-race"); + stageInFlightReadCompletion(); + List board = takeTwoRacingTasks(); // Critical interleaving: run the redeliver's re-arm BEFORE the stale read completion. internalRedeliver // clears havePendingRead and arms a fresh read; the stale completion then clears havePendingRead again @@ -512,12 +636,12 @@ public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() // Observation only (NOT asserted): the race mints the seed (armed op + havePendingRead=false, still // queued). The seed is deliberately not asserted -- a completion-side fix (a read-generation guard on // havePendingRead) would prevent the seed from ever forming, so binding on it would couple this test to - // one fix strategy. The single binding assertion is assertFalse(stranded) after the disconnect + wake. + // one fix strategy. The binding assertions are the behavioural ones after the disconnect + reconnect. StuckState seed = snapshot(); log.info().attr("state", seed).attr("seedPreconditionHeld", seed.seed()) .log("post-redeliver-race quiescent state (observation only)"); - assertCursorStrandedAfterDisconnectAndWake(seed); + assertSubscriptionDeliversAfterReconnect("redeliver, then stale completion", seed); } // =============================================================================================== @@ -528,29 +652,8 @@ public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() @Test(groups = "broker") public void testRedeliverAfterReadCompletionDoesNotStrandCursor() throws Exception { buildFixture("sac-stuck-read-fifo"); - - arm(); - Assert.assertTrue(cursor.hasPendingReadRequest(), "the tail-wait read op should be armed after addConsumer"); - Assert.assertEquals(ledger.getWaitingCursorsCount(), 1, "the cursor should be registered in waitingCursors"); - Assert.assertTrue(dispatcher.havePendingRead, "havePendingRead should be true while the tail read is armed"); - - Position m1 = publish(); - awaitBoardHasAtLeast(1); - Assert.assertFalse(cursor.hasPendingReadRequest(), - "the armed op should have been consumed by notifyEntriesAvailable"); - - // Same telescoped-ack staging as the primary test, so the benign schedule is equally deterministic - // (no off-board async read). The ONLY difference from the primary test is the board drain order below. - ackUpTo(m1); - - dispatcher.redeliverUnacknowledgedMessages(consumer, 1L); - awaitBoardHasAtLeast(2); - - List board = dispatcherExecutor.takeAll(); - Assert.assertEquals(board.size(), 2, - "the board must hold exactly two tasks from two producers: the read completion " - + "(posted by the managed-ledger/BookKeeper completion chain) and internalRedeliver " - + "(posted by the client-command thread)"); + stageInFlightReadCompletion(); + List board = takeTwoRacingTasks(); // Benign FIFO order: the stale completion runs FIRST (clearing havePendingRead), then the redeliver // re-arms and sets havePendingRead=true. The armed op and havePendingRead stay coupled, so no seed @@ -564,20 +667,7 @@ public void testRedeliverAfterReadCompletionDoesNotStrandCursor() throws Excepti Assert.assertFalse(healthy.stranded(), "the benign FIFO order must not strand the cursor: " + healthy); // The disconnect now finds havePendingRead=true, so cancelPendingRead actually cancels the armed op: - // no orphan is created, and the invariant holds end to end. - applyLastConsumerDisconnect(); - quiesce(); - StuckState postDisconnect = snapshot(); - log.info().attr("state", postDisconnect).log("post-disconnect quiescent state"); - - publish(); - quiesce(); - StuckState afterWake = snapshot(); - log.info().attr("state", afterWake).log("post-wake-publish state"); - - Assert.assertFalse(afterWake.stranded(), - "NEGATIVE CONTROL: with the benign FIFO order the cursor must never be stranded. " - + "healthy=[" + healthy + "] post-disconnect=[" + postDisconnect + "] after-wake-publish=[" - + afterWake + "]"); + // no orphan is created, the reconnect re-arms cleanly and the next publish is delivered. + assertSubscriptionDeliversAfterReconnect("stale completion, then redeliver (negative control)", healthy); } }