(pulsar) 03/11: [fix][broker] Trigger max read position callback for messages published during transaction buffer recovery (#26234)

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit d7a7a6881fb7dfcd13634bc174bfc60b8f4650eb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 23 19:15:23 2026 +0300

    [fix][broker] Trigger max read position callback for messages published during transaction buffer recovery (#26234)
    
    (cherry picked from commit 520b0daa3f632e283930624bf29e7a5b4ed45828)
---
 .../buffer/impl/TopicTransactionBuffer.java        |  21 ++-
 .../impl/TopicTransactionBufferRecoveryTest.java   | 144 +++++++++++++++++++++
 2 files changed, 163 insertions(+), 2 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index 7ae311a54639..aad9862cf4fa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -150,7 +150,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
                     public void recoverComplete() {
                         synchronized (TopicTransactionBuffer.this) {
                             if (ongoingTxns.isEmpty()) {
-                                maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+                                updateMaxReadPositionAfterRecovery();
                             }
                             if (!changeToReadyState()) {
                                 log.error("[{}]Transaction buffer recover fail, current state: {}",
@@ -171,7 +171,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
                     @Override
                     public void noNeedToRecover() {
                         synchronized (TopicTransactionBuffer.this) {
-                            maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+                            updateMaxReadPositionAfterRecovery();
                             if (!changeToNoSnapshotState()) {
                                 log.error("[{}]Transaction buffer recover fail", topic.getName());
                             } else {
@@ -627,6 +627,23 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
         }
     }
 
+    /**
+     * Advance the max read position to the last confirmed entry when recovery finishes. While the transaction
+     * buffer is recovering, {@link #syncMaxReadPositionForNormalPublish(Position, boolean)} ignores publishes, so
+     * messages published during recovery would otherwise never trigger the maxReadPositionMovedForward callback.
+     * {@link PersistentTopic} uses that callback to maintain lastMaxReadPositionMovedForwardTimestamp, which
+     * ReplicatedSubscriptionsController relies on to detect new data when deciding whether to start a snapshot.
+     * Must be called while synchronized on this transaction buffer.
+     */
+    private void updateMaxReadPositionAfterRecovery() {
+        Position preMaxReadPosition = this.maxReadPosition;
+        this.maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+        if (this.maxReadPosition != null
+                && (preMaxReadPosition == null || preMaxReadPosition.compareTo(this.maxReadPosition) < 0)) {
+            this.maxReadPositionCallBack.maxReadPositionMovedForward(preMaxReadPosition, this.maxReadPosition);
+        }
+    }
+
     @Override
     public CompletableFuture<Void> purgeTxns(List<Long> dataLedgers) {
         return null;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
new file mode 100644
index 000000000000..101f69b263f8
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.buffer.impl;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/**
+ * Tests for {@link TopicTransactionBuffer} recovery completion behavior.
+ */
+@Test(groups = "broker")
+public class TopicTransactionBufferRecoveryTest extends ProducerConsumerBase {
+
+    @BeforeClass(alwaysRun = true)
+    @Override
+    protected void setup() throws Exception {
+        conf.setTransactionCoordinatorEnabled(true);
+        super.internalSetup();
+        super.producerBaseSetup();
+    }
+
+    @AfterClass(alwaysRun = true)
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @DataProvider(name = "snapshotExists")
+    public Object[][] snapshotExists() {
+        return new Object[][] { { false }, { true } };
+    }
+
+    /**
+     * While the transaction buffer is recovering, normal publishes don't move the max read position, so they
+     * don't update the topic's lastMaxReadPositionMovedForwardTimestamp either. When recovery completes, the
+     * transaction buffer must account for the messages published during recovery and trigger the
+     * maxReadPositionMovedForward callback; otherwise ReplicatedSubscriptionsController would consider the topic
+     * to have no new data and never start a subscription snapshot until further traffic arrives.
+     */
+    @Test(dataProvider = "snapshotExists")
+    public void testMaxReadPositionMovedForwardForMessagesPublishedDuringRecovery(boolean snapshotExists)
+            throws Exception {
+        String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp-tb-recovery");
+        CompletableFuture<Position> recoverFuture = new CompletableFuture<>();
+        TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+        pulsar.setTransactionBufferProvider(originTopic -> {
+            AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+            when(processor.recoverFromSnapshot()).thenReturn(recoverFuture);
+            when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+            return new TopicTransactionBuffer(
+                    (PersistentTopic) originTopic, processor, AbortedTxnProcessor.SnapshotType.Single);
+        });
+        try {
+            @Cleanup
+            Producer<byte[]> producer = pulsarClient.newProducer().topic(tpName).create();
+            // The transaction buffer is stuck in the recovering state until recoverFuture completes, so these
+            // messages are published while it is still recovering.
+            for (int i = 0; i < 3; i++) {
+                producer.send(("msg-" + i).getBytes(StandardCharsets.UTF_8));
+            }
+            PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
+                    .getTopicIfExists(tpName).get().orElseThrow();
+            assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+                    "Publishes during recovery are not expected to move the max read position forward");
+
+            // Let the recovery finish, either replaying from a snapshot position or without any snapshot.
+            recoverFuture.complete(snapshotExists ? PositionFactory.EARLIEST : null);
+
+            Awaitility.await().untilAsserted(() -> {
+                assertTrue(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp() > 0,
+                        "Completed recovery should move the max read position forward for the messages"
+                                + " published during recovery");
+                assertEquals(persistentTopic.getTransactionBuffer().getMaxReadPosition(),
+                        persistentTopic.getManagedLedger().getLastConfirmedEntry());
+            });
+        } finally {
+            pulsar.setTransactionBufferProvider(originalProvider);
+        }
+    }
+
+    /**
+     * When nothing is published while the transaction buffer recovers, completing the recovery must not trigger
+     * the maxReadPositionMovedForward callback, so an idle topic isn't mistaken for one with new data.
+     */
+    @Test
+    public void testMaxReadPositionNotMovedForwardWhenNothingPublishedDuringRecovery() throws Exception {
+        String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp-tb-recovery-idle");
+        CompletableFuture<Position> recoverFuture = new CompletableFuture<>();
+        TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+        pulsar.setTransactionBufferProvider(originTopic -> {
+            AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+            when(processor.recoverFromSnapshot()).thenReturn(recoverFuture);
+            when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+            return new TopicTransactionBuffer(
+                    (PersistentTopic) originTopic, processor, AbortedTxnProcessor.SnapshotType.Single);
+        });
+        try {
+            PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
+                    .getTopic(tpName, true).get().orElseThrow();
+            recoverFuture.complete(null);
+            Awaitility.await().untilAsserted(() ->
+                    assertEquals(persistentTopic.getTransactionBuffer().getMaxReadPosition(),
+                            persistentTopic.getManagedLedger().getLastConfirmedEntry()));
+            assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+                    "Recovery of an idle topic should not move the max read position forward");
+        } finally {
+            pulsar.setTransactionBufferProvider(originalProvider);
+        }
+    }
+}