# UNDF: UNDF-2026-000000435
# kafka-0003: RoundRobinAssignor — topics().contains() inside while-in-for loop

## Defect ID
kafka-0003

## File:Line
`clients/src/main/java/org/apache/kafka/clients/consumer/RoundRobinAssignor.java:118`

## Description
`Subscription.topics()` returns `List<String>`. The `assign()` method iterates over
all partitions in an outer `for` loop, and inside that loop calls a `while` that
calls `subscriptions.get(assigner.peek().memberId).topics().contains(topic)`.

Each `.contains()` call is O(T) (linear scan over the topic subscription list).
The while loop can spin up to M times per partition. Total: O(P × M × T).

With P=10,000 partitions, M=200 consumers, T=50 topics per consumer → ~100M operations
for a single partition assignment pass. Fix: convert each subscription's topics to
`HashSet<String>` at construction time so `.contains()` is O(1).

## Complexity
- Slow: O(P × M × T) — List.contains() = O(T) per call
- Fast: O(P × M) — HashSet.contains() = O(1) per call

## Severity
HIGH

## Speedup Estimate
~T× improvement = 50x at T=50 topics/consumer; scales with topic count.

## Fix
In `assign()`, before the outer loop, build a `Map<String, Set<String>> memberTopics`
from the subscription list, replacing the inner `topics().contains(topic)` call.

```diff
-            while (!subscriptions.get(assigner.peek().memberId).topics().contains(topic))
+            while (!memberTopics.get(assigner.peek().memberId).contains(topic))
```

Where `memberTopics` is pre-built as `HashMap<>(subscriptions.size())` with
`HashSet` values from `subscription.topics()`.
