[improve][client] PIP-468: Make Backoff jitter configurable

pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/BackoffPolicy.java

@@ -26,7 +26,9 @@
 /**
  * Backoff configuration for broker reconnection attempts.
  *
- * <p>The delay for attempt {@code n} is {@code min(initialInterval * multiplier^(n-1), maxInterval)}.
+ * <p>The base delay for attempt {@code n} is {@code min(initialInterval * multiplier^(n-1), maxInterval)}.
+ * A symmetric random jitter of {@code ±jitterPercent/2} is applied to each delay (including the
+ * first one) to spread out concurrent retries.
  *
  * <p>Use {@link #fixed(Duration, Duration)} or {@link #exponential(Duration, Duration)} for
  * the common cases, or {@link #builder()} to configure all knobs explicitly.
@@ -35,23 +37,31 @@
 @ToString
 public final class BackoffPolicy {
 
+    /** Default jitter percentage applied when not explicitly specified. */
+    public static final double DEFAULT_JITTER_PERCENT = 10.0;
+
     private final Duration initialInterval;
     private final Duration maxInterval;
     private final double multiplier;
+    private final double jitterPercent;
 
-    private BackoffPolicy(Duration initialInterval, Duration maxInterval, double multiplier) {
+    private BackoffPolicy(Duration initialInterval, Duration maxInterval, double multiplier, double jitterPercent) {
         Objects.requireNonNull(initialInterval, "initialInterval must not be null");
         Objects.requireNonNull(maxInterval, "maxInterval must not be null");
         if (multiplier < 1.0) {
             throw new IllegalArgumentException("multiplier must be >= 1.0");
         }
+        if (jitterPercent < 0 || jitterPercent > 100) {
+            throw new IllegalArgumentException("jitterPercent must be in [0, 100]");
+        }
         this.initialInterval = initialInterval;
         this.maxInterval = maxInterval;
         this.multiplier = multiplier;
+        this.jitterPercent = jitterPercent;
     }
 
     /**
-     * @return the delay before the first reconnection attempt
+     * @return the base delay before the first reconnection attempt
      */
     public Duration initialInterval() {
         return initialInterval;
@@ -72,25 +82,33 @@ public double multiplier() {
     }
 
     /**
-     * Create a fixed backoff (no increase between retries).
+     * @return the symmetric jitter percentage applied to each delay; {@code 0} means no jitter
+     */
+    public double jitterPercent() {
+        return jitterPercent;
+    }
+
+    /**
+     * Create a fixed backoff (no increase between retries) with the default jitter.
      *
-     * @param initialInterval the constant delay between reconnection attempts
+     * @param initialInterval the constant base delay between reconnection attempts
      * @param maxInterval     the maximum delay between reconnection attempts
-     * @return a {@link BackoffPolicy} with a multiplier of 1.0
+     * @return a {@link BackoffPolicy} with a multiplier of 1.0 and the default jitter
      */
     public static BackoffPolicy fixed(Duration initialInterval, Duration maxInterval) {
-        return new BackoffPolicy(initialInterval, maxInterval, 1.0);
+        return new BackoffPolicy(initialInterval, maxInterval, 1.0, DEFAULT_JITTER_PERCENT);
     }
 
     /**
-     * Create an exponential backoff with the given bounds and a default multiplier of 2.
+     * Create an exponential backoff with the given bounds, a default multiplier of 2 and the
+     * default jitter.
      *
-     * @param initialInterval the delay before the first reconnection attempt
+     * @param initialInterval the base delay before the first reconnection attempt
      * @param maxInterval     the maximum delay between reconnection attempts
-     * @return a {@link BackoffPolicy} with a multiplier of 2.0
+     * @return a {@link BackoffPolicy} with a multiplier of 2.0 and the default jitter
      */
     public static BackoffPolicy exponential(Duration initialInterval, Duration maxInterval) {
-        return new BackoffPolicy(initialInterval, maxInterval, 2.0);
+        return new BackoffPolicy(initialInterval, maxInterval, 2.0, DEFAULT_JITTER_PERCENT);
     }
 
     /**
@@ -107,6 +125,7 @@ public static final class Builder {
         private Duration initialInterval;
         private Duration maxInterval;
         private double multiplier = 2.0;
+        private double jitterPercent = DEFAULT_JITTER_PERCENT;
 
         private Builder() {
         }
@@ -145,11 +164,25 @@ public Builder multiplier(double multiplier) {
             return this;
         }
 
+        /**
+         * Symmetric jitter percentage applied to each returned delay. The actual jitter is the
+         * base delay multiplied by a uniform random factor in
+         * {@code [1 - jitterPercent/200, 1 + jitterPercent/200)}. Default is {@code 10.0}; set to
+         * {@code 0} to disable jitter.
+         *
+         * @param jitterPercent the jitter percentage, must be in {@code [0, 100]}
+         * @return this builder
+         */
+        public Builder jitterPercent(double jitterPercent) {
+            this.jitterPercent = jitterPercent;
+            return this;
+        }
+
         /**
          * @return a new {@link BackoffPolicy} instance
          */
         public BackoffPolicy build() {
-            return new BackoffPolicy(initialInterval, maxInterval, multiplier);
+            return new BackoffPolicy(initialInterval, maxInterval, multiplier, jitterPercent);
         }
     }
 }

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java

@@ -111,7 +111,7 @@ public void testCorrectBackoffConfiguration() {
         ClientConfigurationData clientConfigurationData = new ClientConfigurationData();
         Assert.assertEquals(backoff.getMax().toMillis(),
                 TimeUnit.NANOSECONDS.toMillis(clientConfigurationData.getMaxBackoffIntervalNanos()));
-        Assert.assertEquals(backoff.next().toMillis(),
+        Assert.assertEquals(backoff.getInitial().toMillis(),
                 TimeUnit.NANOSECONDS.toMillis(clientConfigurationData.getInitialBackoffIntervalNanos()));
     }
 

pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java

@@ -28,8 +28,8 @@
  * Exponential backoff with mandatory stop.
  *
  * <p>Delays start at {@code initialDelay} and double on every call to {@link #next()}, up to
- * {@code maxBackoff}. A random jitter of up to 10% is subtracted from each value to avoid
- * thundering-herd retries.
+ * {@code maxBackoff}. A symmetric random jitter of {@code ±jitterPercent/2} is applied to every
+ * returned value (including the first one) to avoid thundering-herd retries.
  *
  * <p>If a {@code mandatoryStop} duration is configured, the backoff tracks wall-clock time from the
  * first {@link #next()} call. Once the elapsed time plus the next delay would exceed the mandatory
@@ -43,6 +43,7 @@
  *         .initialDelay(Duration.ofMillis(100))
  *         .maxBackoff(Duration.ofMinutes(1))
  *         .mandatoryStop(Duration.ofSeconds(30))
+ *         .jitterPercent(10.0)
  *         .build();
  *
  * Duration delay = backoff.next();
@@ -51,6 +52,7 @@
 public class Backoff {
     private static final Duration DEFAULT_INITIAL_DELAY = Duration.ofMillis(100);
     private static final Duration DEFAULT_MAX_BACKOFF_INTERVAL = Duration.ofMinutes(1);
+    private static final double DEFAULT_JITTER_PERCENT = 10.0;
     private static final Random random = new Random();
 
     @Getter
@@ -59,6 +61,8 @@ public class Backoff {
     private final Duration max;
     @Getter
     private final Duration mandatoryStop;
+    @Getter
+    private final double jitterPercent;
     private final Clock clock;
 
     private Duration next;
@@ -67,10 +71,11 @@ public class Backoff {
     @Getter
     private boolean mandatoryStopMade;
 
-    private Backoff(Duration initial, Duration max, Duration mandatoryStop, Clock clock) {
+    private Backoff(Duration initial, Duration max, Duration mandatoryStop, double jitterPercent, Clock clock) {
         this.initial = initial;
         this.max = max;
         this.mandatoryStop = mandatoryStop;
+        this.jitterPercent = jitterPercent;
         this.next = initial;
         this.clock = clock;
         this.firstBackoffTime = Instant.EPOCH;
@@ -101,8 +106,10 @@ public static Builder builder() {
     /**
      * Returns the next backoff delay, advancing the internal state.
      *
-     * <p>The returned duration is never less than the initial delay and never more than the max
-     * backoff. A random jitter of up to 10% is subtracted to spread out concurrent retries.
+     * <p>The underlying delay starts at the initial delay and doubles on each call up to the max
+     * backoff. A symmetric jitter of {@code ±jitterPercent/2} is applied on every call (including
+     * the first one) to spread out concurrent retries; the returned value may therefore be slightly
+     * below the initial delay or slightly above the max backoff.
      *
      * @return the delay to wait before the next retry attempt
      */
@@ -130,13 +137,13 @@ public Duration next() {
             }
         }
 
-        // Randomly decrease the timeout up to 10% to avoid simultaneous retries
         long currentMillis = current.toMillis();
-        if (currentMillis > 10) {
-            currentMillis -= random.nextInt((int) currentMillis / 10);
+        if (jitterPercent > 0 && currentMillis > 0) {
+            // Apply a symmetric jitter of ±jitterPercent/2 around the current delay.
+            double factor = 1.0 + (random.nextDouble() - 0.5) * (jitterPercent / 100.0);
+            currentMillis = Math.max(0L, Math.round(currentMillis * factor));
         }
-        long initialMillis = initial.toMillis();
-        return Duration.ofMillis(Math.max(initialMillis, currentMillis));
+        return Duration.ofMillis(currentMillis);
     }
 
     /**
@@ -162,12 +169,13 @@ public void reset() {
     /**
      * Builder for {@link Backoff}.
      *
-     * <p>Defaults: initial delay 100 ms, max backoff 1 min, no mandatory stop.
+     * <p>Defaults: initial delay 100 ms, max backoff 1 min, no mandatory stop, 10% jitter.
      */
     public static class Builder {
         private Duration initialDelay = DEFAULT_INITIAL_DELAY;
         private Duration maxBackoff = DEFAULT_MAX_BACKOFF_INTERVAL;
         private Duration mandatoryStop = Duration.ZERO;
+        private double jitterPercent = DEFAULT_JITTER_PERCENT;
         private Clock clock = Clock.systemDefaultZone();
 
         /**
@@ -205,6 +213,24 @@ public Builder mandatoryStop(Duration mandatoryStop) {
             return this;
         }
 
+        /**
+         * Sets the jitter percentage applied to each returned delay. The actual jitter is symmetric:
+         * the returned value is multiplied by a uniform random factor in
+         * {@code [1 - jitterPercent/200, 1 + jitterPercent/200)}. Defaults to 10. Set to 0 to disable
+         * jitter.
+         *
+         * @param jitterPercent the jitter percentage, must be in {@code [0, 100]}
+         * @return this builder
+         * @throws IllegalArgumentException if {@code jitterPercent} is outside {@code [0, 100]}
+         */
+        public Builder jitterPercent(double jitterPercent) {
+            if (jitterPercent < 0 || jitterPercent > 100) {
+                throw new IllegalArgumentException("jitterPercent must be in [0, 100]");
+            }
+            this.jitterPercent = jitterPercent;
+            return this;
+        }
+
         Builder clock(Clock clock) {
             this.clock = clock;
             return this;
@@ -216,7 +242,7 @@ Builder clock(Clock clock) {
          * @return a new Backoff
          */
         public Backoff build() {
-            return new Backoff(initialDelay, maxBackoff, mandatoryStop, clock);
+            return new Backoff(initialDelay, maxBackoff, mandatoryStop, jitterPercent, clock);
         }
     }
 }