connectors: runtime discards the sink consume FFI status, so failed batches are counted as processed
- Dominant language
- Rust
- Stars
- 4.9k
- Forks
- 432
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 173
Description
## The shape
`iggy_sink_consume` is declared `-> i32` and the SDK returns non-zero on failure (`-1` when the batch cannot be deserialized, `1` when the sink's `consume()` errors — `sdk/src/sink.rs`). The runtime discards that status: `process_messages` calls `(consume)(...)` without binding the result and returns `Ok(SinkBatchTiming { .. })` unconditionally (`runtime/src/sink.rs`).
Since the consumer commits offsets at poll time (`AutoCommitWhen::PollingMessages`), a sink that cannot write still increments `messages_processed`, keeps `status: running` with no `last_error`, and the batch is permanently lost.
The same file checks the same channel for `iggy_sink_open` (`if result != 0 → Err`); only the per-batch call is unchecked.
## Reproduced end-to-end
postgres_source → server → postgres_sink on master, with a simulated sink DB failure: a 6-row source produces a 5-row sink. The plugin logs the insert error, the runtime still logs `Consumed 1 messages`, `/sinks` reports `running`, and the lost message is never redelivered.
## Fix
Bind the status; on non-zero, return `RuntimeError::SinkConsumeFailed { plugin_id, status }` — the caller's existing `Err` path already logs the failure and skips the processed counter. I have this ready with a regression test (stub `extern "C"` callbacks; the test fails on master with `expected SinkConsumeFailed, got success`).
---
**Affected area**: connectors runtime (`core/connectors/runtime/src/sink.rs`)
**Version**: master @ `cc269ef70`
**Logs** (from the E2E repro):
```
ERROR connector_target="iggy_connector_postgres_sink" Failed to insert batch: Cannot store data: ...
INFO iggy_connectors::sink: Consumed 1 messages in 3.9ms for sink connector with ID: 2
```
## Scope
Propagating the status is fail-fast only — it surfaces the failure but does not redeliver the batch, since offsets are already committed at poll time. The full fix is the offset-commit/DLQ design discussed in #2927.
Contributor guide
Research direction
Start in core/connectors/runtime/src/sink.rs, reading process_messages and the nearby iggy_sink_open status check. Run the connectors runtime tests, including the regression test with stub extern "C" callbacks. Done means a non-zero iggy_sink_consume status returns SinkConsumeFailed and the failed batch is not counted as processed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 86/100