This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 2b2e61253 fix(connectors): draw the retry delay inside the cap instead
of clamping (#4129)
2b2e61253 is described below
commit 2b2e612532f96dbe36f9e447888e8c7b7788e0d3
Author: Ryan Huang <[email protected]>
AuthorDate: Sat Sep 12 21:52:46 2026 +0800
fix(connectors): draw the retry delay inside the cap instead of clamping
(#4129)
---
core/connectors/sdk/src/retry.rs | 91 +++++++++++++++++++++---------
core/connectors/sinks/doris_sink/README.md | 4 +-
2 files changed, 67 insertions(+), 28 deletions(-)
diff --git a/core/connectors/sdk/src/retry.rs b/core/connectors/sdk/src/retry.rs
index b1d63294d..362eebbcd 100644
--- a/core/connectors/sdk/src/retry.rs
+++ b/core/connectors/sdk/src/retry.rs
@@ -153,17 +153,6 @@ pub fn parse_duration(value: Option<&str>, default_value:
&str) -> Duration {
})
}
-/// Apply ±20 % random jitter to `base` to spread retry storms.
-pub(crate) fn jitter(base: Duration) -> Duration {
- let millis = base.as_millis() as u64;
- let jitter_range = millis / 5; // 20% of base
- if jitter_range == 0 {
- return base;
- }
- let delta = rand::rng().random_range(0..=jitter_range * 2);
-
Duration::from_millis(millis.saturating_sub(jitter_range).saturating_add(delta))
-}
-
/// True exponential backoff: `base × 2^attempt`, capped at `max_delay`.
///
/// `attempt` is 0-based. Retry loops count from 1, so passing their counter
@@ -171,11 +160,11 @@ pub(crate) fn jitter(base: Duration) -> Duration {
/// takes a 1-based retry number and applies jitter and the cap.
pub fn exponential_backoff(base: Duration, attempt: u32, max_delay: Duration)
-> Duration {
let factor = 2u64.saturating_pow(attempt);
- let millis = base
- .as_millis()
+ let nanos = base
+ .as_nanos()
.saturating_mul(factor as u128)
- .min(max_delay.as_millis());
- Duration::from_millis(u64::try_from(millis).unwrap_or(u64::MAX))
+ .min(max_delay.as_nanos());
+ Duration::from_nanos(u64::try_from(nanos).unwrap_or(u64::MAX))
}
/// Parse a `Retry-After` header value (integer seconds).
@@ -217,19 +206,30 @@ impl RetryPolicy {
/// retry waits `base_delay`, the second `2 × base_delay`, and so on, which is
/// the convention the `retry_delay` config fields document.
///
-/// The cap is re-applied after jittering, because ±20 % jitter on an
-/// already-capped delay can otherwise land above `max_delay`, which the config
-/// fields document as a strict upper bound.
+/// The delay is drawn uniformly from a ±20 % window around that value, cut off
+/// at `max_delay`, which the config fields document as a strict upper bound.
+/// The window narrows as the backoff approaches the cap and sits entirely
+/// below it once the backoff saturates.
///
-/// Prefer [`retry_async`], which calls this for you. Reach for it directly
-/// only in a loop that cannot be expressed as a retried `Result`.
+/// Prefer [`retry_async`], which calls this for you. Reach for it directly in
+/// a loop that computes its own delay, such as [`HttpRetryMiddleware`], which
+/// retries on an `Ok` response rather than an `Err`.
pub fn retry_backoff(base_delay: Duration, retry: u32, max_delay: Duration) ->
Duration {
- jitter(exponential_backoff(
- base_delay,
- retry.saturating_sub(1),
- max_delay,
- ))
- .min(max_delay)
+ let target = exponential_backoff(base_delay, retry.saturating_sub(1),
max_delay);
+ let target_nanos = u64::try_from(target.as_nanos()).unwrap_or(u64::MAX);
+ let spread = target_nanos / 5; // 20% of the target
+
+ // Draw uniformly inside the window, cut off at `max_delay`. A draw that
+ // clamps instead lands on the bound itself, and at the cap that is half of
+ // every draw, which puts the most-backed-off instances back in step.
+ let low = target_nanos.saturating_sub(spread);
+ let high = target_nanos
+ .saturating_add(spread)
+ .min(u64::try_from(max_delay.as_nanos()).unwrap_or(u64::MAX));
+ if high <= low {
+ return Duration::from_nanos(low);
+ }
+ Duration::from_nanos(rand::rng().random_range(low..=high))
}
/// Why [`retry_async`] stopped.
@@ -576,6 +576,7 @@ pub async fn check_connectivity_with_retry(
mod tests {
use super::*;
use std::cell::Cell;
+ use std::collections::HashSet;
use std::time::Instant;
use wiremock::matchers::method;
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -856,6 +857,44 @@ mod tests {
}
}
+ #[test]
+ fn given_a_sub_millisecond_delay_should_still_spread() {
+ // Whole-millisecond math rounds the window to zero below 5 ms and
+ // returns the target unchanged, so every instance waits the same.
+ let policy = RetryPolicy {
+ max_attempts: 3,
+ base_delay: Duration::from_micros(500),
+ max_delay: Duration::from_millis(10),
+ };
+ let spread: HashSet<Duration> = (0..64).map(|_|
policy.backoff(1)).collect();
+
+ assert!(
+ spread.len() > 1,
+ "the backoff collapsed sub-millisecond delays onto {spread:?}"
+ );
+ assert!(policy.backoff(1) > Duration::ZERO);
+ }
+
+ #[test]
+ fn given_a_saturated_backoff_should_not_pile_onto_the_cap() {
+ // Base far above the cap, so every draw starts saturated.
+ let policy = RetryPolicy {
+ max_attempts: 16,
+ base_delay: Duration::from_secs(30),
+ max_delay: Duration::from_secs(1),
+ };
+
+ let draws = 512;
+ let at_cap = (0..draws)
+ .filter(|_| policy.backoff(8) == policy.max_delay)
+ .count();
+
+ assert!(
+ at_cap * 10 < draws,
+ "{at_cap}/{draws} saturated draws landed exactly on max_delay"
+ );
+ }
+
#[test]
fn given_jitter_on_a_capped_delay_should_never_exceed_max_delay() {
// Base far above the cap, so every draw starts clamped and only jitter
diff --git a/core/connectors/sinks/doris_sink/README.md
b/core/connectors/sinks/doris_sink/README.md
index 0cea21c37..bf47169fb 100644
--- a/core/connectors/sinks/doris_sink/README.md
+++ b/core/connectors/sinks/doris_sink/README.md
@@ -43,8 +43,8 @@ The Doris sink connector consumes JSON messages from Iggy
streams and writes the
| `timeout` | no | `30s` | Per-request HTTP timeout (total request budget), as
a human-readable duration (e.g. `30s`, `1m`). |
| `connect_timeout` | no | `5s` | TCP connect timeout, independent of
`timeout`, as a human-readable duration. Raise it for cross-region or
cold-start FEs. |
| `max_retries` | no | `3` | Total Stream Load attempts per batch on a
*transient* failure (`0` or `1` disables retries). Each retry re-PUTs under the
same label, which Doris dedupes. Values above `10` are honored but emit a
startup warning because they can substantially delay graceful shutdown. |
-| `retry_delay` | no | `200ms` | Base backoff before the first retry; doubles
each attempt up to `max_retry_delay`, with ±20% jitter. |
-| `max_retry_delay` | no | `5s` | Strict upper bound on a single retry
backoff, including jitter. |
+| `retry_delay` | no | `200ms` | Base backoff before the first retry; doubles
each attempt up to `max_retry_delay`, drawn from a ±20% window that is cut off
at the cap. |
+| `max_retry_delay` | no | `5s` | Strict upper bound on a single retry
backoff, including jitter. The jitter window narrows as the backoff nears this
cap, so the bound holds without every instance landing on it. |
| `max_filter_ratio` | no | unset | Forwarded as the `max_filter_ratio` Stream
Load header. Must be a finite value in `[0.0, 1.0]`; an out-of-range value
fails `open()`. |
| `columns` | no | unset | Forwarded as the `columns` Stream Load header.
Validated at startup; an invalid value fails `open()`. |
| `where` | no | unset | Forwarded as the `where` Stream Load header.
Validated at startup; an invalid value fails `open()`. |
