feat(connectors): retry transient Doris Stream Load failures in-request#3574
feat(connectors): retry transient Doris Stream Load failures in-request#3574ryankert01 wants to merge 2 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3574 +/- ##
=============================================
- Coverage 73.49% 43.43% -30.07%
Complexity 937 937
=============================================
Files 1291 1288 -3
Lines 141568 123040 -18528
Branches 117128 98600 -18528
=============================================
- Hits 104046 53442 -50604
- Misses 34300 66887 +32587
+ Partials 3222 2711 -511
🚀 New features to boost your workflow:
|
549e171 to
c8ad134
Compare
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
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.
| /// whose body we couldn't read). A `PermanentHttpError` returns immediately — | ||
| /// retrying bad data would just hammer the FE. | ||
| async fn load_batch(&self, label: &str, body: Bytes) -> Result<StreamLoadResponse, Error> { | ||
| let connected = self.connected.as_ref().ok_or_else(|| { |
There was a problem hiding this comment.
load_batch re-runs the same connected.as_ref().ok_or_else(|| Error::InitError(...)) none-check that send_stream_load already does. the binding is needed here (you read max_retries/retry_delay/retry_max_delay off it), but the none-guard itself is redundant since send_stream_load re-checks on every attempt. minor - could drop the duplicate guard.
|
|
||
| let mut attempt = 0u32; | ||
| loop { | ||
| let result = match self.send_stream_load(label, body.clone()).await { |
There was a problem hiding this comment.
the two stacked matches here collapse into one - the first is exactly what and_then does:
let error = match self
.send_stream_load(label, body.clone())
.await
.and_then(|response| classify_status(&response).map(|()| response))
{
Ok(response) => return Ok(response),
Err(error) => error,
};same semantics, drops the intermediate result binding.
| return Err(error); | ||
| } | ||
|
|
||
| let delay = jitter(exponential_backoff( |
There was a problem hiding this comment.
backoff exponent is attempt - 1, so the first retry waits retry_delay (2^0). the sdk's own callers (HttpRetryMiddleware, check_connectivity_with_retry) pass the already-incremented attempts, so their first retry waits 2 * retry_delay. not wrong - arguably better - but it's a divergent convention across the connectors. worth a one-line comment on why, or align with the siblings.
| } | ||
|
|
||
| /// Total Stream Load attempts per batch: the configured value floored at 1 so | ||
| /// `0` (or an unset value) collapses to a single attempt with no retry. |
There was a problem hiding this comment.
this comment is off for the unset case. configured.unwrap_or(DEFAULT_MAX_RETRIES).max(1) maps an unset value to the default 3, not to a single attempt - only Some(0) collapses to 1. so 0 disables retries, unset gives you 3. worth fixing the comment (the config docs also only mention 1 disables, not 0).
| // at `open()`. `max_retries` is the total attempt count (1 = no retry). | ||
| max_retries: u32, | ||
| retry_delay: Duration, | ||
| retry_max_delay: Duration, |
There was a problem hiding this comment.
one knob, three spellings: the config key is max_retry_delay (line 137) but the field here is retry_max_delay and the const is DEFAULT_RETRY_MAX_DELAY (line 70). the config key is wire-facing so keep it - align the field + const to max_retry_delay / DEFAULT_MAX_RETRY_DELAY so it reads as one thing.
| ## Operational guidance | ||
|
|
||
| - **`label_keep_max_second`.** Idempotent replay relies on Doris retaining each label for at least as long as it could take the Iggy runtime to redrive a failed batch. The Doris default is 3 days, which is conservative. If you set this lower on the Doris side, make sure your runtime retry budget fits inside the window — once a label expires, a replay re-loads instead of deduping, producing duplicate rows. | ||
| - **`label_keep_max_second`.** The connector's in-request retry re-PUTs a transiently-failed batch under the same label, so Doris must retain that label for at least the connector's full retry budget for the replay to dedupe. The Doris default is 3 days, which is conservative. If you set this lower on the Doris side, make sure `max_retries × max_retry_delay` fits inside the window — once a label expires, a retry re-loads instead of deduping, producing duplicate rows. |
There was a problem hiding this comment.
the max_retries * max_retry_delay budget here understates the real worst case. a slow BE can burn most of the request timeout before the batch returns a retryable error (a 2xx whose body read stalls, or a BE slow to respond) - with defaults that 30s timeout is larger than the 5s max_retry_delay. jitter also adds up to +20% on top of each capped delay. closer bound is max_retries * (timeout + max_retry_delay).
| - **Filtered-row alerts.** When Doris reports `number_filtered_rows > 0`, the connector emits a `warn!`. This is your signal that upstream message shapes have drifted from the table schema; alert on it. | ||
| - **Multi-chunk batches are best-effort for operational failures.** A poll larger than `batch_size` is split into chunks, each loaded as its own labelled Stream Load. If a chunk fails *operationally* (serialize, HTTP, or status-classification error), the connector still attempts the remaining chunks and then returns the worst error — it does **not** stop at the first such failure. | ||
| The runtime commits the consumer offset for the whole poll before `consume()` runs, so a failed chunk is not replayed regardless; pushing the other chunks through maximizes delivered data, and the worst error is surfaced at the end (logged at `error!` for observability — the runtime currently discards `consume()`'s return value, so there is no retry or DLQ). | ||
| - **Multi-chunk batches are best-effort for operational failures.** A poll larger than `batch_size` is split into chunks, each loaded as its own labelled Stream Load (with its own in-request retry budget for transient failures). If a chunk still fails after its retries (serialize, HTTP, or status-classification error), the connector keeps the worst error, attempts the remaining chunks, and returns that error at the end — it does **not** stop at the first such failure. |
There was a problem hiding this comment.
says the connector "keeps the worst error" (and "worst error is surfaced" just below), but consume() keeps the first error via first_error.get_or_insert(...) - there's no severity comparison anywhere, and the comment right above the loop even says "keep the first error". harmless since the runtime discards the return and every chunk error is logged, but the wording should say first, not worst.
| /// label lets Doris dedupe a prior attempt that actually landed (e.g. a 2xx | ||
| /// whose body we couldn't read). A `PermanentHttpError` returns immediately — | ||
| /// retrying bad data would just hammer the FE. | ||
| async fn load_batch(&self, label: &str, body: Bytes) -> Result<StreamLoadResponse, Error> { |
There was a problem hiding this comment.
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.
github-actions
Bot
added
S-waiting-on-author
|
Updated, sorry for the late reply tho. (I was on a conference trip) |
|
/ready |
The Doris sink classified transient Stream Load outcomes (5xx/408/429, transport errors, Publish Timeout) as retryable but never acted on them: the runtime commits the consumer offset at poll time before consume() runs and discards its return value, so a transient backend blip silently dropped the batch under at-most-once delivery. consume() now retries a transiently-failed batch in-request, 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 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 remain a runtime concern, not something a sink can fix. Relates to apache#3215.
a30fac5 to
e3f7454
Compare
github-actions
Bot
added
S-waiting-on-review
| /// transport errors, and duplicate labels whose existing Doris job is `RUNNING` | ||
| /// or `CANCELLED`. `PermanentHttpError` (4xx, "Fail", schema/redirect problems, | ||
| /// unparsable body) is never retried — re-PUTing bad data just hammers the FE. | ||
| fn is_transient_error(error: &Error) -> bool { |
There was a problem hiding this comment.
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.
|
|
||
| /// Total Stream Load attempts per batch. An unset value uses the default; | ||
| /// configured `0` or `1` both mean one attempt with no retry. | ||
| fn effective_max_retries(configured: Option<u32>) -> u32 { |
There was a problem hiding this comment.
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().
| @@ -575,8 +665,8 @@ fn truncate_for_log(s: &str, max_bytes: usize) -> String { | |||
|
|
|||
| fn parse_stream_load_response(body: &str) -> Result<StreamLoadResponse, Error> { | |||
There was a problem hiding this comment.
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?
github-actions
Bot
added
S-waiting-on-author
2 participants
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) asCannotStoreDatabut never retried them: the runtime commits the consumer offset at poll time beforeconsume()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 newload_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. a2xxwhose body could not be read). Permanent failures (4xx,Fail, schema/redirect problems) are never retried. Backoff and jitter come fromiggy_connector_sdk::retry, bounded by newmax_retries/retry_delay/max_retry_delayconfig (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
license-headershook cannot execute on this machine (its script needs bash 4+mapfile; only bash 3.2 is present), buthawkeye checkpasses directly and CI enforces it.markdownlint,taplo,cargo fmt,cargo clippy -p iggy_connector_doris_sink --all-targets -- -D warnings, andcargo test -p iggy_connector_doris_sink(41 tests) all pass.AI Usage
.expect), and permanent-not-retried. Full crate suite, clippy, and doc lint pass locally.