feat(connectors): retry transient Doris Stream Load failures in-request

[ryankert01]

Which issue does this PR address?

Relates to #3215

Rationale

Follow-up to the Doris sink (#3215). The connector already classified transient Stream Load outcomes as retryable but had no retry path, so under the runtime's at-most-once delivery a transient backend blip silently dropped the batch.

What changed?

The Doris sink classified transient Stream Load outcomes (5xx/408/429, transport errors, Publish Timeout) as CannotStoreData but never retried them: the runtime commits the consumer offset at poll time before consume() runs and discards its return value, so a transient failure dropped the batch with no replay.

consume() now retries a transiently-failed batch in-request via a new load_batch() that wraps send plus status classification, re-PUTing under the same deterministic label so Doris dedupes a prior attempt that actually landed (e.g. a 2xx whose body could not be read). Permanent failures (4xx, Fail, schema/redirect problems) are never retried. Backoff and jitter come from iggy_connector_sdk::retry, bounded by new max_retries/retry_delay/max_retry_delay config (defaults 3 / 200ms / 5s). This shrinks the at-most-once window within a single poll; cross-poll and crash delivery stay a runtime concern.

Local Execution

• Passed
• Pre-commit hooks: checks run manually. The license-headers hook cannot execute on this machine (its script needs bash 4+ mapfile; only bash 3.2 is present), but hawkeye check passes directly and CI enforces it. markdownlint, taplo, cargo fmt, cargo clippy -p iggy_connector_doris_sink --all-targets -- -D warnings, and cargo test -p iggy_connector_doris_sink (41 tests) all pass.

AI Usage

1. Claude Code (Anthropic).
2. Implemented the retry loop, config plumbing, tests, and README/config updates, after verifying the runtime's at-most-once delivery semantics directly in the runtime source.
3. Three new wiremock unit tests pin the behavior: transient-then-success (retry fires), exhausted-budget (exact attempt count via .expect), and permanent-not-retried. Full crate suite, clippy, and doc lint pass locally.
4. Yes.

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[codecov]

Codecov Report

❌ Patch coverage is 96.80365% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 41.91%. Comparing base (a5001d3) to head (c906461).
⚠️ Report is 2 commits behind head on master.

Files with missing lines	Patch %	Lines
core/connectors/sinks/doris_sink/src/lib.rs	96.80%	5 Missing and 9 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3574       +/-   ##
=============================================
- Coverage     74.34%   41.91%   -32.43%     
  Complexity      950      950               
=============================================
  Files          1303     1301        -2     
  Lines        148712   129411    -19301     
  Branches     124225   104924    -19301     
=============================================
- Hits         110562    54246    -56316     
- Misses        34673    72404    +37731     
+ Partials       3477     2761      -716     

Components	Coverage Δ
Rust Core	34.15% <96.80%> (-40.43%)	⬇️
Java SDK	62.64% <ø> (ø)
C# SDK	72.28% <ø> (ø)
Python SDK	92.27% <ø> (ø)
PHP SDK	84.52% <ø> (ø)
Node SDK	92.83% <ø> (+0.12%)	⬆️
Go SDK	43.08% <ø> (ø)

Files with missing lines	Coverage Δ
core/connectors/sinks/doris_sink/src/lib.rs	95.04% <96.80%> (+2.70%)	⬆️

... and 409 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[hubcio]

overall this looks solid, most of below comments are just nits.

one thing not on a changed line so flagging it here: the retry loop is uncancellable by graceful shutdown. the runtime consumer task's tokio::select! only races the shutdown signal against consumer.next() - once a batch is taken, process_messages().await (which runs the sink's consume() -> retry loop) runs to completion with no shutdown arm. so a consume() stuck retrying against a down FE blocks shutdown for the whole budget - up to max_retries * (timeout + backoff) per chunk, tens of seconds to minutes with defaults. only blocks that one consumer's task, and it's not a deadlock. it's runtime-side (not this PR), but this PR stretches the window, so worth a line in operational guidance and maybe a follow-up to make the retry sleep shutdown-aware.

[hubcio]

the backoff/jitter reuse from iggy_connector_sdk::retry is good. the retry loop itself is now the third hand-written copy of the same skeleton (HttpRetryMiddleware::handle and check_connectivity_with_retry in the sdk both have it). you can't reuse the middleware here - doris signals failure as http 200 + {"Status":"Fail"} which it never inspects - but that's really an argument for the sdk exposing a generic retry_async(op, policy) the three call sites could share. non-blocking, sdk-level follow-up.

[ryankert01]

Agreed, filed #3702 to track a shared retry_async(op, policy) in the SDK. Keeping the hand-rolled loop here since the helper touches the two existing SDK call sites and belongs in its own PR.

[ryankert01]

Updated, sorry for the late reply tho. (I was on a conference trip)

[ryankert01]

/ready

[hubcio]

CannotStoreData is one of the two transient markers for the retry loop, but consume() also wraps a local serialize failure in the same variant (the simd_json::to_vec error arm) - that one never goes through load_batch, so it's never retried. harmless since the runtime collapses the return to an i32, just misleading to have the retryable-class variant on a deterministic local failure. a permanent-class variant or a short comment there would make the split cleaner.

[ryankert01]

Good catch. I changed local JSON serialization failures to Error::Serialization, so they are no longer represented by the transient CannotStoreData variant.

[hubcio]

floors at 1 but no ceiling. with the readme's own per-chunk bound (N * 6 * timeout + (N-1) * max_retry_delay), max_retries = 100 at the default 30s timeout works out to ~5h of blocking per chunk against a down FE - and per the graceful-shutdown note, close waits for all of it. a warn above some sane cap (say 10) would match the spirit of the retry_delay > max_retry_delay clamp in open().

[ryankert01]

Agreed. The sink now emits a startup warning when the effective max_retries exceeds 10, and the shutdown-delay risk is documented in the README and example configurations.

[hubcio]

asymmetry with the body-read-failure path: a 2xx whose body can't be read is transient (retry re-PUTs, label dedupe reveals the real outcome), but a 2xx that reads fine yet is empty or garbage is permanent. a proxy stripping the body to empty masks a committed load the same way a mid-stream reset does, and the same retry-under-same-label argument applies. the proxy-html / schema-change case is genuinely permanent, so maybe just the empty-body subcase deserves the transient branch. deliberate?

[ryankert01]

yeah. update to -> An empty 2xx response is transient and retries the identical request under the same Doris label.

[ryankert01]

/ready

[ryerraguntla]

I reviewed once. Need more time for me to make a notes for my self for each of the error state and handling of the duplicate/non duplicate label. Interesting scenarios.