moad-sweep: 9 initial findings — MOAD-0005 (Hungry Regex) x1, MOAD-0006 (Thundering Herd) x8

MOAD-0005 candidates (ReDoS):
- bleach-MOADX-0001: sanitize_css O(2^N) — 71-char input causes 12s hang HIGH

MOAD-0006 candidates (Thundering Herd / cache stampede):
- hibernate-orm-MOADX-0001: QueryInterpretationCacheStandardImpl HQL plan cache HIGH
- elasticsearch-MOADX-0001: EnrichCache "intentionally non-locking" enrich search HIGH
- hadoop-MOADX-0001: FederationJCache/FederationCaffeineCache YARN default config HIGH
- celery-MOADX-0001: BaseBackend.get_task_meta chord fan-in Redis stampede HIGH
- traefik-MOADX-0001: CNAMEFlatten data race + N*30s DNS stampede HIGH
- kafka-MOADX-0001: CachedConnectors.lookup classloader scan MEDIUM
- spring-MOADX-0001: AbstractFallbackCacheOperationSource @Cacheable cold start MEDIUM
- django-MOADX-0001: cached.Loader.get_template template compile MEDIUM

CLEAN: Consul (singleflight), Kubernetes (RWMutex), Prometheus (per-scraper),
Pulsar (AsyncLoadingCache), Cassandra (LoadingCache), Flink (CAS loop),
ZooKeeper (synchronized), Airflow (hand-rolled singleflight)

defects/kafka/patch/kafka-MOADX-0001-connect-cached-connectors-stampede.md

@@ -0,0 +1,93 @@
+# UNDF: (pending)
+# kafka-MOADX-0001: CachedConnectors.lookup — TOCTOU stampede on connector class loading
+
+## MOAD-0006 Candidate — The Thundering Herd
+
+| Field | Value |
+|-------|-------|
+| ID | kafka-MOADX-0001 |
+| Severity | MEDIUM |
+| Ecosystem | Apache Kafka Connect |
+| File | `connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java` lines 57–73 |
+| Cache type | Connector class instance cache (connector name + version → Connector object) |
+| Pattern | containsKey+get (non-atomic) → null check → plugins.newConnector() (classloader scan + instantiation) → put (unsynchronized) |
+| Trigger | N concurrent connector-start requests for the same connector type all fail the containsKey check → N classloader scans and instantiations |
+| Hot path | `WorkerConnector.start()` and `Worker.startConnector()` — called once per connector instance startup |
+
+## Defect
+
+`CachedConnectors.lookup` performs a check-then-act that is not atomic, allowing multiple threads to all miss the cache for the same (connectorName, version) key simultaneously:
+
+```java
+private Connector lookup(String connectorName, VersionRange range) {
+    String version = range == null ? LATEST_VERSION : range.toString();
+    // TOCTOU: containsKey + get is not atomic — another thread can insert between these two calls
+    if (connectors.containsKey(connectorName) && connectors.get(connectorName).containsKey(version)) {
+        return connectors.get(connectorName).get(version);   // (1) all N threads: miss (concurrent startup)
+    }
+
+    try {
+        Connector connector = plugins.newConnector(connectorName, range);  // (2) N×classloader scan + instantiation
+        connectors.computeIfAbsent(connectorName, k -> new ConcurrentHashMap<>()).put(version, connector);  // (3) N puts
+        return connector;
+    } catch (VersionedPluginLoadingException e) {
+        invalidVersions.computeIfAbsent(connectorName, k -> new ConcurrentHashMap<>()).put(version, e);
+        throw e;
+    } catch (Exception e) {
+        invalidConnectors.put(connectorName, e);
+        throw e;
+    }
+}
+```
+
+### What `plugins.newConnector` does
+
+`plugins.newConnector(connectorName, range)` calls `pluginLoader(classOrAlias, range, sourceLoader).loadClass(classOrAlias)` which:
+1. Scans all plugin sources for the connector class (classpath scan if using reflective discovery)
+2. Creates an isolated classloader (`DelegatingClassLoader`) for the plugin
+3. Instantiates the connector via reflection
+
+This is CPU-intensive (classpath scan can take 10–500ms depending on plugin count) and involves `synchronized` blocks inside the JVM class loading machinery. Under concurrent load, multiple threads competing for the same class create contention inside `ClassLoader.loadClass`.
+
+### Scenario
+
+A Kafka Connect cluster running in distributed mode adds a new connector with 100 tasks. All 100 task startup requests arrive at the `Worker` simultaneously. Each calls `getConnector(connectorName, null)` → `lookup()`. All 100 threads fail the `containsKey` check and all fire `plugins.newConnector()`. The JVM class loading lock then serializes them, but the outer scan work is duplicated 100×.
+
+### Note on outer `connectors` map
+
+`connectors` is a `ConcurrentHashMap<String, Map<String, Connector>>` but the inner map is a freshly created `ConcurrentHashMap` added via `computeIfAbsent`. The race is: thread A checks `containsKey`, thread B also checks `containsKey` (both miss), both call `newConnector`, both try to `computeIfAbsent` the inner map and `put`. Both complete — correctness is preserved but the redundant class loading is wasteful.
+
+## Impact
+
+- On startup with many connector tasks: N redundant classloader scans and instantiations
+- JVM class loading lock contention under the concurrent instantiation attempts
+- Connector plugins that have expensive `init()` or static initializers run N times
+- MEDIUM severity because this is not a per-request hot path — it happens at connector start only, not per-message
+
+## Fix
+
+Use `computeIfAbsent` directly to make the check-and-insert atomic:
+
+```java
+private Connector lookup(String connectorName, VersionRange range) {
+    String version = range == null ? LATEST_VERSION : range.toString();
+    // Atomic: computeIfAbsent ensures loader is called exactly once per (name, version) pair
+    Map<String, Connector> versionMap = connectors.computeIfAbsent(
+        connectorName, k -> new ConcurrentHashMap<>());
+    return versionMap.computeIfAbsent(version, v -> {
+        try {
+            return plugins.newConnector(connectorName, range);
+        } catch (VersionedPluginLoadingException e) {
+            invalidVersions.computeIfAbsent(connectorName, k -> new ConcurrentHashMap<>()).put(version, e);
+            throw e;
+        } catch (Exception e) {
+            invalidConnectors.put(connectorName, e);
+            throw e;
+        }
+    });
+}
+```
+
+Note: `computeIfAbsent` on `ConcurrentHashMap` guarantees the mapping function is called at most once per key per miss (per the Java 8+ specification). This eliminates both the redundant instantiation and the TOCTOU window.
+
+The error-caching paths (`invalidConnectors`, `invalidVersions`) should similarly be guarded, but errors are less likely to be concurrent.