Skip to content

[fix][test] Fix flaky test ReplicatorTest.testResumptionAfterBacklogRelaxed #25950

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 2 commits into from
Jun 8, 2026
Merged

[fix][test] Fix flaky test ReplicatorTest.testResumptionAfterBacklogRelaxed #25950

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
04396cb
[fix][test] Stabilize replicator backlog quota recovery test
void-ptr974 Jun 6, 2026
a32bf3a
Merge branch 'master' into fix/replicator-backlog-quota-test-flake
lhotari Jun 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 48 additions & 14 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
Expand Down Expand Up @@ -993,7 +994,7 @@ public static Object[][] retentionPolicies() {
* @throws Exception
*/

@Test(timeOut = 60000, priority = -1, dataProvider = "retentionPolicies")
@Test(timeOut = 120000, priority = -1, dataProvider = "retentionPolicies")
public void testResumptionAfterBacklogRelaxed(RetentionPolicy policy) throws Exception {
// create a unique namespace for this test case to avoid flakiness
String namespace = newUniqueName("pulsar/testResumptionAfterBacklogRelaxed");
Expand All @@ -1020,12 +1021,20 @@ public void testResumptionAfterBacklogRelaxed(RetentionPolicy policy) throws Exc
PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService()
.getTopicReference(dest.toString()).get();
Replicator replicator = topic.getPersistentReplicator("r2");
PersistentTopic remoteTopic = (PersistentTopic) pulsar2.getBrokerService()
.getTopicReference(dest.toString()).get();

// Produce 1 message in r1. This message will be replicated immediately into r2 and it will become part of
// local backlog
producer1.produce(1);
// Produce messages in r1. These messages will be replicated immediately into r2 and become
// part of the local backlog there.
int remoteBacklogMessages = 2;
producer1.produce(remoteBacklogMessages);

Awaitility.await().untilAsserted(() -> assertEquals(replicator.computeStats().replicationBacklog, 0));
long[] remoteBacklogSize = new long[1];
Awaitility.await().untilAsserted(() -> {
remoteBacklogSize[0] = admin2.topics().getStats(dest.toString()).getBacklogSize();
assertTrue(remoteBacklogSize[0] > 1);
});
var attributes = Attributes.of(
OpenTelemetryAttributes.PULSAR_DOMAIN, dest.getDomain().value(),
OpenTelemetryAttributes.PULSAR_TENANT, dest.getTenant(),
Expand All @@ -1037,13 +1046,21 @@ public void testResumptionAfterBacklogRelaxed(RetentionPolicy policy) throws Exc
assertMetricLongSumValue(metrics, OpenTelemetryReplicatorStats.BACKLOG_COUNTER, attributes, 0);
assertMetricDoubleGaugeValue(metrics, OpenTelemetryReplicatorStats.DELAY_GAUGE, attributes, 0.0);

// Restrict backlog quota limit to 1 byte to stop replication
// Restrict the backlog quota to just below the current r2 backlog. This stops replication
// now, but leaves enough room for one small test message after the existing backlog drains.
admin1.namespaces().setBacklogQuota(namespace, BacklogQuota.builder()
.limitSize(1)
.limitSize(remoteBacklogSize[0] - 1)
.retentionPolicy(policy)
.build());

Thread.sleep((TIME_TO_CHECK_BACKLOG_QUOTA + 1) * 1000);
Awaitility.await().atMost(TIME_TO_CHECK_BACKLOG_QUOTA * 6L, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
.ignoreExceptions()
.untilAsserted(() -> {
admin2.brokers().backlogQuotaCheck();
assertTrue(remoteTopic.isSizeBacklogExceeded());
assertFalse(replicator.isConnected());
});

// Next message will not be replicated, because r2 has reached the quota
producer1.produce(1);
Expand All @@ -1057,15 +1074,32 @@ public void testResumptionAfterBacklogRelaxed(RetentionPolicy policy) throws Exc
assertMetricDoubleGaugeValue(metrics, OpenTelemetryReplicatorStats.DELAY_GAUGE, attributes,
aDouble -> assertThat(aDouble).isGreaterThanOrEqualTo(0.0));

// Consumer will now drain 1 message and the replication backlog will be cleared
consumer2.receive(1);
// Consumer will now drain the existing messages and the remote backlog quota will be relaxed.
consumer2.receive(remoteBacklogMessages);

// Wait until the 2nd message got delivered to consumer.
// Use a longer timeout because the replicator needs time to detect that the backlog quota
// has been relaxed (checked every TIME_TO_CHECK_BACKLOG_QUOTA seconds) and resume replication.
consumer2.receive(1, TIME_TO_CHECK_BACKLOG_QUOTA * 4);
// Wait until the remote topic is actually below the size quota before nudging the replicator.
// Otherwise the next producer-create attempt can race with stale backlog accounting and consume
// another reconnect backoff window.
Awaitility.await().atMost(TIME_TO_CHECK_BACKLOG_QUOTA * 6L, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
.ignoreExceptions()
.untilAsserted(() -> {
admin2.brokers().backlogQuotaCheck();
assertFalse(remoteTopic.isSizeBacklogExceeded());
});

// The replicator might already be waiting in an exponential reconnect backoff after the quota
// rejection. startProducer() is idempotent and lets the test observe recovery instead of waiting
// for the next scheduled retry. Once the backlog is zero, the second message has been persisted
// on r2 and can be consumed normally.
Awaitility.await().atMost(TIME_TO_CHECK_BACKLOG_QUOTA * 8L, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
.untilAsserted(() -> {
replicator.startProducer();
assertEquals(replicator.computeStats().replicationBacklog, 0);
});
consumer2.receive(1);

Awaitility.await().untilAsserted(() -> assertEquals(replicator.computeStats().replicationBacklog, 0));
metrics = metricReader1.collectAllMetrics();
assertMetricLongSumValue(metrics, OpenTelemetryReplicatorStats.BACKLOG_COUNTER, attributes, 0);
assertMetricDoubleGaugeValue(metrics, OpenTelemetryReplicatorStats.DELAY_GAUGE, attributes, 0.0);
Expand Down
Loading