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

[minleejae]

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