SteNicholas commented on code in PR #3737:
URL: https://github.com/apache/celeborn/pull/3737#discussion_r3449634909
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -102,12 +105,30 @@ 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)) {
Review Comment:
**[blocker] `get()` can return another app's client — data race on the
trailing `return _instance`.**
Ordering the volatile writes (`_instance` before `_appUniqueId`) correctly
stops the *outer* lock-free guard from handing back a stale instance. But the
method still ends with `return _instance` **outside** the `synchronized` block.
Between this thread releasing the lock and reaching that return, another thread
with a different `appUniqueId` can re-enter this branch and overwrite
`_instance`:
```
Thread A (app "A") Thread B (app "B")
_instance = newInstanceA
_appUniqueId = "A"
<exit synchronized>
<enter synchronized>
_instance = newInstanceB
return _instance // hands B's client to A
```
A then resolves `celebornShuffleId` against B's `LifecycleManager` → the
`ArrayIndexOutOfBoundsException` / CommitMetadata CRC-mismatch / cross-app read
this PR is trying to eliminate. Fix: return the instance you just built —
capture it in a local and `return` that (or return from inside the lock)
instead of re-reading the shared field.
Two lower-severity points on the same branch:
- **Singleton ping-pong:** with the outer guard now `||
!Objects.equals(appUniqueId, _appUniqueId)`, every `get()` carrying a different
id re-enters and swaps the static `_instance`. In a JVM with two
concurrently-live apps (this test harness; also Flink session-cluster
TaskManagers in prod), A's and B's tasks flip the shared singleton back and
forth, so a holder can observe a client with the wrong/empty registration
state. Single-app executors never reach this branch.
- **Orphan leak (intentional — flagging for sign-off):** the replaced
instance is deliberately not `shutdown()` (per the comment, to avoid
`RejectedExecutionException`), leaking one `RpcEnv` + thread pools per distinct
`appUniqueId` for the JVM's life. Bounded and fine for single-app JVMs, but
worth a reviewer explicitly ratifying.
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -126,7 +126,7 @@ class ApplicationHeartbeater(
}
}
},
- 0,
+ appHeartbeatIntervalMs,
Review Comment:
This is a **production behavior change inside a "Fix Flaky CI/CD" PR**, with
no comment explaining it. The initial delay went `0 -> appHeartbeatIntervalMs`,
so the first app heartbeat — which also drives `ReviseLostShuffles` and app
liveness on the master — now fires a full interval later for *every* real
deployment, just to settle a test's timing. Prefer fixing this at the test
layer (widen the test's wait/`eventually` window); if the delay really is
needed in prod, please add a comment justifying it.
##########
master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java:
##########
@@ -119,6 +119,15 @@ public static void resetRaftServer(
while (!serversStarted) {
try {
+ // Re-point each server to a fresh storage directory on every attempt.
Ratis releases the
+ // storage directory lock asynchronously on close(), so a failed
attempt (e.g. a random
+ // ratis port collision) can leave the previous directory locked.
Reusing the same
+ // directory on retry then fails with "directory is already locked";
allocating a clean
+ // directory each time avoids contending for a lock that has not been
released yet.
+ configureServerConf(conf1, 1);
Review Comment:
`configureServerConf` is already invoked by the callers when they build
`conf1/2/3` (e.g. `init()`), so on the **first** loop iteration it now runs
twice per server on the same conf — the storage dir the caller just created is
immediately overwritten and orphaned (an empty `celeborn-ratis*` temp dir,
never cleaned), once per server on every `resetRaftServer`. The stated intent
("fresh dir only on retry") would be served by skipping this on the first
attempt, e.g. guard on a retry counter.
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/memory/MemorySparkTestBase.scala:
##########
@@ -35,7 +35,10 @@ trait MemorySparkTestBase extends AnyFunSuite
override def beforeAll(): Unit = {
logInfo("test initialized , setup Celeborn mini cluster")
val workerConfs =
Map("celeborn.worker.directMemoryRatioForMemoryFileStorage" -> "0.2")
- setupMiniClusterWithRandomPorts(workerConf = workerConfs, workerNum = 5)
+ // 3 workers (the MiniClusterFeature default) instead of 5: these
memory-storage suites run in
+ // the same shared, serial spark-it JVM, so trimming the per-suite worker
footprint reduces the
+ // CPU contention that otherwise starves a worker's fetch handler past the
240s network timeout.
+ setupMiniClusterWithRandomPorts(workerConf = workerConfs, workerNum = 3)
}
override def afterAll(): Unit = {
Review Comment:
This pre-existing `afterAll` calls only `shutdownMiniCluster()` — no
`super.afterAll()` and no `stopActiveSparkSessions()`. The point of the new
`SparkTestBase.afterAll` cleanup in this PR is to stop a leaked
`SparkSession`/`SparkContext` (and run `ShuffleClient.reset()`) before the next
suite runs in the shared serial JVM. Since this override isn't updated,
memory-storage suites skip that cleanup and can still leave a stale
`LifecycleManager` behind — exactly the flakiness the PR targets. Please call
`super.afterAll()` (or `stopActiveSparkSessions()`) here.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala:
##########
@@ -225,17 +236,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 every retry (`var worker =
createWorker(workerConf)`) instead of re-`initialize()`-ing the same instance
has two side effects:
- **Leaked JVM shutdown hooks:** `Worker`'s constructor unconditionally
registers a JVM shutdown hook and nothing ever removes it, so each failed
attempt permanently leaks a hook bound to a now-dead worker. In the shared
serial spark-it/flink-it JVM these accumulate; at JVM exit they all fire
`askSync(WorkerLost)` against a dead master, each blocking on the RPC ask
timeout (slow/noisy shutdown), and they keep the dead `Worker` objects
reachable for the whole run.
- **A dead worker can be published:** with `var worker` reassigned, the
registration-waiting thread may `workerInfos.put` the first (torn-down) attempt
before the replacement is assigned, leaving a dead worker in the map that
`getOneWorker()` / `shutdownMiniCluster()` can later pick — NPE/teardown
flakiness in the very retry path this PR targets.
Consider retrying `initialize()` on a single instance (as before), or
removing the shutdown hook on teardown.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala:
##########
@@ -53,28 +45,19 @@ class JVMQuakeSuite extends CelebornFunSuite {
.set(WORKER_JVM_QUAKE_RUNTIME_WEIGHT.key, "1")
.set(WORKER_JVM_QUAKE_DUMP_THRESHOLD.key, "1s")
.set(WORKER_JVM_QUAKE_KILL_THRESHOLD.key, "2s"))
- quake.start()
- allocateMemory(quake)
- quake.stop()
+
+ // Drive the GC "deficit" bucket deterministically rather than inducing
real GC pressure:
+ // feed a GC-time delta above the 1s dump threshold with no offsetting
execution time, so the
+ // heap dump is triggered exactly once. The previous version spun until
real GC happened to
+ // trip the threshold, which could (and did) hang indefinitely when the
runner had enough
+ // headroom that GC pauses never dominated runtime.
+ assert(!quake.heapDumped)
+ quake.checkAndDump(TimeUnit.SECONDS.toNanos(2), 0L)
Review Comment:
Calling `checkAndDump(...)` directly fixes the hang, but drops all coverage
of `start()/stop()` scheduling and the production `run()` path (jvmstat counter
read + ticks->nanos conversion + the scheduler actually invoking `run()`). A
regression there — bad delta, bad conversion, scheduler never firing — would no
longer be caught by CI, and a worker stuck in GC thrash would silently never
self-terminate. Consider keeping a lightweight test that exercises `run()` (or
at least the scheduling wiring) alongside this deterministic `checkAndDump`
assertion.
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala:
##########
@@ -225,17 +236,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:
`TimeUnit.SECONDS.sleep(Math.pow(2, workerStartRetry))` on the bootstrap
thread adds up to ~2+4+8 = 14s of pure sleep per flaky worker to suite setup
(maxRetries=4). For a transient bind error a much smaller backoff suffices —
consider capping it, e.g. `min(2^n, 2)` seconds.
##########
master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterClusterFeature.scala:
##########
@@ -50,7 +50,7 @@ trait MasterClusterFeature extends Logging {
}
}
def selectRandomPort(): Int = synchronized {
- val port = Utils.selectRandomInt(1024, 65535)
+ val port = Utils.selectRandomInt(1024, 32768)
Review Comment:
The ephemeral-port-floor fix lives only in `MiniClusterFeature` (the named
`maxSelectablePort` constant + the rationale comment). This copy carries a bare
`32768` literal, and `RatisMasterStatusSystemSuiteJ` carries yet another
(`32766`). Three near-identical `selectRandomPort` blocks now diverge — the
next person who tweaks the floor in one place will silently leave the others
wrong and the BindException flakiness reappears in the master suites. Extract
one shared helper/constant and call it from all three.
##########
tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/HybridShuffleWordCountTest.scala:
##########
@@ -186,12 +188,18 @@ class HybridShuffleWordCountTest extends AnyFunSuite with
Logging with MiniClust
}
private def checkFlushingFileLength(): Unit = {
- workers.map(worker => {
- worker.storageManager.workingDirWriters.values().asScala.map(writers => {
- writers.forEach((fileName, fileWriter) => {
- assert(new File(fileName).length() ==
fileWriter.getDiskFileInfo.getFileLength)
+ // getDiskFileInfo.getFileLength is the logical byte count accounted as
data is written, while
+ // the physical file is grown asynchronously by the LocalFlusher. Right
after the job finishes
+ // the flusher may not have drained the last buffers yet, so the on-disk
length can lag (briefly
+ // even 0). Wait for the flush to catch up before asserting equality
instead of reading mid-flush.
+ eventually(timeout(30.seconds), interval(500.milliseconds)) {
Review Comment:
Minor: each 500ms poll re-walks every worker's full `workingDirWriters` map
and issues a fresh `File.length()` syscall per file for up to 30s, so a single
lagging file causes every already-matching file to be re-stat'd repeatedly.
Could poll only the not-yet-matching writers, or drain/await the flush once
before a single assertion pass.
--
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]
