aio-libs / aio-libs/aiokafka

Spurious `NotEnoughReplicasError` on exactly-once read-process-write against Kafka 4.x with Eligible Leader Replicas (KIP-966) enabled

未關閉
#1,172 0 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
主要語言
Python
星號
1.4k
分支
269
平均合併
1 天 1 小時
30 天內合併 PR
6

描述

**Describe the bug**

An exactly-once read-process-write loop — consume, produce, `send_offsets_to_transaction`,
`commit_transaction` — periodically receives `NOT_ENOUGH_REPLICAS` (error code 19) on the
`ProduceRequest` when the Apache Kafka **4.x** broker has **Eligible Leader Replicas**
(KIP-966, feature `eligible.leader.replicas.version=1`) enabled — the default on Kafka 4.0+.

aiokafka logs bursts like this, then refreshes metadata, retries, and the produce lands:

```
2026-07-19 10:15:29,746 WARNING aiokafka.producer.sender Got error produce response on topic-partition TopicPartition(topic='repro-out', partition=4), retrying. Error:
2026-07-19 10:15:29,747 WARNING aiokafka.producer.sender Got error produce response on topic-partition TopicPartition(topic='repro-out', partition=5), retrying. Error:
2026-07-19 10:15:29,747 WARNING aiokafka.producer.sender Got error produce response on topic-partition TopicPartition(topic='repro-out', partition=2), retrying. Error:
```

There is **no data loss and exactly-once is preserved** (the retry succeeds), but the
errors are spurious and noisy.

**Why it is spurious**

It happens on a **single-broker cluster with `replication.factor=1` and
`min.insync.replicas=1`**, where the leader is always its own only in-sync replica, so a
genuine `NOT_ENOUGH_REPLICAS` is impossible. Confirmed the ISR never changed by dumping the
KRaft metadata log — **zero `PartitionChangeRecord`s** for the whole run — and the broker
logged no ISR shrink / offline / WARN / ERROR. The broker returns code 19 for a partition
whose ISR is a constant `[0]` with `min.insync.replicas=1`.

**Root cause — isolated with a controlled A/B**

Same reproducer script (below), same broker, 4 minutes each, **only the
`eligible.leader.replicas.version` cluster feature toggled** (`transaction.version` held at
its default of 2):

| `eligible.leader.replicas.version` | transactions committed | code-19 errors |
|---|---|---|
| **1** (Kafka-4 default) | 1195 | **8** |
| **0** | 1195 | **0** |

Disabling ELR eliminates the error entirely. (Separately, downgrading
`transaction.version` 2→1 with ELR left on only *reduces* the rate, so ELR is the decisive
factor, not transaction verification.)

**Narrowing observation:** a transactional producer that only *produces* in a loop (no
consumer, no `send_offsets_to_transaction`) does **not** reproduce it — 474 transactions,
0 errors, ELR on. The error only appears once consumer offsets are committed inside the
transaction, so the trigger is on the offsets-in-transaction / `AddOffsetsToTxn` path
interacting with ELR-era metadata, not plain transactional produce.

This looks like another facet of the incomplete Kafka 4.0 protocol support tracked in
#1085: aiokafka 0.14.0 predates ELR (KIP-966), so with ELR enabled the broker returns
code 19 in a path the client does not expect. (Error-code-table staleness as in #1093 was
ruled out — code 19 = `NOT_ENOUGH_REPLICAS` is stable and correctly mapped in `errors.py`;
the broker really is emitting it.)

**Expected behaviour**

An exactly-once producer against a healthy Kafka 4.x cluster with ELR enabled should not
receive spurious `NOT_ENOUGH_REPLICAS` while the partition's ISR satisfies
`min.insync.replicas`.

**Environment**

- aiokafka: **0.14.0** (latest release)
- Python: 3.14
- Kafka broker: **apache/kafka 4.3.1**, KRaft, single node; finalized features
`transaction.version=2`, `eligible.leader.replicas.version=1` (Kafka-4 defaults).
- Also observed on a 3-broker Strimzi cluster (`replication.factor=3`) — this is not
single-broker specific.

**Reproducible example**

Broker setup (single-node apache/kafka 4.3.1, KRaft):

```bash
for t in repro-in repro-out; do
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic "$t" \
--partitions 8 --replication-factor 1 --config min.insync.replicas=1
done
# ELR is on by default on Kafka 4.0+. To toggle it for the A/B:
# kafka-features.sh --bootstrap-server localhost:9092 downgrade --feature eligible.leader.replicas.version=0 # -> 0 errors
# kafka-features.sh --bootstrap-server localhost:9092 upgrade --feature eligible.leader.replicas.version=1 # -> errors return
```

Reproducer (`python repro.py 240`):

```python
import asyncio, logging, sys, time
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from aiokafka.structs import TopicPartition

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-8s %(name)s %(message)s")
DURATION = float(sys.argv[1]) if len(sys.argv) > 1 else 240.0
GROUP_ID, IN_TOPIC, OUT_TOPIC = "elr-repro-group", "repro-in", "repro-out"

async def feeder(stop):
p = AIOKafkaProducer(bootstrap_servers="localhost:9092")
await p.start()
try:
while not stop.is_set():
for partition in range(8):
await p.send(IN_TOPIC, value=b'{"x":1}', partition=partition)
await asyncio.sleep(0.2)
finally:
await p.stop()

async def main():
stop = asyncio.Event()
feed = asyncio.create_task(feeder(stop))
consumer = AIOKafkaConsumer(IN_TOPIC, bootstrap_servers="localhost:9092", group_id=GROUP_ID,
enable_auto_commit=False, isolation_level="read_committed",
auto_offset_reset="earliest")
producer = AIOKafkaProducer(bootstrap_servers="localhost:9092", transactional_id="elr-repro-0",
enable_idempotence=True)
await consumer.start(); await producer.start()
deadline = time.monotonic() + DURATION; txns = 0
try:
while time.monotonic() < deadline:
batch = await consumer.getmany(timeout_ms=1000)
if not batch:
continue
async with producer.transaction():
offsets = {}
for tp, msgs in batch.items():
for m in msgs:
await producer.send(OUT_TOPIC, value=m.value, partition=tp.partition)
offsets[TopicPartition(tp.topic, tp.partition)] = msgs[-1].offset + 1
await producer.send_offsets_to_transaction(offsets, GROUP_ID)
txns += 1
finally:
stop.set(); await feed; await consumer.stop(); await producer.stop()
print(f"committed {txns} transactions")

asyncio.run(main())
```

Run it with ELR on → `NotEnoughReplicasError` bursts appear; disable ELR and run again →
none.

**Workaround**

On single-broker / RF=1 deployments (where ELR provides nothing) disable the feature:

```
kafka-features.sh --bootstrap-server localhost:9092 downgrade --feature eligible.leader.replicas.version=0
```

On multi-broker clusters ELR should stay enabled (it has real durability value); there the
retries are benign and the fix belongs in the client.

貢獻指南

開啟貢獻指南

評估

這個 Issue 還沒有評估資料。

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。