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)

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

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

// 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

// 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.

3.2 KiB Raw Permalink Blame History Unescape Escape