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/17 06:51:25 UTC
(pulsar) 01/02: [fix][misc] Fix Backoff mandatory-stop timer tracking (#25282)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5c300b776eee1271344538cee37a55cb89a896af
Author: sinan liu <li...@gmail.com>
AuthorDate: Sat Oct 17 08:31:25 2026 +0800
[fix][misc] Fix Backoff mandatory-stop timer tracking (#25282)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit 7c7efeed3cc6a19976cdd7616a868e00e56cd760)
---
.../org/apache/pulsar/common/util/Backoff.java | 7 +-
.../org/apache/pulsar/common/util/BackoffTest.java | 93 ++++++++++++++++++++++
2 files changed, 99 insertions(+), 1 deletion(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
index 842d3bbaa975..e4942d628c1a 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
@@ -35,6 +35,7 @@ public class Backoff {
private long mandatoryStop;
private long firstBackoffTimeInMillis;
+ private boolean firstBackoffTimeSet;
private boolean mandatoryStopMade = false;
private static final Random random = new Random();
@@ -50,6 +51,7 @@ public class Backoff {
this.mandatoryStop = unitMandatoryStop.toMillis(mandatoryStop);
this.clock = clock;
this.firstBackoffTimeInMillis = 0;
+ this.firstBackoffTimeSet = false;
}
public Backoff(long initial, TimeUnit unitInitial, long max, TimeUnit unitMax, long mandatoryStop,
@@ -67,8 +69,9 @@ public class Backoff {
if (!mandatoryStopMade) {
long now = clock.millis();
long timeElapsedSinceFirstBackoff = 0;
- if (initial == current) {
+ if (!firstBackoffTimeSet) {
firstBackoffTimeInMillis = now;
+ firstBackoffTimeSet = true;
} else {
timeElapsedSinceFirstBackoff = now - firstBackoffTimeInMillis;
}
@@ -95,6 +98,8 @@ public class Backoff {
public void reset() {
this.next = this.initial;
+ this.firstBackoffTimeInMillis = 0;
+ this.firstBackoffTimeSet = false;
if (initial == 0 && max == 0 && mandatoryStop == 0) {
this.mandatoryStopMade = true;
} else {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java
index 9ca9503b5c09..2359fbd405fd 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/BackoffTest.java
@@ -80,6 +80,99 @@ public class BackoffTest {
assertEquals(diffBackOffTime, 300);
}
+ @Test
+ public void mandatoryStopShouldWorkWithConstantBackoff() {
+ Clock mockClock = Mockito.mock(Clock.class);
+ Mockito.when(mockClock.millis())
+ .thenReturn(0L)
+ .thenReturn(400L);
+
+ Backoff backoff = new Backoff(
+ 100, TimeUnit.MILLISECONDS,
+ 100, TimeUnit.MILLISECONDS,
+ 300, TimeUnit.MILLISECONDS,
+ mockClock
+ );
+
+ // first call starts the mandatory-stop timer
+ backoff.next();
+ long firstBackoffTime = backoff.getFirstBackoffTimeInMillis();
+
+ // elapsed wall-clock time should exceed mandatory stop
+ backoff.next();
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), firstBackoffTime,
+ "firstBackoffTime should not be updated after the first call to next()");
+ assertTrue(backoff.isMandatoryStopMade(),
+ "mandatory stop should be reached even when initial == max (constant backoff)");
+ }
+
+ @Test
+ public void reduceToHalfShouldNotResetMandatoryStopTimer() {
+ Clock mockClock = Mockito.mock(Clock.class);
+ Mockito.when(mockClock.millis())
+ .thenReturn(0L)
+ .thenReturn(200L);
+
+ Backoff backoff = new Backoff(
+ 100, TimeUnit.MILLISECONDS,
+ 400, TimeUnit.MILLISECONDS,
+ 250, TimeUnit.MILLISECONDS,
+ mockClock
+ );
+
+ // first call starts the mandatory-stop timer (next becomes 200)
+ backoff.next();
+ long firstBackoffTime = backoff.getFirstBackoffTimeInMillis();
+
+ // This can bring the next delay back to initial, but it should not reset firstBackoffTime.
+ backoff.reduceToHalf();
+
+ backoff.next();
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), firstBackoffTime,
+ "reduceToHalf should not reset firstBackoffTime");
+ assertTrue(backoff.isMandatoryStopMade(),
+ "mandatory stop should be reached based on wall-clock time since the first call to next()");
+ }
+
+ @Test
+ public void resetShouldStartNewMandatoryStopCycle() {
+ Clock mockClock = Mockito.mock(Clock.class);
+ Mockito.when(mockClock.millis())
+ .thenReturn(0L)
+ .thenReturn(400L)
+ .thenReturn(1000L)
+ .thenReturn(1400L);
+
+ Backoff backoff = new Backoff(
+ 100, TimeUnit.MILLISECONDS,
+ 100, TimeUnit.MILLISECONDS,
+ 300, TimeUnit.MILLISECONDS,
+ mockClock
+ );
+
+ // Cycle 1: reach mandatory stop
+ backoff.next();
+ long firstCycleStart = backoff.getFirstBackoffTimeInMillis();
+ backoff.next();
+ assertTrue(backoff.isMandatoryStopMade(), "mandatory stop should be reached in cycle-1");
+
+ // Reset should clear mandatory-stop state so a new cycle can start.
+ backoff.reset();
+ assertFalse(backoff.isMandatoryStopMade(), "reset should clear mandatoryStopMade");
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0L, "reset should clear firstBackoffTime");
+
+ // Cycle 2: should start timing again and be able to reach mandatory stop again.
+ backoff.next();
+ long secondCycleStart = backoff.getFirstBackoffTimeInMillis();
+ assertEquals(secondCycleStart, 1000L,
+ "reset should start a new mandatory-stop timing window");
+ assertEquals(secondCycleStart - firstCycleStart, 1000L,
+ "reset should not reuse the old timing window");
+
+ backoff.next();
+ assertTrue(backoff.isMandatoryStopMade(), "mandatory stop should be reached in cycle-2");
+ }
+
@Test
public void basicTest() {
Clock mockClock = Clock.fixed(Instant.EPOCH, ZoneId.systemDefault());