diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index 9753e68dfb16d..10d616a2b0fa0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -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;
@@ -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");
@@ -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(),
@@ -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);
@@ -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);