# UNDF: UNDF-2026-000000437
# kafka-0006 — Kafka Streams DefaultTaskManager: ArrayList lockedTasks O(T×L) in hot scheduling loop

## Metadata
- **Project**: Apache Kafka
- **Component**: `streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskManager.java`
- **CWE**: CWE-407 (Inefficient Algorithmic Complexity)
- **Severity**: MEDIUM
- **Complexity**: O(T × L) → O(T) where T = active tasks, L = locked tasks
- **Hot path**: `assignNextTask()` is called by every `TaskExecutor` thread on every scheduling cycle

## Location

```
streams/src/main/java/org/apache/kafka/streams/processor/internals/tasks/DefaultTaskManager.java
```

### Defective code — lockedTasks is ArrayList (line 62)

```java
// Line 62
private final List<TaskId> lockedTasks = new ArrayList<>();
```

### Defective code — O(T × L) in assignNextTask() (lines 105–118)

```java
// Line 100: taskExecutors is also ArrayList — O(E) linear scan on every call
if (!taskExecutors.contains(executor)) {
    throw new IllegalArgumentException("...");
}

// Lines 105–118: for every active task, does O(L) linear scan of lockedTasks ArrayList
for (final StreamTask task : tasks.activeInitializedTasks()) {
    if (!assignedTasks.containsKey(task.id()) &&
        !lockedTasks.contains(task.id()) &&       // <-- O(L) ArrayList.contains per task
        canProgress(task, time.milliseconds()) &&
        !hasUncaughtException(task.id())
    ) {
        assignedTasks.put(task.id(), executor);
        return task;
    }
}
```

Also at lines 131, 289 (same pattern in `awaitProcessableTasks`, `remove`).

## Fix

```java
// Change field declaration:
// BEFORE:
private final List<TaskId> lockedTasks = new ArrayList<>();
private final List<TaskExecutor> taskExecutors;

// AFTER:
private final Set<TaskId> lockedTasks = new HashSet<>();
private final List<TaskExecutor> taskExecutors;  // small, bounded by numExecutors — OK
```

The `lockedTasks` field is used only for membership tests (`contains`) and bulk add/remove. A `HashSet<TaskId>` provides O(1) `contains` and `add`. The `taskExecutors` list is bounded by `numExecutors` (typically 1–4) so its linear scan is negligible; leave it as-is unless further optimization is desired.

All callers of `lockedTasks`:
- `lockTasks(Set<TaskId>)` — calls `lockedTasks.addAll(taskIds)` — works with HashSet
- `unlockTasks(Set<TaskId>)` — calls `lockedTasks.removeAll(taskIds)` — works with HashSet
- `lockedTasks.contains(task.id())` — O(1) with HashSet
- `lockedTasks.contains(taskId)` — O(1) with HashSet

No iterator order dependency exists — callers only test membership or add/remove sets.

## Complexity analysis

| Scenario | Before | After |
|----------|--------|-------|
| T active tasks, L locked tasks | O(T × L) | O(T) |
| T=1000 tasks, L=500 locked | 500,000 ops per scheduling cycle | 1,000 ops |
| Rebalancing with many locked tasks | Stalls executor threads | Minimal overhead |

## Notes

During rebalancing events, `lockTasks()` is called with a large set of task IDs to prevent task executors from processing them while partition reassignment occurs. At exactly this moment, the per-scheduling-cycle `assignNextTask()` loop pays the full O(T × L) cost on every executor thread wakeup.
