(pulsar) 01/12: [fix][broker] Ignore replicated subscription updates for the compaction subscription (#27032)

[lh...@apache.org]

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

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

commit 650885b36a22734ac12ba0de318c3b42a1ef1743
Author: kaminski-dev <39...@users.noreply.github.com>
AuthorDate: Thu Oct 22 00:38:22 2026 +0200

    [fix][broker] Ignore replicated subscription updates for the compaction subscription (#27032)
    
    (cherry picked from commit 5511b5a441adfe38e5c64a1726e39a10ba42cf35)
---
 .../ReplicatedSubscriptionsController.java         |  8 ++++++
 .../ReplicatedSubscriptionsControllerTest.java     | 32 +++++++++++++++++++++-
 2 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java
index bf3a6513146a..03c43da88c0d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java
@@ -198,6 +198,14 @@ public class ReplicatedSubscriptionsController implements AutoCloseable, Topic.P
     }
 
     private void receiveSubscriptionUpdated(ReplicatedSubscriptionsUpdate update) {
+        if (PersistentTopic.isCompactionSubscription(update.getSubscriptionName())) {
+            // The compactor owns this cursor: it may only move together with the compacted ledger
+            log.debug()
+                    .attr("subscriptionName", update.getSubscriptionName())
+                    .log("Ignoring replicated subscription update for the compaction subscription");
+            return;
+        }
+
         MarkersMessageIdData updatedMessageId = null;
         for (int i = 0, size = update.getClustersCount(); i < size; i++) {
             ClusterMessageId cmid = update.getClusterAt(i);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java
index 09b86418b253..320bf11177fe 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service.persistent;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -68,6 +69,7 @@ import org.apache.pulsar.common.policies.data.BacklogQuota;
 import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.protocol.Markers;
+import org.apache.pulsar.compaction.Compactor;
 import org.apache.pulsar.utils.TestLogAppender;
 import org.mockito.Mockito;
 import org.testng.Assert;
@@ -155,9 +157,37 @@ public class ReplicatedSubscriptionsControllerTest {
         return subscription;
     }
 
+    @Test
+    public void testReplicatedSubscriptionUpdateForCompactionSubscriptionIsIgnored() {
+        PersistentTopic topic = mockTopicForUpdates();
+        PulsarCompactorSubscription compactorSubscription = mock(PulsarCompactorSubscription.class);
+        // PulsarCompactorSubscription rejects acks without the compacted ledger property synchronously
+        when(compactorSubscription.acknowledgeMessageAsync(any(), any(), any()))
+                .thenThrow(new IllegalArgumentException());
+        when(topic.getSubscription(Compactor.COMPACTION_SUBSCRIPTION)).thenReturn(compactorSubscription);
+
+        receiveUpdate(topic, Compactor.COMPACTION_SUBSCRIPTION, 1, 2);
+
+        verify(compactorSubscription, never()).acknowledgeMessageAsync(any(), any(), any());
+    }
+
+    @Test
+    public void testReplicatedSubscriptionUpdateDoesNotCreateCompactionSubscription() {
+        PersistentTopic topic = mockTopicForUpdates();
+        when(topic.getSubscription(Compactor.COMPACTION_SUBSCRIPTION)).thenReturn(null);
+
+        receiveUpdate(topic, Compactor.COMPACTION_SUBSCRIPTION, 1, 2);
+
+        verify(topic, never()).createSubscription(any(), any(), anyBoolean(), any());
+    }
+
     private static void receiveUpdate(PersistentTopic topic, long ledgerId, long entryId) {
+        receiveUpdate(topic, "sub", ledgerId, entryId);
+    }
+
+    private static void receiveUpdate(PersistentTopic topic, String subscriptionName, long ledgerId, long entryId) {
         ReplicatedSubscriptionsController controller = new ReplicatedSubscriptionsController(topic, "local");
-        ByteBuf marker = Markers.newReplicatedSubscriptionsUpdate("sub",
+        ByteBuf marker = Markers.newReplicatedSubscriptionsUpdate(subscriptionName,
                 Map.of("local", new MarkersMessageIdData().setLedgerId(ledgerId).setEntryId(entryId)));
         try {
             Commands.skipMessageMetadata(marker);