aio-libs / aio-libs/aiokafka

StickyPartitionAssignor.on_generation_assignment is defined but never invoked from GroupCoordinator

Aperta
#1,176 1 commento 0 reazioni 0 assegnatari Vedi su GitHub
Lingua principale
Python
Stelle
1.4k
Fork
269
Merge medio
1g 1h
PR unite (30g)
6

Descrizione

### Description

`StickyPartitionAssignor.on_generation_assignment` is defined at [`sticky_assignor.py:876`](https://github.com/aio-libs/aiokafka/blob/master/aiokafka/coordinator/assignors/sticky/sticky_assignor.py):

```python
@classmethod
def on_generation_assignment(cls, generation: int) -> None:
"""Callback that runs on each assignment. Updates assignor's generation id."""
log.debug("On generation assignment: generation=%d", generation)
cls.generation = generation
```

But it has **no call sites anywhere in the aiokafka codebase**:

```console
$ grep -rn "on_generation_assignment" aiokafka/
aiokafka/coordinator/assignors/sticky/sticky_assignor.py:876: def on_generation_assignment(cls, generation: int) -> None:
```

Only the definition. Compare with `on_assignment` (line 866), which *is* wired up from `GroupCoordinator._on_join_complete` at [`aiokafka/consumer/group_coordinator.py:488`](https://github.com/aio-libs/aiokafka/blob/master/aiokafka/consumer/group_coordinator.py):

```console
$ grep -rn "on_assignment\|on_generation_assignment" aiokafka/consumer/group_coordinator.py
aiokafka/consumer/group_coordinator.py:488: assignor.on_assignment(assignment)
```

Result: `StickyPartitionAssignor.generation` stays at `DEFAULT_GENERATION_ID = -1` for the lifetime of the process, regardless of how many rebalances happen.

### Impact

`_metadata()` (line 861) encodes `cls.generation` into the user data every consumer sends to the group leader on join:

```python
data = StickyAssignorUserDataV1(partitions_by_topic.items(), generation)
user_data = data.encode()
```

The generation is the tiebreaker sticky uses in `_init_current_assignments` (line 258) when two members' user data both claim the same partition:

```python
if partition in sorted_partition_consumers_by_generation:
consumers = sorted_partition_consumers_by_generation[partition]
if (
member_metadata.generation
and member_metadata.generation in consumers
):
log.warning(
"Partition %r is assigned to multiple consumers "
"following sticky assignment generation %d",
partition, member_metadata.generation,
)
else:
consumers[member_metadata.generation] = consumer
```

With every generation frozen at `-1`, the tiebreaker is inert. When two members legitimately have overlapping partition claims (e.g. homogeneous-subscription groups where members previously held partitions that need reassignment), the assignor cannot distinguish "member A held this partition at generation N" from "member B held it at generation N+1" — both look identical, and one gets dropped with the warning.

For heterogeneous-subscription groups (each member subscribed to a disjoint topic set) the tiebreaker is never exercised in normal operation, so sticky still preserves assignments correctly. For homogeneous-subscription groups the sticky mechanism degrades to best-effort.

### Reproducer

```python
from aiokafka.coordinator.assignors.sticky.sticky_assignor import StickyPartitionAssignor, StickyAssignorUserDataV1

# Any rebalance in the process
StickyPartitionAssignor.on_generation_assignment(42) # <-- never invoked in real code
# ... but even if we DID invoke it, cls.generation is only ever read here:

md = StickyPartitionAssignor.metadata(topics=["t"])
user_data = StickyAssignorUserDataV1.decode(md.user_data)
print("generation on the wire:", user_data.generation) # -1 in production

# grep proves the callback is never reached from the coordinator:
import aiokafka, subprocess, pathlib
root = pathlib.Path(aiokafka.__file__).parent
hits = subprocess.check_output(["grep", "-rn", "on_generation_assignment", str(root)], text=True)
print(hits) # only the definition line
```

### Expected behavior

`assignor.on_generation_assignment(generation)` is called after every successful join-complete, so `cls.generation` reflects the current group generation and the sticky tiebreaker works as documented in KIP-54.

### Actual behavior

`cls.generation` stays at `-1` for the lifetime of the process. Sticky degrades to best-effort.

### Environment

- aiokafka 0.13.0
- Confirmed via grep of the installed 0.13.0 wheel.

### Suggested fix

Call `assignor.on_generation_assignment(generation)` from `GroupCoordinator._on_join_complete`, right next to the existing `assignor.on_assignment(assignment)` at line 488. The generation is available in the surrounding join-complete flow. The reference Java client (`kafka-clients`) invokes both callbacks — this appears to be a missing wire-up in the Python port.

### Related

- Filed separately: [`StickyPartitionAssignor` class-level state is unsafe with multiple `AIOKafkaConsumer` in the same process](https://github.com/aio-libs/aiokafka/issues/1175). Fixing the class-state issue is prerequisite to `on_generation_assignment` doing anything useful — otherwise the callback would just clobber the shared class slot the same way `on_assignment` does today.

Guida per i contributori

Apri la guida per i contributori

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.