flink-deeper/storm: CWE-407 findings

flink-0006: MultipleInputNodeCreationProcessor.createMultipleInputNode() group.members ArrayList.contains inside nested loop O(M2*I); fix: HashSet membersSet before loop
flink-0007: MLPredictTypeStrategy.validateTableAndDescriptorArguments() tableFieldNames List<String>.contains in for loop O(D*T); fix: HashSet<String> conversion before loop
storm-0001: WorkerState.refreshConnections() localTaskIds ArrayList<Integer>.contains in per-task loop O(T_out*L); fix: change to HashSet<Integer>

defects/storm/patch/storm-0001-worker-localtaskids-hashset.patch

@@ -0,0 +1,27 @@
+# UNDF: (leave blank)
+--- a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
++++ b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
+@@ -109,7 +109,7 @@ public class WorkerState {
+     // local executors and localTaskIds running in this worker
+-    final ArrayList<Integer> localTaskIds;
++    final Set<Integer> localTaskIds;
+
+@@ -190,7 +190,7 @@ public class WorkerState {
+-        this.localTaskIds = new ArrayList<>();
++        this.localTaskIds = new HashSet<>();
+         for (List<Long> executor : executors) {
+             List<Integer> taskIds = StormCommon.executorIdToTasks(executor);
+             this.localTaskIds.addAll(taskIds);
+         }
+-        Collections.sort(localTaskIds);
+
+@@ -424,7 +424,7 @@ public class WorkerState {
+             for (Map.Entry<Integer, NodeInfo> taskToNodePortEntry : taskToNodePort.entrySet()) {
+                 Integer task = taskToNodePortEntry.getKey();
+                 if (outboundTasks.contains(task)) {
+                     newTaskToNodePort.put(task, taskToNodePortEntry.getValue());
+-                    if (!localTaskIds.contains(task)) {
++                    if (!localTaskIds.contains(task)) {  /* O(1) now that localTaskIds is HashSet */
+                         neededConnections.add(taskToNodePortEntry.getValue());
+                     }
+                 }

defects/storm/unit/StormTest.java

@@ -0,0 +1,123 @@
+import java.util.*;
+
+/**
+ * CWE-407 unit test for Apache Storm deeper scan (storm-0001).
+ *
+ * storm-0001: WorkerState.refreshConnections()
+ *   storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
+ *
+ *   localTaskIds is declared as ArrayList<Integer> and populated from the worker's executors.
+ *   In refreshConnections() (called on a recurring timer in every Storm worker process),
+ *   the code iterates over all outbound tasks and calls localTaskIds.contains(task):
+ *
+ *     for (Map.Entry<Integer, NodeInfo> entry : taskToNodePort.entrySet()) {  // O(T_out)
+ *         Integer task = entry.getKey();
+ *         if (outboundTasks.contains(task)) {         // O(1) — Set
+ *             ...
+ *             if (!localTaskIds.contains(task)) {     // O(L) — ArrayList linear scan — defect
+ *                 neededConnections.add(...);
+ *             }
+ *
+ *   Total per refresh: O(T_out * L) where T_out = outbound task count, L = local task count.
+ *   Since refreshConnections() is called every second (TOPOLOGY_WORKER_RECEIVER_THREAD_COUNT
+ *   timer), this is a sustained hot path in every running Storm worker.
+ *
+ *   Fix: change localTaskIds from ArrayList<Integer> to HashSet<Integer>.
+ *   The Collections.sort(localTaskIds) call (used for deterministic ordering) is removed;
+ *   callers that require ordered output (e.g. getLocalTaskIds()) can return a sorted copy
+ *   on demand when needed.
+ */
+public class StormTest {
+
+    /** Simulate defect: ArrayList.contains in refreshConnections loop */
+    static int simulateRefreshConnections_defect(
+            List<Integer> localTaskIds,      // ArrayList — defect
+            Set<Integer> outboundTasks,
+            Map<Integer, String> taskToNodePort) {
+        Set<String> neededConnections = new HashSet<>();
+        for (Map.Entry<Integer, String> entry : taskToNodePort.entrySet()) {
+            Integer task = entry.getKey();
+            if (outboundTasks.contains(task)) {          // O(1) — HashSet
+                if (!localTaskIds.contains(task)) {       // O(L) — defect
+                    neededConnections.add(entry.getValue());
+                }
+            }
+        }
+        return neededConnections.size();
+    }
+
+    /** Simulate fix: HashSet.contains is O(1) */
+    static int simulateRefreshConnections_fixed(
+            Set<Integer> localTaskIds,       // HashSet — fix
+            Set<Integer> outboundTasks,
+            Map<Integer, String> taskToNodePort) {
+        Set<String> neededConnections = new HashSet<>();
+        for (Map.Entry<Integer, String> entry : taskToNodePort.entrySet()) {
+            Integer task = entry.getKey();
+            if (outboundTasks.contains(task)) {          // O(1)
+                if (!localTaskIds.contains(task)) {       // O(1) — fix
+                    neededConnections.add(entry.getValue());
+                }
+            }
+        }
+        return neededConnections.size();
+    }
+
+    static void testStorm0001() throws Exception {
+        // T_out = outbound tasks (topology-wide tasks visible to worker)
+        // L     = local tasks per worker
+        int T_out = 1000;
+        int L = 20;
+
+        // Populate local task IDs
+        List<Integer> localTaskIdsList = new ArrayList<>();
+        Set<Integer> localTaskIdsSet = new HashSet<>();
+        for (int i = 0; i < L; i++) {
+            localTaskIdsList.add(i);
+            localTaskIdsSet.add(i);
+        }
+
+        // outboundTasks is a HashSet of all tasks this worker routes to
+        Set<Integer> outboundTasks = new HashSet<>();
+        for (int i = 0; i < T_out; i++) outboundTasks.add(i);
+
+        // taskToNodePort: topology assignment map (T_out entries)
+        Map<Integer, String> taskToNodePort = new LinkedHashMap<>();
+        for (int i = 0; i < T_out; i++) {
+            taskToNodePort.put(i, "node" + (i % 50) + ":6700");
+        }
+
+        // Warm up
+        simulateRefreshConnections_defect(localTaskIdsList, outboundTasks, taskToNodePort);
+        simulateRefreshConnections_fixed(localTaskIdsSet, outboundTasks, taskToNodePort);
+
+        // Simulate timer-based refresh: many iterations
+        int RUNS = 5000;
+
+        long t0 = System.nanoTime();
+        for (int r = 0; r < RUNS; r++) {
+            simulateRefreshConnections_defect(localTaskIdsList, outboundTasks, taskToNodePort);
+        }
+        long defectNs = System.nanoTime() - t0;
+
+        long t1 = System.nanoTime();
+        for (int r = 0; r < RUNS; r++) {
+            simulateRefreshConnections_fixed(localTaskIdsSet, outboundTasks, taskToNodePort);
+        }
+        long fixedNs = System.nanoTime() - t1;
+
+        double ratio = (double) defectNs / Math.max(fixedNs, 1);
+        System.out.printf("storm-0001 [T_out=%d L=%d RUNS=%d]: defect=%.1fms fixed=%.1fms ratio=%.1fx%n",
+                T_out, L, RUNS, defectNs / 1e6, fixedNs / 1e6, ratio);
+
+        if (ratio < 1.5) {
+            throw new AssertionError("Expected defect to be slower; ratio=" + ratio);
+        }
+        System.out.println("storm-0001 PASS");
+    }
+
+    public static void main(String[] args) throws Exception {
+        testStorm0001();
+        System.out.println("ALL PASS");
+    }
+}