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/10/01 18:10:51 UTC
(pulsar) 01/38: [improve][broker] Skip unused replicated subscription timestamp updates (#26655)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.2-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f1e85d0fccd7e398b8c63225abcbc9406381c17a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sun Sep 20 01:41:05 2026 +0300
[improve][broker] Skip unused replicated subscription timestamp updates (#26655)
(cherry picked from commit 3947c95f0c66c2c1217b6a60eb0a719ca1037fd2)
---
microbench/pom.xml | 5 +
.../ReplicatedSubscriptionTimestampBenchmark.java | 135 +++++++++++++++++++++
.../broker/service/persistent/package-info.java | 22 ++++
.../broker/service/persistent/PersistentTopic.java | 19 ++-
.../buffer/impl/TopicTransactionBuffer.java | 8 +-
.../pulsar/broker/service/PersistentTopicTest.java | 3 +-
.../ReplicatedSubscriptionsControllerTest.java | 119 ++++++++++++++++++
.../broker/transaction/TransactionProduceTest.java | 18 +--
.../impl/TopicTransactionBufferRecoveryTest.java | 15 +--
9 files changed, 321 insertions(+), 23 deletions(-)
diff --git a/microbench/pom.xml b/microbench/pom.xml
index b8ae7cf7cb63..459f75a32f20 100644
--- a/microbench/pom.xml
+++ b/microbench/pom.xml
@@ -96,6 +96,11 @@
<artifactId>pulsar-broker</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionTimestampBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionTimestampBenchmark.java
new file mode 100644
index 000000000000..2458151fd7b2
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionTimestampBenchmark.java
@@ -0,0 +1,135 @@
+/*
+ * 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.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.time.Clock;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+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.stats.OpenTelemetryReplicatedSubscriptionStats;
+import org.apache.pulsar.common.policies.data.BacklogQuota;
+import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+/** Measures the publish-path max-read-position callback with and without replicated-subscription snapshots. */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class ReplicatedSubscriptionTimestampBenchmark {
+
+ @Param({"false", "true"})
+ public boolean controllerEnabled;
+
+ private PersistentTopic topic;
+ private Position oldPosition;
+ private Position newPosition;
+
+ @Setup
+ @SuppressWarnings("unchecked")
+ public void setup() {
+ PulsarService pulsar = mock(PulsarService.class);
+ ScheduledExecutorService executor = mock(ScheduledExecutorService.class);
+ @SuppressWarnings("rawtypes")
+ ScheduledFuture timer = mock(ScheduledFuture.class);
+ ServiceConfiguration config = new ServiceConfiguration();
+ config.setClusterName("local");
+ config.setEnableReplicatedSubscriptions(true);
+ MonotonicClock monotonicClock = System::nanoTime;
+ BacklogQuotaManager backlogQuotaManager = mock(BacklogQuotaManager.class);
+ when(backlogQuotaManager.getDefaultQuota()).thenReturn(BacklogQuotaImpl.builder()
+ .limitSize(0)
+ .limitTime(0)
+ .retentionPolicy(BacklogQuota.RetentionPolicy.producer_request_hold)
+ .build());
+ BrokerService brokerService = mock(BrokerService.class);
+ when(brokerService.getClock()).thenReturn(Clock.systemUTC());
+ when(brokerService.pulsar()).thenReturn(pulsar);
+ when(brokerService.getPulsar()).thenReturn(pulsar);
+ when(brokerService.getBacklogQuotaManager()).thenReturn(backlogQuotaManager);
+ when(pulsar.getExecutor()).thenReturn(executor);
+ when(pulsar.getConfiguration()).thenReturn(config);
+ when(pulsar.getOpenTelemetryReplicatedSubscriptionStats())
+ .thenReturn(mock(OpenTelemetryReplicatedSubscriptionStats.class));
+ when(pulsar.getMonotonicClock()).thenReturn(monotonicClock);
+ when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenReturn(timer);
+
+ TestPersistentTopic testTopic = new TestPersistentTopic("persistent://public/default/benchmark",
+ brokerService, mock(ManagedLedger.class), mock(MessageDeduplication.class));
+ if (controllerEnabled) {
+ testTopic.setReplicationClusters(List.of("local", "remote"));
+ PersistentSubscription subscription = mock(PersistentSubscription.class);
+ when(subscription.isReplicated()).thenReturn(true);
+ testTopic.getSubscriptions().put("sub", subscription);
+ testTopic.checkReplicatedSubscriptionControllerState();
+ }
+ topic = testTopic;
+ oldPosition = PositionFactory.create(1, 1);
+ newPosition = PositionFactory.create(1, 2);
+ }
+
+ @Benchmark
+ public long maxReadPositionMovedForward() {
+ topic.getMaxReadPositionCallBack().maxReadPositionMovedForward(oldPosition, newPosition);
+ return topic.getLastMaxReadPositionMovedForwardTimestamp();
+ }
+
+ @TearDown
+ public void tearDown() {
+ topic.getReplicatedSubscriptionController().ifPresent(ReplicatedSubscriptionsController::close);
+ }
+
+ private static final class TestPersistentTopic extends PersistentTopic {
+ TestPersistentTopic(String topic, BrokerService brokerService, ManagedLedger ledger,
+ MessageDeduplication messageDeduplication) {
+ super(topic, brokerService, ledger, messageDeduplication);
+ }
+
+ void setReplicationClusters(List<String> clusters) {
+ topicPolicies.getReplicationClusters().updateBrokerValue(clusters);
+ }
+ }
+}
diff --git a/microbench/src/main/java/org/apache/pulsar/broker/service/persistent/package-info.java b/microbench/src/main/java/org/apache/pulsar/broker/service/persistent/package-info.java
new file mode 100644
index 000000000000..77c1874b5963
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/broker/service/persistent/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * Benchmarks for persistent broker service internals.
+ */
+package org.apache.pulsar.broker.service.persistent;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index df307e651e06..7e6ca2b198f1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -290,7 +290,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private final TopicTransactionBuffer.MaxReadPositionCallBack maxReadPositionCallBack =
(oldPosition, newPosition) -> updateMaxReadPositionMovedForwardTimestamp();
- // Record the last time max read position is moved forward, unless it's a marker message.
+ // Record the last time max read position moved forward while replicated-subscription snapshots are active.
+ // Controller creation seeds this timestamp so that data published before activation is included in a snapshot.
@Getter
private volatile long lastMaxReadPositionMovedForwardTimestamp = 0;
@@ -746,9 +747,20 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private void updateMaxReadPositionMovedForwardTimestamp() {
+ if (replicatedSubscriptionsController.isEmpty()) {
+ return;
+ }
lastMaxReadPositionMovedForwardTimestamp = Clock.systemUTC().millis();
}
+ private void seedMaxReadPositionMovedForwardTimestamp() {
+ // An empty topic has no data to snapshot. Recheck after publishing the controller reference to cover
+ // entries added during construction, while publishes after activation update the timestamp themselves.
+ if (ledger.getNumberOfEntries() > 0) {
+ lastMaxReadPositionMovedForwardTimestamp = Clock.systemUTC().millis();
+ }
+ }
+
@Override
public void addComplete(Position pos, ByteBuf entryData, Object ctx) {
PublishContext publishContext = (PublishContext) ctx;
@@ -4823,8 +4835,13 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
if (shouldBeEnabled && !isCurrentlyEnabled && isEnableReplicatedSubscriptions && replicationEnabled) {
log.info("[{}] Enabling replicated subscriptions controller", topic);
+ // Force the new controller's first snapshot to cover messages published before it was enabled.
+ // Seed before construction because the controller schedules its first snapshot from its constructor.
+ seedMaxReadPositionMovedForwardTimestamp();
replicatedSubscriptionsController = Optional.of(new ReplicatedSubscriptionsController(this,
brokerService.pulsar().getConfiguration().getClusterName()));
+ // Cover a max-read-position advance racing with construction, before the controller became visible.
+ seedMaxReadPositionMovedForwardTimestamp();
} else if (isCurrentlyEnabled && (!shouldBeEnabled || !isEnableReplicatedSubscriptions
|| !replicationEnabled)) {
log.info("[{}] Disabled replicated subscriptions controller", topic);
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 fb8314a420af..1b4eb3d32479 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
@@ -648,7 +648,8 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
/**
* update the max read position. if the new position is greater than the current max read position,
* we will trigger the callback, unless the disableCallback is true.
- * Currently, we only use the callback to update the lastMaxReadPositionMovedForwardTimestamp.
+ * Currently, we only use the callback to update the lastMaxReadPositionMovedForwardTimestamp while a
+ * replicated-subscription controller is active.
* For non-transactional production, some marker messages will be sent to the topic, in which case we don't need
* to trigger the callback.
* @param newPosition new max read position to update.
@@ -671,8 +672,9 @@ 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.
+ * {@link PersistentTopic} uses that callback to maintain lastMaxReadPositionMovedForwardTimestamp while a
+ * ReplicatedSubscriptionsController is active. The controller relies on it to detect new data when deciding
+ * whether to start a snapshot.
* Must be called while synchronized on this transaction buffer.
*/
private void updateMaxReadPositionAfterRecovery() {
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 055bc3a5d1cf..76c82ace99d1 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
@@ -354,7 +354,8 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
topic.publishMessage(payload, publishContext);
assertTrue(latch.await(1, TimeUnit.SECONDS));
- assertTrue(topic.getLastMaxReadPositionMovedForwardTimestamp() > lastMaxReadPositionMovedForwardTimestamp);
+ assertEquals(topic.getLastMaxReadPositionMovedForwardTimestamp(), lastMaxReadPositionMovedForwardTimestamp,
+ "Topics without a replicated-subscription controller should not maintain the snapshot timestamp");
}
@Test
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 cd1fb44be208..b334cc686e47 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
@@ -66,6 +66,125 @@ import org.testng.annotations.Test;
@Test(groups = "broker-replication")
public class ReplicatedSubscriptionsControllerTest {
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testMaxReadPositionTimestampFollowsControllerLifecycle() {
+ PulsarService pulsar = mock(PulsarService.class);
+ ScheduledExecutorService executor = mock(ScheduledExecutorService.class);
+ @SuppressWarnings("rawtypes")
+ ScheduledFuture timer = mock(ScheduledFuture.class);
+ ServiceConfiguration config = new ServiceConfiguration();
+ config.setClusterName("local");
+ config.setEnableReplicatedSubscriptions(true);
+ OpenTelemetryReplicatedSubscriptionStats stats = mock(OpenTelemetryReplicatedSubscriptionStats.class);
+ MonotonicClock monotonicClock = System::nanoTime;
+ BacklogQuotaManager backlogQuotaManager = mock(BacklogQuotaManager.class);
+ when(backlogQuotaManager.getDefaultQuota()).thenReturn(BacklogQuotaImpl.builder()
+ .limitSize(0)
+ .limitTime(0)
+ .retentionPolicy(BacklogQuota.RetentionPolicy.producer_request_hold)
+ .build());
+ BrokerService brokerService = mock(BrokerService.class);
+ ManagedLedger ledger = mock(ManagedLedger.class);
+ MessageDeduplication messageDeduplication = mock(MessageDeduplication.class);
+
+ when(brokerService.getClock()).thenReturn(Clock.systemUTC());
+ when(brokerService.pulsar()).thenReturn(pulsar);
+ when(brokerService.getPulsar()).thenReturn(pulsar);
+ when(brokerService.getBacklogQuotaManager()).thenReturn(backlogQuotaManager);
+ when(pulsar.getExecutor()).thenReturn(executor);
+ when(pulsar.getConfiguration()).thenReturn(config);
+ when(pulsar.getOpenTelemetryReplicatedSubscriptionStats()).thenReturn(stats);
+ when(pulsar.getMonotonicClock()).thenReturn(monotonicClock);
+ when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenReturn(timer);
+
+ TestPersistentTopic topic =
+ new TestPersistentTopic("persistent://public/default/t1", brokerService, ledger, messageDeduplication);
+ topic.setReplicationClusters(List.of("local", "remote"));
+ PersistentSubscription subscription = mock(PersistentSubscription.class);
+ when(subscription.isReplicated()).thenReturn(true);
+
+ topic.getMaxReadPositionCallBack().maxReadPositionMovedForward(null, PositionFactory.create(1, 1));
+ Assert.assertEquals(topic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+ "Topics without a controller should not maintain the snapshot timestamp");
+
+ topic.getSubscriptions().put("sub", subscription);
+ topic.checkReplicatedSubscriptionControllerState();
+ try {
+ Assert.assertTrue(topic.getReplicatedSubscriptionController().isPresent());
+ Assert.assertEquals(topic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+ "Enabling the controller on an empty topic must not trigger a snapshot");
+
+ topic.getSubscriptions().clear();
+ topic.checkReplicatedSubscriptionControllerState();
+ when(ledger.getNumberOfEntries()).thenReturn(1L);
+ topic.getMaxReadPositionCallBack().maxReadPositionMovedForward(
+ PositionFactory.create(1, 1), PositionFactory.create(1, 2));
+ Assert.assertEquals(topic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+ "Publishing before controller activation must leave the timestamp disabled");
+
+ topic.getSubscriptions().put("sub", subscription);
+ topic.checkReplicatedSubscriptionControllerState();
+ long enabledTimestamp = topic.getLastMaxReadPositionMovedForwardTimestamp();
+ Assert.assertTrue(enabledTimestamp > 0,
+ "Controller creation must seed the timestamp for data published before activation");
+
+ waitUntilNextMillisecond(enabledTimestamp);
+ topic.getMaxReadPositionCallBack().maxReadPositionMovedForward(
+ PositionFactory.create(1, 1), PositionFactory.create(1, 2));
+ long updatedTimestamp = topic.getLastMaxReadPositionMovedForwardTimestamp();
+ Assert.assertTrue(updatedTimestamp > enabledTimestamp);
+
+ topic.getSubscriptions().clear();
+ topic.checkReplicatedSubscriptionControllerState();
+ Assert.assertTrue(topic.getReplicatedSubscriptionController().isEmpty());
+
+ waitUntilNextMillisecond(updatedTimestamp);
+ topic.getMaxReadPositionCallBack().maxReadPositionMovedForward(
+ PositionFactory.create(1, 2), PositionFactory.create(1, 3));
+ Assert.assertEquals(topic.getLastMaxReadPositionMovedForwardTimestamp(), updatedTimestamp,
+ "Timestamp should remain unchanged while the controller is disabled");
+
+ waitUntilNextMillisecond(updatedTimestamp);
+ when(ledger.getNumberOfEntries()).thenReturn(0L);
+ when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> {
+ // Publish after the initial seed check but before the controller reference is visible.
+ when(ledger.getNumberOfEntries()).thenReturn(1L);
+ topic.getMaxReadPositionCallBack().maxReadPositionMovedForward(
+ PositionFactory.create(1, 3), PositionFactory.create(1, 4));
+ Assert.assertEquals(topic.getLastMaxReadPositionMovedForwardTimestamp(), updatedTimestamp,
+ "Publishes racing with construction still see the controller as disabled");
+ return timer;
+ });
+ topic.getSubscriptions().put("sub", subscription);
+ topic.checkReplicatedSubscriptionControllerState();
+ Assert.assertTrue(topic.getReplicatedSubscriptionController().isPresent());
+ Assert.assertTrue(topic.getLastMaxReadPositionMovedForwardTimestamp() > updatedTimestamp,
+ "Re-enabling the controller must cover entries added during its construction");
+ } finally {
+ topic.getReplicatedSubscriptionController().ifPresent(ReplicatedSubscriptionsController::close);
+ }
+ }
+
+ private static void waitUntilNextMillisecond(long timestamp) {
+ while (Clock.systemUTC().millis() <= timestamp) {
+ Thread.onSpinWait();
+ }
+ }
+
+ private static final class TestPersistentTopic extends PersistentTopic {
+ TestPersistentTopic(String topic, BrokerService brokerService, ManagedLedger ledger,
+ MessageDeduplication messageDeduplication) {
+ super(topic, brokerService, ledger, messageDeduplication);
+ }
+
+ void setReplicationClusters(List<String> clusters) {
+ topicPolicies.getReplicationClusters().updateBrokerValue(clusters);
+ }
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testFinalSnapshotMarkerPublishFailureKeepsSnapshotPending() {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
index 5dc71db4d732..9d318bcb346b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionProduceTest.java
@@ -19,7 +19,7 @@
package org.apache.pulsar.broker.transaction;
import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.assertEquals;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashSet;
@@ -242,8 +242,8 @@ public class TransactionProduceTest extends TransactionTestBase {
}
@Test
- public void testUpdateLastMaxReadPositionMovedForwardTimestampForTransactionalPublish() throws Exception {
- final String topic = NAMESPACE1 + "/testUpdateLastMaxReadPositionMovedForwardTimestampForTransactionalPublish";
+ public void testSkipMaxReadPositionTimestampWithoutReplicatedSubscriptions() throws Exception {
+ final String topic = NAMESPACE1 + "/testSkipMaxReadPositionTimestampWithoutReplicatedSubscriptions";
PulsarClient pulsarClient = this.pulsarClient;
Transaction txn = pulsarClient.newTransaction()
.withTransactionTimeout(5, TimeUnit.SECONDS)
@@ -257,15 +257,15 @@ public class TransactionProduceTest extends TransactionTestBase {
PersistentTopic persistentTopic = getTopic(topic);
long lastMaxReadPositionMovedForwardTimestamp = persistentTopic.getLastMaxReadPositionMovedForwardTimestamp();
- // transactional publish will not update lastMaxReadPositionMovedForwardTimestamp
+ // A transactional publish does not move the max read position until the transaction is resolved.
producer.newMessage(txn).value("hello world".getBytes()).send();
- assertTrue(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp()
- == lastMaxReadPositionMovedForwardTimestamp);
+ assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(),
+ lastMaxReadPositionMovedForwardTimestamp);
- // commit transaction will update lastMaxReadPositionMovedForwardTimestamp
+ // Committing moves the max read position, but an ordinary topic does not need the snapshot timestamp.
txn.commit().get();
- assertTrue(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp()
- > lastMaxReadPositionMovedForwardTimestamp);
+ assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(),
+ lastMaxReadPositionMovedForwardTimestamp);
}
private PersistentTopic getTopic(String topic) throws ExecutionException, InterruptedException {
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
index 16bdfacc6c05..6a66fc51dc07 100644
--- 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
@@ -31,7 +31,6 @@ import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -159,11 +158,9 @@ public class TopicTransactionBufferRecoveryTest extends ProducerConsumerBase {
}
/**
- * 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.
+ * While the transaction buffer is recovering, normal publishes don't move the max read position. Recovery
+ * must account for those messages when it completes. Without a replicated-subscription controller, handling
+ * the resulting callback must leave the snapshot timestamp disabled.
*/
@Test(dataProvider = "snapshotExists")
public void testMaxReadPositionMovedForwardForMessagesPublishedDuringRecovery(boolean snapshotExists)
@@ -195,11 +192,11 @@ public class TopicTransactionBufferRecoveryTest extends ProducerConsumerBase {
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());
+ assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+ "Topics without a replicated-subscription controller should not maintain"
+ + " the snapshot timestamp");
});
} finally {
pulsar.setTransactionBufferProvider(originalProvider);