apache / apache/iggy

connectors: runtime discards the sink consume FFI status, so failed batches are counted as processed

Open Beginner friendly
#3,950 0 comments 0 reactions 0 assignees View on GitHub
bug connectors rust
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.