apache / apache/iggy

Source forwarding channel still grows past the batch-result timeout, and a timed-out batch that succeeds is reported as an error

Open
#3,981 0 comments 0 reactions 0 assignees Claimed by @rohankumardubey View on GitHub
bug connectors rust
Dominant language
Rust
Stars
4.9k
Forks
432
Avg merge
2d 10h
Merged PRs (30d)
174

Description

Two things that outlived #3795. That PR proposed bounding the runtime's source forwarding channel, and it was closed because `af9ce9548` (#3855) made its premise obsolete: with one batch in flight the channel cannot grow. That holds while the runtime answers inside the batch-result timeout. It stops holding past it, and closing the PR left nothing tracking the remainder.

Everything below is read off master `03f5b3034`.

## 1. The forwarding channel can still grow, one batch per `result_timeout`

`BATCH_RESULT_TIMEOUT` is 30s (`core/connectors/sdk/src/source.rs:54`). When the forwarding loop lags past it, `handle_messages` (`:360-392`) stops waiting on the oneshot, NACKs, increments `batch_id`, and polls again. The timed-out batch is still sitting in the channel, which is `flume::unbounded()` (`core/connectors/runtime/src/source.rs:625`), so each timeout adds one batch that nothing removes on the producing side.

`MAX_CONSECUTIVE_NACKS = 5` (`sdk/src/source.rs:57`) looks like the ceiling, but `apply_batch_result` resets the counter on any `Ack` (`:521-528`). A run that alternates ack and timeout never trips it, so growth is bounded only by how long the lag persists, at roughly one batch per 30s per source.

The rate is low and the batches are drained eventually, so this is memory pressure rather than loss, and ordering stays correct because the loop is sequential and the channel is FIFO. It matters most for a source whose batches are large: `http_source` at the default `max_batch_size` of 500 and a 1 MiB body cap can put half a gigabyte in a batch.

It also silently caps backpressure for any source that relies on the one-in-flight rule to propagate a slow Iggy back to its own producer. `http_source` answers 429 once its bridge fills, which only happens while the poll loop is stalled; past 30s the loop resumes, the bridge drains, and the sender stops being told anything is wrong.

## 2. A batch that times out and then succeeds is reported as a failure

Same timeout path. The runtime finishes the send and calls `batch_result_callback` for the original `batch_id` (`runtime/src/source.rs:559`). By then the plugin has cleared that pending batch and moved on, so `take_pending_batch` sees an id mismatch (`sdk/src/source.rs:466-484`), logs "Batch result ID mismatch", and returns `None`. `complete_pending_batch` then returns non-zero, and the runtime treats that as a delivery failure: `error!` plus `inc_errors_with_labels` plus `set_error` on the connector (`runtime/src/source.rs:566-578`).

So a batch that reached Iggy intact leaves the connector in an error state with a misleading message, and the error counter records a failure that did not happen. The plugin has also already applied `on_batch_result(Nack)` for it and re-polled the same data, so the duplicate is expected and documented; the spurious error state is not.

## Possible directions

For 1, the options are to bound the channel and block the FFI callback rather than drop (which is what #3795 would have become had it been reshaped), to make `result_timeout` configurable per connector so an operator can match it to their broker, or to have the runtime drop a batch whose result the plugin has already given up on. That last one needs the runtime to know the plugin timed out, which it currently cannot.

For 2, the runtime could treat an id mismatch as benign when the batch it is reporting is older than the plugin's current pending id, since that is precisely the timed-out case and not a protocol violation.

Related: #3941 covers the other end of the same policy, the five-NACK stop and its invisibility. If `BatchPolicy` becomes configurable there, `result_timeout` is the third knob.

Contributor guide

Open the contributing guide

Research direction

Start with core/connectors/sdk/src/source.rs, especially BATCH_RESULT_TIMEOUT, handle_messages, take_pending_batch, and apply_batch_result, then trace the matching forwarding channel and callback handling in core/connectors/runtime/src/source.rs. Determine the chosen handling for timed-out batches and verify that sustained timeouts do not allow unbounded channel growth, while a late successful result does not mark the connector as failed.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
distributed-systems
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.