fix(connectors): bring quickwit_sink up to convention

[mfyuce]

Which issue does this PR address?

Closes #3814

Rationale

quickwit_sink was written before the shared connector retry helpers existed
and never caught up with them. Compared to the other HTTP sinks it had no
request timeout, no retry middleware, no readiness probe, and no handling for
two instances racing to create the same index. It also accepted only JSON
payloads and dropped everything else. Each of those is a way for the connector
to hang or lose data in a real deployment rather than fail visibly.

What changed?

reqwest::Client::new() has no timeout, so a network partition or a
slow-starting Quickwit left has_index() and ingest() hanging forever, and a
5xx during startup killed the connector outright. Two instances opening at once
both saw has_index() == false, both POSTed, and the loser got a 409 it treated
as a fatal InitError.

The client is now built in open() with a configurable timeout and wrapped in
the SDK's retry middleware, after check_connectivity_with_retry waits for
/health/readyz. create_index absorbs a 409, and a 400 whose body says the
index already exists, as success. Non-JSON payloads are wrapped instead of
discarded: raw bytes are parsed as JSON when they can be, otherwise carried as
text or base64.

Local Execution

• Passed

cargo fmt --all
cargo sort --check --no-format --workspace
cargo clippy -p iggy_connector_quickwit_sink --all-features --all-targets -- -D warnings
cargo test -p iggy_connector_quickwit_sink                      # 11 passed
cargo test -p integration -- connectors::quickwit               # 4 passed
./scripts/ci/taplo.sh --check
./scripts/ci/markdownlint.sh --check
./scripts/ci/license-headers.sh
./scripts/ci/trailing-whitespace.sh
./scripts/ci/trailing-newline.sh

The four integration tests run against a real Quickwit container and pass. Note
for anyone reproducing them locally: on this host they only get as far as
starting the container unless create_shard_executor keeps a worker pool. The
unconditional proactor.thread_pool_limit(0) in
core/server_common/src/executor.rs makes iggy-server abort on startup with
"the thread pool is needed but no worker thread is running", which takes down
every connectors integration test, not just these. That is unrelated to this PR
and is not touched here.

• Pre-commit hooks ran

AI Usage

1. Claude Code.
2. Used throughout: drafting the retry and payload-handling changes, writing the
   unit tests, and auditing the diff against reviewer comments.
3. Verified by running the full local check list above, including the four
   integration tests against a real Quickwit container, and by reading the diff
   against influxdb_sink, which this sink mirrors.
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.

[ryerraguntla]

lib.rs:ingest() — create_index() treats 409 Conflict as InitError; concurrent open() calls (multi-instance, restart race) both see has_index()=false, both POST, second gets 4xx → InitError → connector
never opens. Fix: absorb 409 (and 400 "already exists") as Ok(()) in create_index(). Retry middleware also retries a 5xx create that succeeded server-side; on retry, server returns 409 → same path. Same fix
covers both.

[mfyuce]

Sorry for the late response -- I was running a local benchmark to make sure the setup is working end-to-end.

Fixed in 512b71f04: create_index now handles 409 CONFLICT as a success case -- another instance beat us to it, but the index exists, which is what we wanted. Only other 4xx errors propagate as InitError.

[ryerraguntla]

reqwest::Client::new() has no timeout; health probe and has_index()/create_index() can hang indefinitely under network partition or slow-starting Quickwit. Fix: reqwest::Client::builder().timeout(...) with configurable or sensible default (e.g. 30s)

[mfyuce]

Fixed in 512b71f04: added request_timeout: Option<String> to QuickwitSinkConfig (default "30s") and wired it into reqwest::Client::builder().timeout(request_timeout).build() in open(). The example config TOML has the field commented in as a reference.

[ryerraguntla]

@mfyuce - all the above are minors. I am not sure of quickwit s production data set distribution to mention about the need for circuit breakers (if there are huge number of records/documents for the same cursor key for a given batch size) . Please make a judgement call about the need for circuit breaker implementation . otherwise it is all set for second reviewer's comments before merging.

[ryerraguntla]

/author

[mfyuce]

Thanks for the thorough review @ryerraguntla! Both issues addressed in the latest push:

409 / "already exists" on create_index() — create_index() now absorbs 409 CONFLICT and 400 BAD_REQUEST whose body contains "already exists" as Ok(()) with an info! log. Covers the concurrent-open race and the retry-after-succeeded-5xx path.

No timeout on reqwest::Client — Added request_timeout: Option<String> to QuickwitSinkConfig (default 30s, configurable via TOML). The raw client is now built with Client::builder().timeout(request_timeout), bounding health probes, has_index(), and create_index() under network partition.

/ready

[mfyuce]

@ryerraguntla — judgment call on the circuit breaker:

The existing build_retry_client wraps a HttpRetryMiddleware that already retries 429 and 5xx with exponential backoff, and the PermanentHttpError mapping cuts retries for permanent 4xx errors (bad data won't hammer the backend). Ingest throughput is also naturally rate-bounded by batch_length and poll_interval.

QuickWit is typically an internal service in the same cluster, so prolonged partition is rare, and the runtime already isolates failures per connector. A full half-open / trip-threshold circuit breaker on top of this would add complexity without clear benefit for this specific sink. Happy to revisit if the second reviewer sees a concrete failure mode that the existing retry policy doesn't cover.

[codecov]

Codecov Report

❌ Patch coverage is 83.96226% with 34 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.28%. Comparing base (e4e2564) to head (076773f).
⚠️ Report is 48 commits behind head on master.

Files with missing lines	Patch %	Lines
core/connectors/sinks/quickwit_sink/src/lib.rs	83.96%	27 Missing and 7 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3523       +/-   ##
=============================================
- Coverage     75.99%   63.28%   -12.72%     
  Complexity      969      969               
=============================================
  Files          1325     1324        -1     
  Lines        162383   151242    -11141     
  Branches     135262   124200    -11062     
=============================================
- Hits         123406    95715    -27691     
- Misses        35317    51762    +16445     
- Partials       3660     3765      +105     

Components	Coverage Δ
Rust Core	60.28% <83.96%> (-15.64%)	⬇️
Java SDK	62.71% <ø> (ø)
C# SDK	71.16% <ø> (-1.13%)	⬇️
Python SDK	89.45% <ø> (ø)
PHP SDK	84.52% <ø> (ø)
Node SDK	96.36% <ø> (+0.13%)	⬆️
Go SDK	43.08% <ø> (ø)

Files with missing lines	Coverage Δ
core/connectors/sinks/quickwit_sink/src/lib.rs	83.26% <83.96%> (+15.45%)	⬆️

... and 264 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.

[kriti-sc]

Can you please add a brief rationale/problem statement/why behind the change? I am finding the PR hard to review without clarity on the goal being achieved.
@mfyuce

[mfyuce]

Sorry for the late response -- was doing a local benchmark to make sure the setup is working.

@kriti-sc good call, here is the rationale:

The original quickwit_sink was missing several patterns that exist in every other production-grade connector in this repo. Without them, it would fail silently or hang in real deployments:

1. No request timeout -- reqwest::Client::new() has no timeout. Under network partition, has_index(), create_index(), and ingest() block forever. This PR adds request_timeout (default 30 s).

2. No connectivity check on open() -- the connector would report Running in the runtime's /stats endpoint even if QuickWit was unreachable, then fail silently on first ingest. This PR adds the same check_connectivity_with_retry probe that postgres_sink, http_sink, and elasticsearch_sink all use.

3. No retry middleware -- transient 5xx or 429 responses from QuickWit caused immediate batch failure and offset advancement. This PR wires HttpRetryMiddleware with exponential backoff, matching the behavior of the other HTTP-based sinks.

4. 409 Conflict on create_index() crashed open() -- in a multi-instance or restart race, two connectors can call create_index() simultaneously. The second one got a 409 and propagated it as an InitError, killing the connector. This PR absorbs 409 (and "already exists" 400) as Ok(()).

The goal is to bring quickwit_sink to the same robustness level as postgres_sink before it sees production traffic.

[kriti-sc]

instead of 30s, define a const like the DEFAULT_* defined at the top of this file, and use that here

[mfyuce]

Fixed in 3807e63. Added const DEFAULT_REQUEST_TIMEOUT: &str = "30s"; at the top of the file and updated the parse_duration call in open() to use it as the fallback. Also added explicit logs for server (5xx) versus client (4xx) errors on index creation.

[kriti-sc]

this and the else arm below are the same. is this intentioanl? what kind of errors is the else arm expected to catch?

also, would be good to add an error! to the else arm.

[mfyuce]

Good catch. The two arms are intentional -- 4xx (client errors) are permanent failures such as invalid index YAML or a misconfigured URL; 5xx (server errors) are transient and normally handled by the retry middleware, but they surface here when retries are exhausted.

The error! on the else arm was missing by mistake. Fixed in the latest push: both arms now log at error! level with distinct messages (Permanent client error vs Server error) so they are easy to tell apart in operator logs.

[kriti-sc]

these errors are the same pattern as in create_index above. here PermanentError and HttpRequestFailed error is used, while above InitError. What is the difference and is this intentional?

[mfyuce]

Intentional -- the two functions are called from different lifecycle phases:

• create_index is called from open() (the initialization phase). Failure there means the connector cannot start at all, so InitError is correct -- the runtime treats it as a fatal setup failure.
• ingest is called from consume() (the operation phase). Failure there is an HTTP-level error during normal operation, so HttpRequestFailed is correct.

The distinction matters for how the runtime handles each: InitError aborts the connector on first open; HttpRequestFailed during consume is subject to the retry/backoff policy.

[kriti-sc]

PR largely ok. A few minor changes + justification of some changes requested.

[mfyuce]

@kriti-sc The requested DEFAULT_REQUEST_TIMEOUT constant and explicit HTTP 4xx/5xx logging changes have been implemented and pushed. Ready for review!

[mfyuce]

/ready

[ryerraguntla]

/author please check the prechecks issues and resolve the conflicts

[mfyuce]

/ready

Hi @ryerraguntla,

I have resolved the conflicts and cleaned up the branch:

1. Rebase: Rebased the branch on origin/master (upstream master) and cleanly isolated the quickwit connector commits, purging any unrelated commits.
2. Prechecks: Verified that formatting (cargo fmt), clippy, whitespaces, and newlines checks pass perfectly.
3. Payload Handling: Brought the connector further up to project conventions by adding support for Payload::Raw (tries parsing JSON, falls back to Base64 wrapping if binary/non-JSON) and Payload::Text (wrapped inside a JSON object) to match the elasticsearch connector behavior.
4. Performance: Optimized the NDJSON generation logic during ingestion to build the payload using a single string buffer and avoid intermediate vector allocations.
5. Testing: Added comprehensive unit tests for the conversion logic. All 11 unit tests pass cleanly.

[ryerraguntla]

@mfyuce - while I review, please add the issue it is closing and the motivation for implementing. Follow the published Iggy PR template . Try to add code coverage to the maximum possible with out significantly increasing the test time.

[hubcio]

this is a clear improvement overall (drops the old panic, adds retry + a startup probe, handles raw/text payloads). inline notes are the should-fix items.

PR description nits (no line to anchor to): says "5 unit tests" but there are 11; the "map 4xx so the circuit breaker is not tripped" line describes a mechanism this sink doesn't have (no circuit breaker here); and "drop unused dashmap / once_cell" - once_cell was never a dependency here, only dashmap was.

one runtime-wide caveat, already tracked so not this pr's job: a permanently-failing consume() batch is silently dropped because the offset is already committed at poll and the FFI return code is ignored. that's #2927 (consume return discarded) + #2928 (commit-before-processing ordering) - together they make at-least-once unachievable for any sink today, which is worth keeping in mind next to the at-least-once comment inline. one small thing not yet noted on #2927: the dropped batch is still counted by the messages-processed metric, so dashboards read clean while data is lost.

[hubcio]

QuickwitSinkConfig is missing #[serde(deny_unknown_fields)]. with every knob optional, a typo'd key like max_retires is silently ignored and the connector just runs on defaults with no error. influxdb_sink (the sink this mirrors) sets it.

two smaller things on this struct: the Serialize derive is unused (the config is only ever deserialized; the runtime serializes an untyped value, not this type) so it can be dropped, and the retry/timeout knobs are undocumented while http_sink documents each field with its default.

[mfyuce]

All three done. #[serde(deny_unknown_fields)] is on QuickwitSinkConfig, the
Serialize derive is gone, and every field carries a doc comment with its
default. I dropped the same unused Serialize from the private IndexConfig
while I was there.

[hubcio]

max_retry_delay reads backwards from the rest. influxdb, where this startup-probe path is copied from, calls the same knob retry_max_delay and pairs it with open_retry_max_delay. worth renaming the field + the DEFAULT_MAX_RETRY_DELAY const to match, since config keys turn into a compat contract once this ships.

[mfyuce]

Renamed to retry_max_delay, with the const following as
DEFAULT_RETRY_MAX_DELAY, so it pairs with open_retry_max_delay the way
influxdb does.

[hubcio]

every other http sink names this timeout (influxdb, http_sink, doris) - request_timeout is the lone outlier. rename the field + DEFAULT_REQUEST_TIMEOUT const before release, config keys are hard to change later.

[mfyuce]

Renamed the field to timeout and the const to DEFAULT_TIMEOUT.

[hubcio]

this comment only covers the duplicate-write window. it's silent on the loss window: on a permanent 4xx, or once retries are exhausted, the offset was already committed at poll, so the batch is dropped and never redelivered. worth stating both so the real guarantee (at-most-once on final failure) is clear.

[mfyuce]

Rewritten to state both windows:

// At-least-once during transient retries, but at-most-once on final failure:
// Quickwit ingest carries no dedup key, so a retry after a transient 5xx/timeout
// that actually committed double-writes those rows. Conversely, if a batch permanently
// fails (e.g. 4xx client error or retries exhausted), the offset was already committed
// at poll, so the batch is silently dropped and never redelivered.

[hubcio]

these urls (here, plus has_index and create_index) are rebuilt with format! on every batch, and none trim a trailing slash - a url ending in / produces //api/v1/.... build them once in open() and trim_end_matches('/') the base, like influxdb's base_url().

[mfyuce]

base_url is now resolved once in new() with trim_end_matches('/') and
reused by has_index, create_index and ingest, so a configured
http://host:7280/ no longer produces //api/v1/....

[hubcio]

ndjson starts from String::new() and reallocs as it grows over the batch (up to batch_length records). pre-size it with String::with_capacity(...) - the commit says "optimize allocations" but this is the main alloc site.

separately: records where to_string fails get skipped silently while messages_count still counts them, so the success log can overcount.

[mfyuce]

@hubcio

About NDJSON Pre-allocation & Memory Footprint: To optimize the ndjson allocation without introducing magic numbers or overhead, a few approaches:

Pre-serializing all records to get the exact length: Cons=> holding all individual Strings in a Vec alongside the final joined String doubles the peak memory consumption of the batch payload.

Sample-and-estimate (using the first message length): Can lead to over/under-allocation in heterogeneous batches.

Current locally:

let mut ndjson = String::with_capacity(messages.len() * 512);

This prevents most reallocations without any CPU or memory overhead.
Does this standard messages.len() * 512 pre-allocation heuristic look good to you, or would you prefer a different approach here?

[mfyuce]

Following up: ndjson is pre-sized with
String::with_capacity(messages.len() * 512). The 512 is a rough per-record
estimate rather than a measured figure, so it trades a little slack for dropping
the realloc chain. Happy to change it if you would rather see a different basis.

The overcounting is fixed separately: messages_count now increments inside the
if let Ok(...) arm, so records that fail to serialize are no longer counted in
the success log.

[hubcio]

this permanent-vs-transient split has no runtime effect: the error is collapsed to a return code at the ffi boundary and the runtime discards it, and this sink owns no circuit breaker - so the description's "so the circuit breaker is not tripped" doesn't apply here. the two distinct log lines are still handy for triage, keep those; it's just the error variant that doesn't gate anything.

[mfyuce]

You are right that the split is inert in this sink today. consume() collapses
every Err into a return code at the FFI boundary (sdk/src/sink.rs:154), and
per #2927 the runtime discards that code for all sinks, so nothing reads the
classification.

I kept PermanentHttpError rather than flattening it, for two reasons. It is
what the other HTTP sinks already return for non-retryable 4xx
(clickhouse_sink/src/client.rs:232, doris_sink/src/lib.rs:263,
meilisearch_sink/src/lib.rs:566, influxdb_sink/src/lib.rs:700), and in
influxdb it is not inert: the circuit breaker skips record_failure for
permanent errors (influxdb_sink/src/lib.rs:825-831). Making quickwit the one
sink that drops the distinction would be the odd case, and if this sink ever
grows the same guard the classification is already in the right place.

The two log lines stay, as you suggested. That is also the workaround #2927
itself recommends: log inside consume() before returning Err.

The "so the circuit breaker is not tripped" line in the PR description was wrong
and I have removed it.

[hubcio]

these two arms build an identical Err(InitError(...)) and differ only in the log wording, and the status is already in the message. collapse them into one else. keep the success and 409 arms separate - the 409-absorb is load-bearing for the create race.

[mfyuce]

Collapsed into a single else. The success and 409 / 400-already-exists arms
stay separate.

[hubcio]

String::from_utf8(bytes.clone()) clones a second time. from_utf8(bytes) can consume the buffer, and on the error path e.into_bytes() hands it back for the base64 branch. the earlier clone at the parse call is load-bearing (simd parses destructively) so keep that one.

[mfyuce]

Done, applied as described.

[hubcio]

/health/livez is liveness - the node can be live but not yet ready to serve. /health/readyz is the readiness gate you actually want before create_index/ingest. minor, since the retry client papers over an early 5xx, but readyz is the more correct probe.

[mfyuce]

Switched to /health/readyz.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.

If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either @core on Discord or by mentioning them directly here on the PR.

Thank you for your contribution!

[hubcio]

@mfyuce when PR is ready for re-review, please put /ready somewhere at the beginning of line in any top level comment under this PR.

[mfyuce]

/ready

@mfyuce - while I review, please add the issue it is closing and the motivation for implementing. Follow the published Iggy PR template . Try to add code coverage to the maximum possible with out significantly increasing the test time.

Done, all three.
Closes #3814, opened for the gaps this PR fixes. Description rewritten to the template with the motivation.

On coverage: the sink had no tests, this adds 11 covering payload handling (JSON, raw JSON, raw text, raw binary, unsupported), config defaults, and index_id extraction. They are pure functions with no I/O, so the suite runs in under a second. The four integration tests in core/integration/tests/connectors/quickwit/ are unchanged and pass against a real Quickwit container, so total test time is unaffected.