You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/07/25 15:28:30 UTC
(pulsar) 02/02: [fix][broker] Prevent completing replicated snapshot before marker publish (#26119)
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 14afc4db04f2ac04257fcb99dc9169f2ec520070
Author: sinan liu <li...@gmail.com>
AuthorDate: Sat Jul 25 23:18:20 2026 +0800
[fix][broker] Prevent completing replicated snapshot before marker publish (#26119)
(cherry picked from commit bc0e8b557915f00538200185858b10a161ad8a03)
---
.../ReplicatedSubscriptionsController.java | 52 ++++++--
.../ReplicatedSubscriptionsSnapshotBuilder.java | 25 +++-
.../ReplicatedSubscriptionsControllerTest.java | 140 +++++++++++++++++++++
...ReplicatedSubscriptionsSnapshotBuilderTest.java | 9 ++
4 files changed, 214 insertions(+), 12 deletions(-)
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 f19570afcd6e..d14cd1d2fe73 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
@@ -30,6 +30,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledFuture;
@@ -289,6 +290,10 @@ public class ReplicatedSubscriptionsController implements AutoCloseable, Topic.P
while (it.hasNext()) {
Map.Entry<String, ReplicatedSubscriptionsSnapshotBuilder> entry = it.next();
if (entry.getValue().isTimedOut()) {
+ if (!pendingSnapshots.remove(entry.getKey(), entry.getValue())) {
+ continue;
+ }
+
if (log.isDebugEnabled()) {
log.debug("[{}] Snapshot creation timed out for {}", topic.getName(), entry.getKey());
}
@@ -297,23 +302,23 @@ public class ReplicatedSubscriptionsController implements AutoCloseable, Topic.P
timedoutSnapshotsMetric.inc();
var latencyMillis = entry.getValue().getDurationMillis();
stats.recordSnapshotTimedOut(latencyMillis);
- it.remove();
}
}
}
void snapshotCompleted(String snapshotId) {
ReplicatedSubscriptionsSnapshotBuilder snapshot = pendingSnapshots.remove(snapshotId);
- lastCompletedSnapshotId = snapshotId;
+ if (snapshot == null) {
+ return;
+ }
- if (snapshot != null) {
- lastCompletedSnapshotStartTime = snapshot.getStartTimeMillis();
+ lastCompletedSnapshotId = snapshotId;
+ lastCompletedSnapshotStartTime = snapshot.getStartTimeMillis();
- pendingSnapshotsMetric.dec();
- var latencyMillis = snapshot.getDurationMillis();
- ReplicatedSubscriptionsSnapshotBuilder.SNAPSHOT_METRIC.observe(latencyMillis);
- stats.recordSnapshotCompleted(latencyMillis);
- }
+ pendingSnapshotsMetric.dec();
+ var latencyMillis = snapshot.getDurationMillis();
+ ReplicatedSubscriptionsSnapshotBuilder.SNAPSHOT_METRIC.observe(latencyMillis);
+ stats.recordSnapshotCompleted(latencyMillis);
}
void writeMarker(ByteBuf marker) {
@@ -324,6 +329,35 @@ public class ReplicatedSubscriptionsController implements AutoCloseable, Topic.P
}
}
+ CompletableFuture<Position> writeMarkerAndGetPosition(ByteBuf marker) {
+ CompletableFuture<Position> future = new CompletableFuture<>();
+ Topic.PublishContext publishContext = new Topic.PublishContext() {
+ @Override
+ public void completed(Exception e, long ledgerId, long entryId) {
+ ReplicatedSubscriptionsController.this.completed(e, ledgerId, entryId);
+ if (e != null) {
+ future.completeExceptionally(e);
+ } else {
+ future.complete(PositionFactory.create(ledgerId, entryId));
+ }
+ }
+
+ @Override
+ public boolean isMarkerMessage() {
+ return true;
+ }
+ };
+
+ try {
+ topic.publishMessage(marker, publishContext);
+ } catch (Exception e) {
+ publishContext.completed(e, -1, -1);
+ } finally {
+ marker.release();
+ }
+ return future;
+ }
+
/**
* From Topic.PublishContext.
*/
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilder.java
index e08b549f8aec..df66794d751f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilder.java
@@ -47,6 +47,7 @@ public class ReplicatedSubscriptionsSnapshotBuilder {
private final boolean needTwoRounds;
private boolean firstRoundComplete;
+ private boolean finalSnapshotMarkerPending;
private long startTimeMillis;
private final long timeoutMillis;
@@ -116,15 +117,33 @@ public class ReplicatedSubscriptionsSnapshotBuilder {
return;
}
+ if (finalSnapshotMarkerPending) {
+ return;
+ }
+
+ finalSnapshotMarkerPending = true;
if (log.isDebugEnabled()) {
log.debug("[{}] Snapshot is complete {}", controller.topic().getName(), snapshotId);
}
// Snapshot is now complete, store it in the local topic
Position p = position;
- controller.writeMarker(
+ controller.writeMarkerAndGetPosition(
Markers.newReplicatedSubscriptionsSnapshot(snapshotId, controller.localCluster(),
- p.getLedgerId(), p.getEntryId(), responses));
- controller.snapshotCompleted(snapshotId);
+ p.getLedgerId(), p.getEntryId(), responses))
+ .whenComplete((ignored, exception) -> {
+ if (exception != null) {
+ synchronized (ReplicatedSubscriptionsSnapshotBuilder.this) {
+ finalSnapshotMarkerPending = false;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Failed to publish completed snapshot marker {}",
+ controller.topic().getName(), snapshotId, exception);
+ }
+ return;
+ }
+
+ controller.snapshotCompleted(snapshotId);
+ });
}
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 92a15ad26055..cd1fb44be208 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
@@ -24,11 +24,14 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import io.netty.buffer.ByteBuf;
import java.time.Clock;
import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -37,6 +40,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback;
import org.apache.bookkeeper.mledger.ManagedLedger;
@@ -47,6 +52,7 @@ import org.apache.pulsar.broker.qos.MonotonicClock;
import org.apache.pulsar.broker.service.BacklogQuotaManager;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Replicator;
+import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.stats.OpenTelemetryReplicatedSubscriptionStats;
import org.apache.pulsar.common.api.proto.MarkerType;
import org.apache.pulsar.common.policies.data.BacklogQuota;
@@ -60,6 +66,140 @@ import org.testng.annotations.Test;
@Test(groups = "broker-replication")
public class ReplicatedSubscriptionsControllerTest {
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testFinalSnapshotMarkerPublishFailureKeepsSnapshotPending() {
+ PulsarService pulsar = mock(PulsarService.class);
+ ScheduledExecutorService executor = mock(ScheduledExecutorService.class);
+ @SuppressWarnings("rawtypes")
+ ScheduledFuture timer = mock(ScheduledFuture.class);
+ ServiceConfiguration config = new ServiceConfiguration();
+ config.setReplicatedSubscriptionsSnapshotFrequencyMillis(60_000);
+ config.setReplicatedSubscriptionsSnapshotTimeoutSeconds(3);
+ OpenTelemetryReplicatedSubscriptionStats stats = mock(OpenTelemetryReplicatedSubscriptionStats.class);
+ BrokerService brokerService = mock(BrokerService.class);
+ PersistentTopic topic = mock(PersistentTopic.class);
+ Replicator replicator = mock(Replicator.class);
+ List<Topic.PublishContext> publishContexts = new ArrayList<>();
+ AtomicReference<Runnable> scheduledSnapshotTask = new AtomicReference<>();
+
+ when(topic.getName()).thenReturn("persistent://public/default/t1");
+ when(topic.getBrokerService()).thenReturn(brokerService);
+ when(topic.getLastMaxReadPositionMovedForwardTimestamp()).thenReturn(1L);
+ when(topic.getReplicators()).thenReturn(Map.of("remote", replicator));
+ when(replicator.isConnected()).thenReturn(true);
+ when(brokerService.pulsar()).thenReturn(pulsar);
+ when(pulsar.getExecutor()).thenReturn(executor);
+ when(pulsar.getConfiguration()).thenReturn(config);
+ when(pulsar.getOpenTelemetryReplicatedSubscriptionStats()).thenReturn(stats);
+ when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> {
+ scheduledSnapshotTask.set(invocation.getArgument(0, Runnable.class));
+ return timer;
+ });
+ doAnswer(invocation -> {
+ publishContexts.add(invocation.getArgument(1, Topic.PublishContext.class));
+ return null;
+ }).when(topic).publishMessage(any(ByteBuf.class), any(Topic.PublishContext.class));
+
+ ReplicatedSubscriptionsController controller = new ReplicatedSubscriptionsController(topic, "local");
+ String snapshotId = null;
+ try {
+ Assert.assertNotNull(scheduledSnapshotTask.get());
+ scheduledSnapshotTask.get().run();
+ Assert.assertEquals(controller.pendingSnapshots().size(), 1);
+ snapshotId = controller.pendingSnapshots().keySet().iterator().next();
+ Assert.assertEquals(publishContexts.size(), 1);
+
+ ByteBuf responseMarker = Markers.newReplicatedSubscriptionsSnapshotResponse(snapshotId, "local",
+ "remote", 11, 11);
+ try {
+ Commands.skipMessageMetadata(responseMarker);
+ controller.receivedReplicatedSubscriptionMarker(PositionFactory.create(1, 1),
+ MarkerType.REPLICATED_SUBSCRIPTION_SNAPSHOT_RESPONSE_VALUE, responseMarker);
+ } finally {
+ responseMarker.release();
+ }
+ Assert.assertEquals(publishContexts.size(), 2);
+
+ publishContexts.get(1).completed(new RuntimeException("final marker publish failed"), -1, -1);
+
+ Assert.assertTrue(controller.pendingSnapshots().containsKey(snapshotId),
+ "Snapshot should remain pending after the final snapshot marker publish fails");
+ Assert.assertFalse(controller.getLastCompletedSnapshotId().isPresent(),
+ "Failed final snapshot marker publish should not update the last completed snapshot id");
+ } finally {
+ if (snapshotId != null && controller.pendingSnapshots().containsKey(snapshotId)) {
+ controller.snapshotCompleted(snapshotId);
+ }
+ controller.close();
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTimeoutCleanupDoesNotRecordTimeoutWhenSnapshotCompletedConcurrently() {
+ PulsarService pulsar = mock(PulsarService.class);
+ ScheduledExecutorService executor = mock(ScheduledExecutorService.class);
+ @SuppressWarnings("rawtypes")
+ ScheduledFuture timer = mock(ScheduledFuture.class);
+ ServiceConfiguration config = new ServiceConfiguration();
+ config.setReplicatedSubscriptionsSnapshotFrequencyMillis(60_000);
+ config.setReplicatedSubscriptionsSnapshotTimeoutSeconds(3);
+ OpenTelemetryReplicatedSubscriptionStats stats = mock(OpenTelemetryReplicatedSubscriptionStats.class);
+ BrokerService brokerService = mock(BrokerService.class);
+ PersistentTopic topic = mock(PersistentTopic.class);
+ Replicator replicator = mock(Replicator.class);
+ AtomicReference<Runnable> scheduledSnapshotTask = new AtomicReference<>();
+
+ when(topic.getName()).thenReturn("persistent://public/default/t1");
+ when(topic.getBrokerService()).thenReturn(brokerService);
+ when(topic.getLastMaxReadPositionMovedForwardTimestamp()).thenReturn(1L, 1L, 0L, 0L);
+ when(topic.getReplicators()).thenReturn(Map.of("remote", replicator));
+ when(replicator.isConnected()).thenReturn(true);
+ when(brokerService.pulsar()).thenReturn(pulsar);
+ when(pulsar.getExecutor()).thenReturn(executor);
+ when(pulsar.getConfiguration()).thenReturn(config);
+ when(pulsar.getOpenTelemetryReplicatedSubscriptionStats()).thenReturn(stats);
+ when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> {
+ scheduledSnapshotTask.set(invocation.getArgument(0, Runnable.class));
+ return timer;
+ });
+
+ ReplicatedSubscriptionsController controller = new ReplicatedSubscriptionsController(topic, "local");
+ try {
+ Assert.assertNotNull(scheduledSnapshotTask.get());
+ scheduledSnapshotTask.get().run();
+ Assert.assertEquals(controller.pendingSnapshots().size(), 1);
+ String snapshotId = controller.pendingSnapshots().keySet().iterator().next();
+ ReplicatedSubscriptionsSnapshotBuilder originalBuilder = controller.pendingSnapshots().get(snapshotId);
+
+ AtomicBoolean completedDuringTimeoutCheck = new AtomicBoolean();
+ ReplicatedSubscriptionsSnapshotBuilder completedBuilder =
+ new ReplicatedSubscriptionsSnapshotBuilder(controller, Set.of("remote"), config,
+ Clock.systemUTC()) {
+ @Override
+ boolean isTimedOut() {
+ if (completedDuringTimeoutCheck.compareAndSet(false, true)) {
+ controller.snapshotCompleted(snapshotId);
+ }
+ return true;
+ }
+ };
+ Assert.assertTrue(controller.pendingSnapshots().replace(snapshotId, originalBuilder, completedBuilder));
+
+ scheduledSnapshotTask.get().run();
+
+ Assert.assertTrue(completedDuringTimeoutCheck.get());
+ Assert.assertFalse(controller.pendingSnapshots().containsKey(snapshotId));
+ verify(stats).recordSnapshotCompleted(anyLong());
+ verify(stats, never()).recordSnapshotTimedOut(anyLong());
+ } finally {
+ controller.close();
+ }
+ }
+
@Test
public void testSnapshotRequestWhenReplicatorRemovedConcurrentlyDoesNotThrow() throws Exception {
// Use a real PersistentTopic instance so that the replicator removal happens via the production code path
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilderTest.java
index 69c11edb3469..400046136f5d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsSnapshotBuilderTest.java
@@ -31,6 +31,7 @@ import java.time.Clock;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.common.api.proto.ReplicatedSubscriptionsSnapshot;
@@ -74,6 +75,14 @@ public class ReplicatedSubscriptionsSnapshotBuilderTest {
})
.when(controller)
.writeMarker(any(ByteBuf.class));
+ doAnswer(invocation -> {
+ ByteBuf marker = invocation.getArgument(0, ByteBuf.class);
+ Commands.skipMessageMetadata(marker);
+ markers.add(marker);
+ return CompletableFuture.completedFuture(PositionFactory.create(1, 1));
+ })
+ .when(controller)
+ .writeMarkerAndGetPosition(any(ByteBuf.class));
}
@AfterMethod