aio-libs / aio-libs/aiokafka

Fetcher advances the consumed position past a record it never returned (RecordTooLargeError) — silent record loss under async iteration

未关闭
#1,173 1 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
主要语言
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 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。