wave11: 488/227 — ROS2/OpenCV/Open3D/metaflow/kubeflow/optuna/openssl/mbedtls/wolfssl + Kafka Streams/Pulsar

defects/kafka/patch/kafka-0006-streams-defaulttaskmanager-lockedtasks-hashset.md

@@ -0,0 +1,80 @@
+# 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.

defects/kafka/unit/KafkaStreamsDefaultTaskManagerTest.java

@@ -0,0 +1,130 @@
+package unit;
+
+import java.util.*;
+
+/**
+ * kafka-0006: DefaultTaskManager lockedTasks ArrayList → HashSet
+ *
+ * Demonstrates that ArrayList.contains() inside a task-scheduling loop is O(T×L)
+ * while HashSet.contains() is O(T).
+ *
+ * Compile: javac -d . KafkaStreamsDefaultTaskManagerTest.java
+ * Run:     java unit.KafkaStreamsDefaultTaskManagerTest
+ */
+public class KafkaStreamsDefaultTaskManagerTest {
+
+    // Simulated TaskId (Integer wrapper for simplicity)
+    static class TaskId {
+        final int id;
+        TaskId(int id) { this.id = id; }
+
+        @Override public boolean equals(Object o) {
+            return o instanceof TaskId && ((TaskId) o).id == this.id;
+        }
+        @Override public int hashCode() { return Integer.hashCode(id); }
+        @Override public String toString() { return "Task-" + id; }
+    }
+
+    // --- SLOW: ArrayList lockedTasks (defective, mirrors DefaultTaskManager line 62) ---
+    static class DefectiveTaskManager {
+        private final List<TaskId> lockedTasks = new ArrayList<>();
+        private final List<TaskId> activeTasks;
+
+        DefectiveTaskManager(List<TaskId> activeTasks) {
+            this.activeTasks = activeTasks;
+        }
+
+        void lockTasks(Set<TaskId> ids) { lockedTasks.addAll(ids); }
+
+        // Returns count of assignable tasks — mirrors assignNextTask() loop body
+        int countAssignable() {
+            int count = 0;
+            for (TaskId task : activeTasks) {
+                if (!lockedTasks.contains(task)) {  // O(L) ArrayList scan
+                    count++;
+                }
+            }
+            return count;
+        }
+    }
+
+    // --- FAST: HashSet lockedTasks (fixed) ---
+    static class FixedTaskManager {
+        private final Set<TaskId> lockedTasks = new HashSet<>();
+        private final List<TaskId> activeTasks;
+
+        FixedTaskManager(List<TaskId> activeTasks) {
+            this.activeTasks = activeTasks;
+        }
+
+        void lockTasks(Set<TaskId> ids) { lockedTasks.addAll(ids); }
+
+        int countAssignable() {
+            int count = 0;
+            for (TaskId task : activeTasks) {
+                if (!lockedTasks.contains(task)) {  // O(1) HashSet lookup
+                    count++;
+                }
+            }
+            return count;
+        }
+    }
+
+    static long bench(Runnable r, int iters) {
+        // warmup
+        for (int i = 0; i < 3; i++) r.run();
+        long t0 = System.nanoTime();
+        for (int i = 0; i < iters; i++) r.run();
+        return System.nanoTime() - t0;
+    }
+
+    public static void main(String[] args) {
+        System.out.println("kafka-0006: DefaultTaskManager lockedTasks ArrayList → HashSet");
+        System.out.println("=".repeat(65));
+
+        int[] sizes = {100, 500, 1000};
+        int iters = 500;
+        boolean allPass = true;
+
+        for (int n : sizes) {
+            // Build n active tasks, lock half
+            List<TaskId> active = new ArrayList<>();
+            Set<TaskId> toLock = new HashSet<>();
+            for (int i = 0; i < n; i++) {
+                TaskId t = new TaskId(i);
+                active.add(t);
+                if (i % 2 == 0) toLock.add(t);
+            }
+
+            DefectiveTaskManager slow = new DefectiveTaskManager(active);
+            slow.lockTasks(toLoad(toLoad(toLock)));
+            FixedTaskManager fast = new FixedTaskManager(active);
+            fast.lockTasks(toLoad(toLock));
+
+            // Correctness check
+            int slowResult = slow.countAssignable();
+            int fastResult = fast.countAssignable();
+            boolean correct = (slowResult == fastResult) && (slowResult == n / 2);
+            if (!correct) allPass = false;
+
+            long slowNs = bench(slow::countAssignable, iters);
+            long fastNs = bench(fast::countAssignable, iters);
+            double ratio = (double) slowNs / fastNs;
+
+            System.out.printf("N=%-5d  slow=%7.2f ms  fast=%7.2f ms  ratio=%.1fx  assignable=%d  %s%n",
+                n,
+                slowNs / 1_000_000.0 / iters,
+                fastNs / 1_000_000.0 / iters,
+                ratio,
+                slowResult,
+                correct ? "PASS" : "FAIL(result mismatch)");
+        }
+
+        System.out.println("=".repeat(65));
+        System.out.println(allPass ? "ALL PASS" : "SOME FAILED");
+        if (!allPass) System.exit(1);
+    }
+
+    // Helper: copy set into set (simulate addAll contract)
+    static Set<TaskId> toLoad(Set<TaskId> s) { return s; }
+}