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