Fetcher advances the consumed position past a record it never returned (RecordTooLargeError) — silent record loss under async iteration
- 主要语言
- Python
- 星标
- 1.4k
- 派生
- 269
- 平均合并
- 1 天 1 小时
- 30 天内合并 PR
- 6
描述
## Summary
When a fetch response contains bytes for a partition but no complete record batch (a batch larger than `max_partition_fetch_bytes` on a non-first partition, which the broker truncates), the fetcher records a `RecordTooLargeError` **and advances the consumed position past the record it never delivered**:
https://github.com/aio-libs/aiokafka/blob/0ff6e7123a887fe254bb06b5d1f7b2742e6c4274/aiokafka/consumer/fetcher.py#L862-L878
```python
elif records.size_in_bytes() > 0:
...
err = RecordTooLargeError(...)
self._set_error(tp, err)
tp_state.consumed_to(tp_state.position + 1) # <-- skips the record
```
Two consequences, in increasing severity:
1. **`getmany()` / `getone()`**: the error is raised to the caller, but the position has already advanced — an application that catches `RecordTooLargeError` and continues (the natural reading of a per-record error) has silently lost the record.
2. **`async for` iteration**: `Consumer.__anext__` explicitly catches `RecordTooLargeError` and only logs it:
https://github.com/aio-libs/aiokafka/blob/0ff6e7123a887fe254bb06b5d1f7b2742e6c4274/aiokafka/consumer/consumer.py#L1250-L1263
Combined with the advanced position, the most common consumption pattern **drops the record entirely, with nothing but a log line** — and if `enable_auto_commit` is on (or the application commits the later offsets it receives), the skip becomes permanent.
## Why this still happens on modern brokers
With `FetchRequest` v3+ (KIP-74), the broker guarantees progress only for the **first non-empty partition of the request** — that partition's first batch is returned in full regardless of limits. For every other partition in the request, `max_partition_fetch_bytes` still truncates the response, so an oversized batch on a non-first partition arrives partial and takes this code path. Since the fetcher shuffles partition order per request (`random.shuffle(partition_data)`), whether a given oversized record is delivered (partition happened to be first) or skipped (it wasn't) is a coin flip per fetch.
For comparison, the Java client after KIP-74 never advances past an undelivered batch: the partition simply makes progress on a later fetch when it is first in the request; `RecordTooLargeException` is reserved for pre-v3 brokers and does not skip.
## Expected behavior
A record the consumer could not return must not be consumed past. Either:
- leave the position unchanged and let KIP-74 progress the partition on a later fetch (matching Java), or
- if skipping is ever desirable, make it an explicit opt-in rather than the silent default.
## Reproduction sketch
1. Topic with 2 partitions, broker `message.max.bytes` raised (e.g. 8 MB).
2. Produce one ~2 MB record to partition 0 (producer `max_request_size=8388608`) and a stream of small records to partition 1.
3. Consume both partitions with one consumer, default `max_partition_fetch_bytes` (1 MB), using `async for`.
4. Observe: whenever partition 0 is not first in the fetch request, the 2 MB record is skipped with only `log.exception("error in consumer iterator: %s")`; the consumer continues from the next offset.
Observed on aiokafka 0.14.0 (Python 3.14); the code paths are unchanged on current master (permalinks above). Found while auditing changelog-restore behavior for an exactly-once stream-processing framework built on aiokafka — happy to provide more detail or test against a branch.
贡献指南
评估
这个 Issue 还没有评估数据。