(pulsar) branch branch-4.2 updated: [fix][broker] Fix geo-replication stuck after a failed publish to the remote cluster (#26002)

[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


The following commit(s) were added to refs/heads/branch-4.2 by this push:
     new 323ed2366e52 [fix][broker] Fix geo-replication stuck after a failed publish to the remote cluster (#26002)
323ed2366e52 is described below

commit 323ed2366e52ca840774386a80f5b9cd679deae8
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Tue Jun 23 04:16:10 2026 +0800

    [fix][broker] Fix geo-replication stuck after a failed publish to the remote cluster (#26002)
    
    (cherry picked from commit 9b9e67a9d710f6e4cc4d0665a6cd8dab24c45973)
---
 .../service/persistent/PersistentReplicator.java   |  3 ++
 .../PersistentReplicatorInflightTaskTest.java      | 32 ++++++++++++++++++++++
 2 files changed, 35 insertions(+)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index b125bd0306bd..565efdc456cf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -414,6 +414,9 @@ public abstract class PersistentReplicator extends AbstractReplicator
                 // cursor should be rewound since it was incremented when readMoreEntries
                 replicator.beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Failed_Publishing);
                 replicator.doRewindCursor(false);
+                // The failed send has completed from the producer queue perspective. The cursor rewind
+                // makes the entry readable again, so this in-flight task must release its permit.
+                inFlightTask.incCompletedEntries();
             } else {
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] Message persisted on remote broker", replicator.replicatorId, exception);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
index c825d49caa62..0c22b14acf74 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentReplicatorInflightTaskTest.java
@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.service.persistent;
 
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.testng.Assert.assertEquals;
@@ -43,8 +44,11 @@ import org.apache.pulsar.broker.service.AbstractReplicator;
 import org.apache.pulsar.broker.service.BrokerServiceInternalMethodInvoker;
 import org.apache.pulsar.broker.service.OneWayReplicatorTestBase;
 import org.apache.pulsar.broker.service.persistent.PersistentReplicator.InFlightTask;
+import org.apache.pulsar.broker.service.persistent.PersistentReplicator.ProducerSendCallback;
+import org.apache.pulsar.broker.service.persistent.PersistentReplicator.ReasonOfWaitForCursorRewinding;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.awaitility.Awaitility;
 import org.mockito.invocation.InvocationOnMock;
@@ -167,6 +171,34 @@ public class PersistentReplicatorInflightTaskTest extends OneWayReplicatorTestBa
         }
     }
 
+    @Test
+    public void testFailedPublishCompletesInFlightTask() throws Exception {
+        PersistentReplicator replicator = spy(getReplicator(topicName));
+        doNothing().when(replicator).beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Failed_Publishing);
+        doNothing().when(replicator).doRewindCursor(false);
+        doNothing().when(replicator).readMoreEntries();
+
+        LinkedList<InFlightTask> inFlightTasks = replicator.inFlightTasks;
+        List<InFlightTask> originalTasks = new ArrayList<>(inFlightTasks);
+        inFlightTasks.clear();
+
+        try {
+            InFlightTask task = new InFlightTask(PositionFactory.create(1, 1), 1, replicator.getReplicatorId());
+            task.setEntries(Collections.singletonList(mock(Entry.class)));
+            inFlightTasks.add(task);
+            assertEquals(replicator.getPermitsIfNoPendingRead(), 999);
+
+            ProducerSendCallback callback = ProducerSendCallback.create(replicator, mock(Entry.class), null, task);
+            callback.sendComplete(new PulsarClientException.ProducerBlockedQuotaExceededException("mocked"), null);
+
+            assertTrue(task.isDone());
+            assertEquals(replicator.getPermitsIfNoPendingRead(), 1000);
+        } finally {
+            inFlightTasks.clear();
+            inFlightTasks.addAll(originalTasks);
+        }
+    }
+
     @Test
     public void testCreateOrRecycleInFlightTaskIntoQueue() throws Exception {
         log.info("Starting testCreateOrRecycleInFlightTaskIntoQueue");