SteNicholas commented on code in PR #3737:
URL: https://github.com/apache/celeborn/pull/3737#discussion_r3500113552
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -90,7 +93,7 @@ public static ShuffleClient get(
CelebornConf conf,
UserIdentifier userIdentifier,
byte[] extension) {
- if (null == _instance || !initialized) {
+ if (null == _instance || !initialized || !Objects.equals(appUniqueId,
_appUniqueId)) {
Review Comment:
`get()` is now app-aware, but the lock-free fast path still has a TOCTOU
that can hand back the **wrong app's** client.
The in-lock `return _instance` (134) is safe, but the fast-path `return
_instance` (137) reads the static volatile without the lock. The publish-order
comment (127-129) only closes the *new-id + stale-instance* direction; writing
`_instance` (130) before `_appUniqueId` (131) opens the reverse window:
- Thread A is in branch-3 for app A: it has run `_instance = A` (130) but
not yet `_appUniqueId = A` (131), so `_appUniqueId` is still `B`.
- Concurrent thread B calls `get(B)`: the guard here sees `_appUniqueId ==
B` → `Objects.equals(B,B)` true → guard false → B skips the lock and `return
_instance` (137) returns **A's** client.
B (the live app) then drives its shuffle against A's `LifecycleManager` →
ArrayIndexOutOfBounds / CommitMetadata CRC mismatch, i.e. the cross-app
corruption this PR is fixing. Only reachable in the multi-app spark-it JVM
(prod uses one `appUniqueId` per JVM, so branch-3 never fires there) — but that
is exactly the scenario branch-3 was added for.
Fix: return the instance captured under the lock (each branch assigns a
local `result`, then `return result;`), or hold `(appId, client)` in a single
`AtomicReference` so id and instance can't be read torn.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala:
##########
@@ -225,17 +226,30 @@ trait MiniClusterFeature extends Logging {
workerStarted = true
worker.initialize()
} catch {
+ case ie: InterruptedException =>
+
Utils.tryLogNonFatalError(worker.stop(CelebornExitKind.EXIT_IMMEDIATELY))
+ Utils.tryLogNonFatalError(worker.rpcEnv.shutdown())
+ Thread.currentThread().interrupt()
+ throw ie
case ex: Exception =>
- if (workers(i - 1) != null) {
- workers(i - 1).shutdownGracefully()
- }
+ Utils.tryLogNonFatalError(worker.exitImmediately())
+
Utils.tryLogNonFatalError(worker.stop(CelebornExitKind.EXIT_IMMEDIATELY))
+ Utils.tryLogNonFatalError(worker.rpcEnv.shutdown())
workerStarted = false
workerStartRetry += 1
logError(s"cannot start worker $i, retrying: ", ex)
if (workerStartRetry == maxRetries) {
logError(s"cannot start worker $i, reached to max retrying",
ex)
throw ex
}
+ try {
+ TimeUnit.SECONDS.sleep(Math.pow(2, workerStartRetry).toLong)
Review Comment:
`Math.pow(2, workerStartRetry)` seconds is OK today (`maxRetries = 4` →
2+4+8 = 14s before the final throw), but 14s of blind backoff plus the
per-attempt stop/shutdown already eats a large slice of the 60s
`workersWaitingTimeoutMs` window — so a worker that retries a couple of times
can itself trip the startup timeout this PR is trying to harden. It's also
brittle: raising `maxRetries` makes this grow as 2^(retry-1) s (e.g.
`maxRetries = 10` → a 512s sleep). Consider clamping it, e.g. `min(Math.pow(2,
workerStartRetry).toLong, <few seconds>)`.
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -102,14 +105,33 @@ public static ShuffleClient get(
_instance = new ShuffleClientImpl(appUniqueId, conf, userIdentifier);
_instance.setupLifecycleManagerRef(driverHost, port);
_instance.setExtension(extension);
+ _appUniqueId = appUniqueId;
initialized = true;
} else if (!initialized) {
_instance.shutdown();
_instance = new ShuffleClientImpl(appUniqueId, conf, userIdentifier);
_instance.setupLifecycleManagerRef(driverHost, port);
_instance.setExtension(extension);
+ _appUniqueId = appUniqueId;
+ initialized = true;
+ } else if (!Objects.equals(appUniqueId, _appUniqueId)) {
+ // Do NOT shutdown() the old _instance. Callers cache the reference
returned by get(),
+ // and shutdown() is an immediate teardown that would terminate the
RpcEnv/pools still in
+ // use, causing RejectedExecutionException. Teardown is owned by
stop()->shutdown(). The
+ // orphan is bounded (one per appUniqueId) and unreachable in normal
single-app JVMs.
+ // The spark-it suite runs multiple apps in one reused JVM with
overlapping lifecycles, so
+ // a shutdown() here tears down an instance still in use by the
previous app and fails.
+ ShuffleClientImpl newInstance = new ShuffleClientImpl(appUniqueId,
conf, userIdentifier);
Review Comment:
branch-3 replaces `_instance` without `shutdown()`-ing the old one, so its
RpcEnv (bound port + dispatcher threads), Netty `dataClientFactory`,
`pushDataRetryPool` and `reviveManager` leak until JVM exit.
The comment calls the orphan "bounded (one per appUniqueId)", but under two
apps alternately calling `get()`, every call observes a mismatch and rebuilds —
so it's bounded by the number of A↔B switches, not by app count. `reset()` only
nulls the reference (no `shutdown()`), so these are never reclaimed there
either. In the long-lived serial spark-it JVM this accumulates
threads/FDs/ports, feeding the same contention this PR lowers worker counts to
fight. If teardown of an in-use instance is the concern, that's a signal the
static singleton shouldn't be shared across apps at all (per-app map / injected
client).
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala:
##########
@@ -225,17 +226,30 @@ trait MiniClusterFeature extends Logging {
workerStarted = true
worker.initialize()
} catch {
+ case ie: InterruptedException =>
+
Utils.tryLogNonFatalError(worker.stop(CelebornExitKind.EXIT_IMMEDIATELY))
+ Utils.tryLogNonFatalError(worker.rpcEnv.shutdown())
+ Thread.currentThread().interrupt()
+ throw ie
case ex: Exception =>
- if (workers(i - 1) != null) {
- workers(i - 1).shutdownGracefully()
- }
+ Utils.tryLogNonFatalError(worker.exitImmediately())
Review Comment:
The retry cleanup now calls `worker.exitImmediately()`, which is heavier
than the old `shutdownGracefully()`: it issues a **blocking**
`masterClient.askSync(WorkerLost(...))` (Worker.scala:1056) and adds the
worker's host:ports to the master's excluded list.
Two effects on the start-retry path:
- The synchronous `WorkerLost` RPC sits in the hot retry loop. If the master
is slow/contended at startup (the exact scenario this PR targets), it blocks up
to the ask timeout per failed attempt, compounding the `2^retry` backoff
against the 60s `workersWaitingTimeoutMs`.
- If a worker briefly registered before failing, `WorkerLost` excludes it on
the master; the recreated worker re-registers under new random ports, so stale
host:port exclusions accumulate on the master across retries.
A failed `initialize()` usually means the worker never came up, so
`exitImmediately()` (which assumes a registered worker reporting itself lost)
seems like the wrong teardown here — the `InterruptedException` branch above
just does `stop()` + `rpcEnv.shutdown()`, which is probably what this branch
wants too.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala:
##########
@@ -225,17 +226,30 @@ trait MiniClusterFeature extends Logging {
workerStarted = true
worker.initialize()
} catch {
+ case ie: InterruptedException =>
+
Utils.tryLogNonFatalError(worker.stop(CelebornExitKind.EXIT_IMMEDIATELY))
+ Utils.tryLogNonFatalError(worker.rpcEnv.shutdown())
+ Thread.currentThread().interrupt()
+ throw ie
case ex: Exception =>
- if (workers(i - 1) != null) {
- workers(i - 1).shutdownGracefully()
- }
+ Utils.tryLogNonFatalError(worker.exitImmediately())
+
Utils.tryLogNonFatalError(worker.stop(CelebornExitKind.EXIT_IMMEDIATELY))
+ Utils.tryLogNonFatalError(worker.rpcEnv.shutdown())
workerStarted = false
workerStartRetry += 1
logError(s"cannot start worker $i, retrying: ", ex)
if (workerStartRetry == maxRetries) {
logError(s"cannot start worker $i, reached to max retrying",
ex)
throw ex
}
+ try {
+ TimeUnit.SECONDS.sleep(Math.pow(2, workerStartRetry).toLong)
+ } catch {
+ case ie: InterruptedException =>
+ Thread.currentThread().interrupt()
+ throw ie
+ }
+ worker = createWorker(workerConf)
Review Comment:
Recreating the worker on retry leaks the **failed** worker into the returned
cluster set.
`workers(i-1) = worker` (224) is written *before* `initialize()`, and the
main poll loop does `workerInfos.put(workers(i), …)` (270) *before* the
`registered` check (272), with no removal anywhere. On a failed first attempt
the main thread can observe the failed `worker` (during its `initialize()` /
backoff window) and put it into `workerInfos`; this line then creates a **new**
object, which is also put on a later poll. `Worker` overrides neither `equals`
nor `hashCode` (identity equality), so the dead and live workers are distinct
keys and **both** survive in `workerInfos.keySet`, which `setUpWorkers`
returns. A suite iterating the set (e.g. `CelebornHashCheckDiskSuite`
`workers.foreach { _.storageManager.updateDiskInfos() }`) then operates on a
stopped worker → NPE/RejectedExecutionException, and `shutdownMiniCluster`
double-stops it. The old `val worker` (single identity, idempotent put) didn't
have this.
Related: `workers` is written under `flagUpdateLock` (223-225) but read
without it at 267 — a visibility race the new `var` reassignment makes easier
to hit. And the `InterruptedException` branch (229) tears the worker down but
does **not** recreate it, so an interrupt during `initialize()`/backoff leaves
a stopped worker pinned in `workers(i-1)` that the poll loop then inserts and
waits on until timeout. Suggest only inserting into `workerInfos` once the
worker is `registered` (move the put after the check), and/or overwriting the
prior failed entry for slot `i`.
##########
build/mvn:
##########
@@ -46,23 +47,44 @@ install_app() {
wget_opts="--progress=bar:force ${wget_opts}"
if [ -z "$3" -o ! -f "$binary" ]; then
- # check if we already have the tarball
- # check if we have curl installed
- # download application
- [ ! -f "${local_tarball}" ] && [ $(command -v curl) ] && \
- echo "exec: curl ${curl_opts} ${remote_tarball}" 1>&2 && \
- curl ${curl_opts} "${remote_tarball}" > "${local_tarball}"
- # if the file still doesn't exist, lets try `wget` and cross our fingers
- [ ! -f "${local_tarball}" ] && [ $(command -v wget) ] && \
- echo "exec: wget ${wget_opts} ${remote_tarball}" 1>&2 && \
- wget ${wget_opts} -O "${local_tarball}" "${remote_tarball}"
- # if both were unsuccessful, exit
- [ ! -f "${local_tarball}" ] && \
- echo -n "ERROR: Cannot download $2 with cURL or wget; " && \
- echo "please install manually and try again." && \
- exit 2
- cd "${_DIR}" && tar -xzf "$2"
- rm -rf "$local_tarball"
+ local attempt=1
+ while [ "${attempt}" -le "${max_attempts}" ]; do
+ # remove any partial/corrupt download left over from a previous attempt
+ rm -f "${local_tarball}"
+
+ # download application with `curl`, falling back to `wget`
+ if command -v curl >/dev/null 2>&1; then
+ echo "exec: curl ${curl_opts} ${remote_tarball}" 1>&2
+ curl ${curl_opts} "${remote_tarball}" > "${local_tarball}"
+ elif command -v wget >/dev/null 2>&1; then
Review Comment:
Three behavior changes in this rewrite worth a second look:
- `rm -f "${local_tarball}"` at the top of every attempt drops the old
"reuse an already-downloaded tarball" path. For an offline/air-gapped build
that pre-stages the tarball (but not the extracted dir), this turns a working
build into a hard failure. (The extracted-binary short-circuit at the top of
`install_app` still works.)
- curl `elif` wget means when curl is present, **wget is never tried**
across all 3 attempts. The old script fell through to wget when curl produced
no file; a curl-specific failure (proxy/TLS/CA) that wget would survive now
fails outright.
- In `install_mvn`, since `APACHE_MIRROR` now defaults to
`archive.apache.org`, the "fall back to archive" block (124) is dead code for
the default config (the `%/` compare is equal → straight to `exit 2`). Minor:
`sleep 3` (82) also runs after the final failed attempt before giving up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
