minleejae opened a new issue, #3950:
URL: https://github.com/apache/iggy/issues/3950
## 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
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
