wave13: 514/239 — flink/nifi/artemis + K8s/cilium/linkerd2 + ES/OpenSearch/Solr + hadoop/hbase/spark

defects/spark/patch/spark-0003-ticket.md

@@ -0,0 +1,57 @@
+# spark-0003: Spark Standalone Master — completedApps ArrayBuffer.contains() O(n²) on worker failure
+
+## Severity
+MEDIUM — called on each worker failure; O(A × C) where A = running apps, C = completed apps;
+materializes in long-running clusters where C grows to spark.deploy.retainedApplications (default 200)
+
+## File
+`core/src/main/scala/org/apache/spark/deploy/master/Master.scala`
+
+## Lines
+83 (`completedApps` field declaration), 1114 (`apps.filterNot(completedApps.contains(_))`)
+
+## Pattern
+CWE-407: O(n) ArrayBuffer.contains() used as predicate in filterNot over all running apps.
+
+```scala
+// DEFECTIVE (line 83)
+private val completedApps = new ArrayBuffer[ApplicationInfo]
+
+// DEFECTIVE (line 1114) — inside workerRemoved(worker), called for every worker failure
+apps.filterNot(completedApps.contains(_)).foreach { app =>
+    ...  // notify app of lost worker
+}
+```
+
+`completedApps` is a `mutable.ArrayBuffer`. `filterNot(completedApps.contains(_))` iterates
+every element of `apps` (HashSet[ApplicationInfo]) and calls `completedApps.contains(app)` —
+O(completedApps.size) per app. Total: O(|apps| × |completedApps|).
+
+By default, Spark retains 200 completed applications (`spark.deploy.retainedApplications`).
+Each worker failure call becomes O(200 × A) where A is the current number of running apps.
+In a large Spark cluster with 100 running apps and 200 completed: 20,000 comparisons instead of 100.
+
+## Fix
+
+Pre-build a `HashSet` before the filter, or use `toSet`:
+
+```scala
+// FIXED — option 1: convert at call site
+val completedAppsSet = completedApps.toSet
+apps.filterNot(completedAppsSet.contains(_)).foreach { app =>
+
+// FIXED — option 2: maintain as HashSet
+private val completedApps = new mutable.HashSet[ApplicationInfo]
+```
+
+Option 2 is preferred as `completedApps` is only ever tested for membership or iterated;
+changing to HashSet gives O(1) contains with no behavior change.
+
+## Complexity
+- Before: O(A × C) per worker failure event
+- After: O(A) per worker failure event
+
+## Impact
+Worker failures trigger master-level app notification. In a cluster under stress (many worker
+failures), this compounds — each failure event is more expensive just as cluster load peaks.
+With 200 completed apps, this is a 200x overhead per failure event.

defects/spark/patch/spark-0003.patch

@@ -0,0 +1,18 @@
+diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
+--- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
++++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
+@@ -80,7 +80,8 @@ private[deploy] class Master(
+   val apps = new HashSet[ApplicationInfo]
+
+-  private val completedApps = new ArrayBuffer[ApplicationInfo]
++  // CWE-407 fix: HashSet for O(1) contains(); completedApps is only ever tested for membership.
++  private val completedApps = new mutable.HashSet[ApplicationInfo]
+   private var nextAppNumber = 0
+
+@@ -1111,7 +1111,7 @@ private[deploy] class Master(
+     logInfo(log"Telling app of lost worker: ${MDC(LogKeys.WORKER_ID, worker.id)}")
+-    apps.filterNot(completedApps.contains(_)).foreach { app =>
++    apps.filterNot(completedApps).foreach { app =>
+       app.driver.send(WorkerRemoved(worker.id, worker.host, message))
+     }
+   }

defects/spark/unit/MasterCompletedAppsAlgorithm.java

@@ -0,0 +1,128 @@
+package unit;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Unit test for spark-0003: Master.completedApps ArrayBuffer.contains() O(n²) on worker failure.
+ *
+ * Standalone — no JUnit. Run: javac -d . MasterCompletedAppsAlgorithm.java && java -ea unit.MasterCompletedAppsAlgorithm
+ */
+public class MasterCompletedAppsAlgorithm {
+
+    // ---------- Slow: ArrayList.contains inside filterNot ----------
+    static long slowFilterNot(Set<Integer> apps, List<Integer> completedApps) {
+        long ops = 0;
+        List<Integer> runningApps = new ArrayList<>();
+        for (Integer app : apps) {
+            ops += completedApps.size(); // O(size) per contains
+            if (!completedApps.contains(app)) {
+                runningApps.add(app);
+            }
+        }
+        return ops;
+    }
+
+    // ---------- Fast: HashSet.contains O(1) ----------
+    static long fastFilterNot(Set<Integer> apps, Set<Integer> completedAppsSet) {
+        long ops = 0;
+        List<Integer> runningApps = new ArrayList<>();
+        for (Integer app : apps) {
+            ops += 1; // O(1) per contains
+            if (!completedAppsSet.contains(app)) {
+                runningApps.add(app);
+            }
+        }
+        return ops;
+    }
+
+    static void testCorrectness() {
+        int A = 50;  // running apps
+        int C = 20;  // completed apps
+
+        Set<Integer> apps = new HashSet<>();
+        for (int i = 0; i < A + C; i++) apps.add(i);
+
+        List<Integer> completedList = new ArrayList<>();
+        for (int i = A; i < A + C; i++) completedList.add(i);
+        Set<Integer> completedSet = new HashSet<>(completedList);
+
+        // Compute running apps both ways
+        Set<Integer> slowRunning = new HashSet<>();
+        for (Integer app : apps) {
+            if (!completedList.contains(app)) slowRunning.add(app);
+        }
+        Set<Integer> fastRunning = new HashSet<>();
+        for (Integer app : apps) {
+            if (!completedSet.contains(app)) fastRunning.add(app);
+        }
+
+        assert slowRunning.equals(fastRunning) :
+            "Result mismatch: slow=" + slowRunning.size() + " fast=" + fastRunning.size();
+        assert slowRunning.size() == A :
+            "Expected " + A + " running apps, got " + slowRunning.size();
+        System.out.println("PASS: correctness — filterNot produces same " + A + " running apps");
+    }
+
+    static void testPerformance() {
+        int A = 100;  // running apps
+        int C = 200;  // completed apps (default retainedApplications)
+
+        Set<Integer> apps = new HashSet<>();
+        for (int i = 0; i < A + C; i++) apps.add(i);
+
+        List<Integer> completedList = new ArrayList<>();
+        Set<Integer> completedSet = new HashSet<>();
+        for (int i = A; i < A + C; i++) {
+            completedList.add(i);
+            completedSet.add(i);
+        }
+
+        long slowOps = slowFilterNot(apps, completedList);
+        long fastOps = fastFilterNot(apps, completedSet);
+        double ratio = (double) slowOps / fastOps;
+
+        System.out.printf("filterNot A=%d C=%d: slow_ops=%,d fast_ops=%,d ratio=%.1fx%n",
+            A, C, slowOps, fastOps, ratio);
+        assert ratio >= 10.0 :
+            "Expected >=10x speedup, got ratio=" + ratio;
+        System.out.println("PASS: performance — ratio >= 10x");
+    }
+
+    static void testLargeRetainedApps() {
+        // Stress test: many completed apps retained
+        int A = 500;
+        int C = 1000;
+
+        Set<Integer> apps = new HashSet<>();
+        for (int i = 0; i < A + C; i++) apps.add(i);
+
+        List<Integer> completedList = new ArrayList<>();
+        Set<Integer> completedSet = new HashSet<>();
+        for (int i = A; i < A + C; i++) {
+            completedList.add(i);
+            completedSet.add(i);
+        }
+
+        long slowOps = slowFilterNot(apps, completedList);
+        long fastOps = fastFilterNot(apps, completedSet);
+        double ratio = (double) slowOps / fastOps;
+
+        System.out.printf("filterNot A=%d C=%d: slow_ops=%,d fast_ops=%,d ratio=%.1fx%n",
+            A, C, slowOps, fastOps, ratio);
+        assert ratio >= 100.0 :
+            "Expected >=100x speedup, got ratio=" + ratio;
+        System.out.println("PASS: large retained — ratio >= 100x");
+    }
+
+    public static void main(String[] args) {
+        testCorrectness();
+        testPerformance();
+        testLargeRetainedApps();
+        System.out.println("ALL PASS (3/3)");
+    }
+}