apache / apache/iggy

bug(connectors): consume() return value discarded in sink runtime

Open
#2,927 14 comments 0 reactions 1 assignee Claimed by @kriti-sc View on GitHub
connectors
Dominant language
Rust
Stars
4.9k
Forks
432
Avg merge
2d 10h
Merged PRs (30d)
173

Description

## Description

The connector runtime calls `consume()` on each sink plugin via FFI but discards the return value. If a sink's `consume()` returns an error (e.g., HTTP timeout, database connection failure, serialization error), the runtime ignores it — no logging, no retry, no metric. It proceeds to the next poll cycle as if delivery succeeded.

## Location

`core/connectors/runtime/src/sink.rs`, lines 585-593:

```rust
(consume)(
plugin_id,
topic_meta.as_ptr(),
topic_meta.len(),
messages_meta.as_ptr(),
messages_meta.len(),
messages.as_ptr(),
messages.len(),
);
```

The FFI function returns an `i32` (0 = success, 1 = error), but the return value is never checked.

## Impact

This affects **all 7 sink connectors** (Elasticsearch, Iceberg, MongoDB, PostgreSQL, Quickwit, Stdout, and the new HTTP sink). Every `Err(...)` returned by any sink's `consume()` implementation is silently swallowed by the runtime.

As a workaround, sink implementations must log errors internally before returning `Err`. The HTTP sink (#2925) documents this limitation and logs all errors at `error!` level inside `consume()`.

## Suggested Fix

Check the return value and log at minimum:

```rust
let result = (consume)(
plugin_id,
topic_meta.as_ptr(),
topic_meta.len(),
messages_meta.as_ptr(),
messages_meta.len(),
messages.as_ptr(),
messages.len(),
);

if result != 0 {
error!(
"Sink plugin {} consume() returned error code {}",
plugin_id, result
);
// Optionally: increment error metric, trigger retry, etc.
}
```

## Related

- Discussed in #2919 (HTTP sink connector proposal)
- Related to #2928 (offset commit ordering)
- HTTP sink PR: #2925

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.