This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new da8c1cec32 [CELEBORN-2362] Fix Flaky CI/CD
da8c1cec32 is described below
commit da8c1cec328110452458ee4174f03570ba8dddea
Author: afterincomparableyum
<[email protected]>
AuthorDate: Tue Jul 14 15:36:53 2026 +0800
[CELEBORN-2362] Fix Flaky CI/CD
### What changes were proposed in this pull request?
Address several independent root causes of CI/CD flakiness, spanning the
build tooling, JaCoCo instrumentation, the test mini-cluster, and the
process-wide shuffle client.
CI infrastructure
- build/mvn — resilient Maven bootstrap. Default to archive.apache.org
instead of the
closer.lua mirror redirector, and validate that each download is a real
gzip tarball before
extracting (closer.lua intermittently returns an HTML mirror-chooser page
with HTTP 200).
Retry up to 3 times, backing off only between attempts. A valid
pre-staged/previously-downloaded tarball is reused (keeps
offline/air-gapped builds working),
and curl falls through to wget when curl is absent or produced no valid
file (proxy/TLS/CA
failures wget would survive).
- pom.xml — exclude io/netty/** from JaCoCo. JaCoCo's agent rejects Netty
4.2's
already-enhanced JFR event classes, tearing down channels mid-write and
flaking the Flink
integration tests.
Process-wide ShuffleClient made per-application
The static single-slot ShuffleClient singleton couldn't represent the
multiple applications a
reused spark-it JVM runs with overlapping lifecycles: on an app switch it
evicted the
previous app's client without teardown, and a lock-free get() fast path
could hand a live app
the wrong app's client — surfacing as celebornShuffleId 0 corruption
(ArrayIndexOutOfBoundsException / CommitMetadata CRC mismatch).
- Replaced the single slot with a per-appUniqueId registry
(ConcurrentHashMap<String,
ShuffleClient>). Each application gets its own isolated client; only
fully-initialized
instances are ever published, closing the torn read.
- Added ShuffleClient.removeInstance(...), and SparkShuffleManager.stop()
now removes+shuts
down only its own app's client instead of calling the global reset()
(which would tear down
other live applications' clients in a multi-app JVM). This also
eliminates the
orphaned-client leak (RpcEnv, Netty factory, retry pool, reviveManager)
that previously
accumulated across app switches.
Spark-it lifecycle isolation
- SparkTestBase — stop any SparkSession/SparkContext still alive in the
JVM in afterAll
(triggering SparkShuffleManager.stop()) and reset the client, so a
straggler task from a
leaked context can't bind to a later suite's LifecycleManager through the
shared client.
Reduced worker count from 5 to 3 (the MiniCluster default): the suites
run serially in one
JVM, so each surplus worker multiplies the long-lived thread/CPU
footprint and, under CI
contention, starves RPC/fetch handlers past the 240s network timeout.
Cluster-setup test robustness
- MiniClusterFeature — retry worker startup with exponential backoff,
recreating the worker
each attempt. Record a worker in the returned set only once it has
registered, so a torn-down
failed worker (distinct identity — Worker has no equals/hashCode) doesn't
leak into the
returned set and get double-stopped. Tear failed workers down locally
(stop +
rpcEnv.shutdown) instead of exitImmediately(), which issues a blocking
WorkerLost master RPC
and pollutes the master's excluded list inside the retry loop.
- Random port selection — draw ports below the ephemeral floor
(Utils.selectRandomPort() /
MAX_SELECTABLE_PORT) to avoid a TOCTOU with OS-assigned/TIME_WAIT ports
that look free at
selection time but fail to bind. Consolidated the duplicated picker into
a shared
RandomPortSupport test trait.
- RatisMasterStatusSystemSuiteJ — re-point each server to a fresh storage
directory on every
start attempt; Ratis releases the directory lock asynchronously, so a
retry would otherwise
hit "directory is already locked".
- JVMQuake / JVMQuakeSuite — extract checkAndDump(...) so the
threshold/heap-dump logic can
be exercised deterministically without inducing real GC pressure.
- LifecycleManagerUnregisterShuffleSuite — shorten the expired-check
interval to 5s so the
shuffle unregister runs within the eventually() window with retry margin.
### Why are the changes needed?
CI/CD is always failing. With this, CI/CD rarely fails.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
CI/CD
Closes #3737 from afterincomparableyum/flaky-cicd.
Authored-by: afterincomparableyum
<[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
build/mvn | 108 +++++++++++++++------
.../shuffle/celeborn/SparkShuffleManager.java | 5 +-
.../org/apache/celeborn/client/ShuffleClient.java | 106 +++++++++++++-------
.../apache/celeborn/client/ShuffleClientImpl.java | 6 +-
.../celeborn/client/WithShuffleClientSuite.scala | 7 +-
.../org/apache/celeborn/common/util/Utils.scala | 6 +-
.../org/apache/celeborn/RandomPortSupport.scala | 55 +++++++++++
.../ha/RatisMasterStatusSystemSuiteJ.java | 16 ++-
.../deploy/master/MasterClusterFeature.scala | 28 +-----
pom.xml | 18 ++++
.../tests/flink/HybridShuffleWordCountTest.scala | 18 +++-
.../ChangePartitionManagerUpdateWorkersSuite.scala | 1 +
.../LifecycleManagerUnregisterShuffleSuite.scala | 7 ++
.../tests/spark/CelebornHashCheckDiskSuite.scala | 6 +-
.../tests/spark/CelebornIntegrityCheckSuite.scala | 22 ++++-
.../tests/spark/ShuffleFallbackSuite.scala | 2 +-
.../celeborn/tests/spark/SparkTestBase.scala | 24 ++++-
.../fetch/failure/ShuffleReaderGetHooks.scala | 12 +--
.../tests/spark/memory/MemorySparkTestBase.scala | 6 +-
.../spark/shuffle/celeborn/SparkUtilsSuite.scala | 6 +-
.../service/deploy/worker/monitor/JVMQuake.scala | 16 ++-
.../service/deploy/MiniClusterFeature.scala | 68 ++++++-------
.../deploy/worker/monitor/JVMQuakeSuite.scala | 38 +++-----
23 files changed, 404 insertions(+), 177 deletions(-)
diff --git a/build/mvn b/build/mvn
index cd6c0c796d..15b30aea1c 100755
--- a/build/mvn
+++ b/build/mvn
@@ -38,6 +38,7 @@ install_app() {
local remote_tarball="$1/$2$4"
local local_tarball="${_DIR}/$2"
local binary="${_DIR}/$3"
+ local max_attempts=3
# setup `curl` and `wget` silent options if we're running on Jenkins
local curl_opts="-L"
@@ -46,23 +47,56 @@ 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
+ # Reuse an already-valid tarball (e.g. one pre-staged for an offline
build, or downloaded on
+ # a previous attempt); only (re)download when it is missing or not a
valid gzip tarball.
+ if [ ! -f "${local_tarball}" ] || ! tar -tzf "${local_tarball}"
>/dev/null 2>&1; then
+ if ! command -v curl >/dev/null 2>&1 && ! command -v wget >/dev/null
2>&1; then
+ echo "ERROR: Cannot download $2: neither cURL nor wget is
installed." 1>&2
+ exit 2
+ fi
+
+ # Try curl first; remove its output if it fails so wget below gets a
turn.
+ rm -f "${local_tarball}"
+ 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}" || rm -f
"${local_tarball}"
+ fi
+
+ # Fall back to wget when curl is absent or produced no valid tarball
(e.g. a curl-specific
+ # proxy/TLS/CA failure that wget would survive).
+ if [ ! -f "${local_tarball}" ] || ! tar -tzf "${local_tarball}"
>/dev/null 2>&1; then
+ if command -v wget >/dev/null 2>&1; then
+ echo "exec: wget ${wget_opts} ${remote_tarball}" 1>&2
+ rm -f "${local_tarball}"
+ wget ${wget_opts} -O "${local_tarball}" "${remote_tarball}" || rm
-f "${local_tarball}"
+ fi
+ fi
+ fi
+
+ # Validate the download before trusting it. A flaky Apache mirror can
+ # return an HTML page (mirror chooser / error) with HTTP 200, which is
+ # not a gzip tarball; extracting it later would fail with a confusing
+ # exit code. `tar -tzf` lists the archive without extracting and
+ # exits non-zero on a non-tarball body.
+ if [ -f "${local_tarball}" ] && tar -tzf "${local_tarball}" >/dev/null
2>&1; then
+ if cd "${_DIR}" && tar -xzf "$2"; then
+ rm -rf "${local_tarball}"
+ return 0
+ fi
+ fi
+
+ # Drop the invalid download and retry, backing off only between attempts.
+ rm -f "${local_tarball}"
+ echo "WARN: Download of $2 from $1 was not a valid tarball" \
+ "(attempt ${attempt}/${max_attempts}); retrying..." 1>&2
+ attempt=$((attempt + 1))
+ [ "${attempt}" -le "${max_attempts}" ] && sleep 3
+ done
+
+ echo "WARN: Failed to download a valid $2 from $1 after ${max_attempts}
attempts." 1>&2
+ return 1
fi
}
@@ -77,25 +111,41 @@ install_mvn() {
# See simple version normalization:
http://stackoverflow.com/questions/16989598/bash-comparing-version-numbers
function version { echo "$@" | awk -F. '{ printf("%03d%03d%03d\n",
$1,$2,$3); }'; }
if [ $(version $MVN_DETECTED_VERSION) -ne $(version $MVN_VERSION) ]; then
- local
APACHE_MIRROR=${APACHE_MIRROR:-'https://www.apache.org/dyn/closer.lua'}
- local MIRROR_URL_QUERY="?action=download"
+ # Default to archive.apache.org: it serves the exact tarball
+ # deterministically, avoiding the closer.lua mirror redirector which
+ # intermittently routes to a mirror that returns an HTML page instead of
+ # the binary. Override with APACHE_MIRROR to use a closer mirror.
+ local APACHE_MIRROR=${APACHE_MIRROR:-'https://archive.apache.org/dist'}
local MVN_TARBALL="apache-maven-${MVN_VERSION}-bin.tar.gz"
local FILE_PATH="maven/maven-3/${MVN_VERSION}/binaries"
- if [ $(command -v curl) ]; then
- if ! curl -L --output /dev/null --silent --head --fail
"${APACHE_MIRROR}/${FILE_PATH}/${MVN_TARBALL}${MIRROR_URL_QUERY}" ; then
- # Fall back to archive.apache.org for older Maven
- echo "Falling back to archive.apache.org to download Maven"
- APACHE_MIRROR="https://archive.apache.org/dist";
- MIRROR_URL_QUERY=""
- fi
- fi
+ # closer.lua needs the ?action=download query to redirect to a mirror;
+ # archive.apache.org and most plain mirrors serve the file directly.
+ local MIRROR_URL_QUERY=""
+ case "${APACHE_MIRROR}" in
+ *closer.lua*) MIRROR_URL_QUERY="?action=download" ;;
+ esac
- install_app \
+ if ! install_app \
"${APACHE_MIRROR}/${FILE_PATH}" \
"${MVN_TARBALL}" \
"apache-maven-${MVN_VERSION}/bin/mvn" \
- "${MIRROR_URL_QUERY}"
+ "${MIRROR_URL_QUERY}"; then
+ # Last resort: fall back to archive.apache.org, which serves the exact
+ # tarball deterministically. Skip if it was already the chosen mirror.
+ local ARCHIVE_MIRROR="https://archive.apache.org/dist";
+ if [ "${APACHE_MIRROR%/}" != "${ARCHIVE_MIRROR}" ]; then
+ echo "WARN: falling back to ${ARCHIVE_MIRROR} to download Maven" 1>&2
+ install_app \
+ "${ARCHIVE_MIRROR}/${FILE_PATH}" \
+ "${MVN_TARBALL}" \
+ "apache-maven-${MVN_VERSION}/bin/mvn" \
+ "" || { echo "ERROR: Failed to download Maven; please install
manually." 1>&2; exit 2; }
+ else
+ echo "ERROR: Failed to download Maven; please install manually." 1>&2
+ exit 2
+ fi
+ fi
MVN_BIN="${_DIR}/apache-maven-${MVN_VERSION}/bin/mvn"
fi
diff --git
a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
index f528934b39..71aae81f31 100644
---
a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
+++
b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
@@ -275,8 +275,9 @@ public class SparkShuffleManager implements ShuffleManager {
public void stop() {
sortShuffleIds.clear();
if (shuffleClient != null) {
- shuffleClient.shutdown();
- ShuffleClient.reset();
+ // Remove and shut down only THIS application's client. Do not call
ShuffleClient.reset():
+ // in a multi-app JVM that would also tear down other live applications'
clients.
+ ShuffleClient.removeInstance(shuffleClient);
shuffleClient = null;
}
if (lifecycleManager != null) {
diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
index 8bc1a8911c..9fb15fdf7a 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
@@ -20,6 +20,7 @@ package org.apache.celeborn.client;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@@ -51,13 +52,18 @@ import
org.apache.celeborn.common.write.LocationPushFailedBatches;
import org.apache.celeborn.common.write.PushState;
/**
- * ShuffleClient may be a process singleton, the specific PartitionLocation
should be hidden in the
- * implementation
+ * ShuffleClient hides the specific PartitionLocation from callers. A process
holds one client per
+ * appUniqueId (see {@link #get}); most deployments run a single application
per JVM and therefore a
+ * single client, but a process that drives several applications (e.g. the
multi-app spark-it JVM)
+ * holds one isolated client per app.
*/
public abstract class ShuffleClient {
private static Logger logger = LoggerFactory.getLogger(ShuffleClient.class);
- private static volatile ShuffleClient _instance;
- private static volatile boolean initialized = false;
+ // One client per appUniqueId. Keying by appUniqueId keeps concurrent
applications isolated (each
+ // with its own LifecycleManager) and lets each be torn down independently
when its application
+ // stops, instead of a single static slot that would have to evict (and
orphan the resources of)
+ // the previous app's client on every switch.
+ private static final ConcurrentHashMap<String, ShuffleClient> clients = new
ConcurrentHashMap<>();
private static volatile Map<StorageInfo.Type, FileSystem> hadoopFs;
private static LongAdder totalReadCounter = new LongAdder();
private static LongAdder localShuffleReadCounter = new LongAdder();
@@ -68,9 +74,42 @@ public abstract class ShuffleClient {
// for testing
public static void reset() {
- _instance = null;
- initialized = false;
- hadoopFs = null;
+ List<ShuffleClient> toShutdown;
+ synchronized (ShuffleClient.class) {
+ toShutdown = new ArrayList<>(clients.values());
+ clients.clear();
+ hadoopFs = null;
+ }
+ // Shut down outside the lock: shutdown() tears down an RpcEnv and pools
and can block.
+ for (ShuffleClient client : toShutdown) {
+ try {
+ client.shutdown();
+ } catch (Throwable t) {
+ logger.warn("Failed to shutdown shuffle client during reset.", t);
+ }
+ }
+ }
+
+ /**
+ * Removes {@code client} from the registry and shuts it down exactly once,
reclaiming its RpcEnv,
+ * Netty data client factory, push-retry pool and reviveManager. An engine's
shuffle-manager calls
+ * this from its own stop() so a stopped application's client does not leak
until JVM exit, and
+ * (unlike {@link #reset()}) without touching other live applications'
clients. Keyed by the
+ * instance rather than appUniqueId because executor-side managers never
populate their
+ * appUniqueId field. No-op if the client was already removed (e.g. by a
concurrent {@link
+ * #reset()}), so the instance is never shut down twice.
+ */
+ public static void removeInstance(ShuffleClient client) {
+ if (client == null) {
+ return;
+ }
+ boolean removed;
+ synchronized (ShuffleClient.class) {
+ removed = clients.values().removeIf(existing -> existing == client);
+ }
+ if (removed) {
+ client.shutdown();
+ }
}
protected ShuffleClient() {}
@@ -103,38 +142,39 @@ public abstract class ShuffleClient {
UserIdentifier userIdentifier,
byte[] extension,
Optional<CryptoHandler> cryptoHandler) {
- if (null == _instance || !initialized) {
+ ShuffleClient client = clients.get(appUniqueId);
+ if (client == null) {
synchronized (ShuffleClient.class) {
- if (null == _instance) {
- // During the execution of Spark tasks, each task may be interrupted
due to speculative
- // tasks. If the Task is interrupted while obtaining the
ShuffleClient and the
- // ShuffleClient is building a singleton, it may cause the
LifecycleManagerEndpoint to not
- // be
- // assigned. An Executor will only construct a ShuffleClient
singleton once. At this time,
- // when communicating with LifecycleManager, it will cause a
NullPointerException.
- _instance = new ShuffleClientImpl(appUniqueId, conf, userIdentifier);
- _instance.setupLifecycleManagerRef(driverHost, port);
- _instance.setExtension(extension);
- _instance.setupCryptoHandler(cryptoHandler);
- initialized = true;
- } else if (!initialized) {
- _instance.shutdown();
- _instance = new ShuffleClientImpl(appUniqueId, conf, userIdentifier);
- _instance.setupLifecycleManagerRef(driverHost, port);
- _instance.setExtension(extension);
- _instance.setupCryptoHandler(cryptoHandler);
- initialized = true;
+ client = clients.get(appUniqueId);
+ if (client == null) {
+ // During Spark task execution a task may be interrupted (e.g. by
speculative execution)
+ // while this builds the client. Fully set the instance up before
publishing it into the
+ // registry, and tear a half-built instance down on failure, so a
later call rebuilds
+ // cleanly instead of returning a client with no LifecycleManagerRef
(which would NPE on
+ // first use). Because only fully-initialized instances are ever put
into the map, the
+ // lock-free clients.get() above can never observe a half-built or
wrong-app client.
+ ShuffleClientImpl instance = new ShuffleClientImpl(appUniqueId,
conf, userIdentifier);
+ try {
+ instance.setupLifecycleManagerRef(driverHost, port);
+ instance.setExtension(extension);
+ instance.setupCryptoHandler(cryptoHandler);
+ } catch (RuntimeException | Error e) {
+ instance.shutdown();
+ throw e;
+ }
+ clients.put(appUniqueId, instance);
+ client = instance;
}
}
}
- // Apply the crypto handler even when the singleton is already
initialized. This handles
- // the case where SparkEnv was transiently unavailable during the first
init call (causing
- // an empty handler to be stored), so that encryption is correctly applied
on retry.
- // setupCryptoHandler is a volatile write and safe to call without the
lock.
+ // Apply the crypto handler even when the client already exists. This
handles the case where
+ // SparkEnv was transiently unavailable during the first init call
(causing an empty handler to
+ // be stored), so that encryption is correctly applied on retry.
setupCryptoHandler is a
+ // volatile write and safe to call without the lock.
if (cryptoHandler != null && cryptoHandler.isPresent()) {
- _instance.setupCryptoHandler(cryptoHandler);
+ client.setupCryptoHandler(cryptoHandler);
}
- return _instance;
+ return client;
}
public static Map<StorageInfo.Type, FileSystem> getHadoopFs(CelebornConf
conf) {
diff --git
a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
index 46c1189ca4..210151241f 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -924,8 +924,10 @@ public class ShuffleClientImpl extends ShuffleClient {
StatusCode statusCode = entry.getValue()._1();
if (entry.getValue()._2() != null) {
PartitionLocation oldLoc = oldLocMap.get(partitionId);
- // Currently, revive only check if main location available, here
won't remove peer loc.
- pushExcludedWorkers.remove(oldLoc.hostAndPushPort());
+ if (oldLoc != null) {
+ // Currently, revive only check if main location available, here
won't remove peer loc.
+ pushExcludedWorkers.remove(oldLoc.hostAndPushPort());
+ }
}
if (StatusCode.SUCCESS == statusCode) {
diff --git
a/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
b/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
index 263a2a9a64..850740f5ff 100644
---
a/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
+++
b/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
@@ -34,7 +34,7 @@ trait WithShuffleClientSuite extends CelebornFunSuite {
protected val celebornConf: CelebornConf = new CelebornConf()
- protected val APP = "app-1"
+ protected var APP: String = _
protected val userIdentifier: UserIdentifier = UserIdentifier("mock", "mock")
private val numMappers = 8
private val mapId = 1
@@ -49,6 +49,11 @@ trait WithShuffleClientSuite extends CelebornFunSuite {
_shuffleId
}
+ override protected def beforeEach(): Unit = {
+ super.beforeEach()
+ APP = s"app-${java.util.UUID.randomUUID()}"
+ }
+
override protected def afterEach() {
if (lifecycleManager != null) {
lifecycleManager.stop()
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
index 5b2dd6a109..9a634c77d0 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
@@ -255,9 +255,13 @@ object Utils extends Logging {
* @return a randomly selected integer within the range [from, until)
*/
def selectRandomInt(from: Int, until: Int): Int = {
- ScalaRandom.nextInt(until - 1 - from) + from
+ ScalaRandom.nextInt(until - from) + from
}
+ val MAX_SELECTABLE_PORT = 32768
+
+ def selectRandomPort(): Int = selectRandomInt(1024, MAX_SELECTABLE_PORT)
+
def startServiceOnPort[T](
startPort: Int,
startService: Int => (T, Int),
diff --git a/common/src/test/scala/org/apache/celeborn/RandomPortSupport.scala
b/common/src/test/scala/org/apache/celeborn/RandomPortSupport.scala
new file mode 100644
index 0000000000..b760c250d6
--- /dev/null
+++ b/common/src/test/scala/org/apache/celeborn/RandomPortSupport.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn
+
+import java.io.IOException
+import java.net.{InetSocketAddress, Socket}
+
+import org.apache.celeborn.common.util.Utils
+
+/**
+ * Shared random-port picker for cluster-setup test traits. Draws ports below
the ephemeral floor
+ * (via [[Utils.selectRandomPort]]) and remembers the ports it has handed out
so repeated calls
+ * within a suite do not collide, retrying if a candidate is already used or
currently bound.
+ */
+trait RandomPortSupport {
+ val usedPorts = new java.util.HashSet[Integer]()
+
+ def portBounded(port: Int): Boolean = {
+ val socket = new Socket()
+ try {
+ socket.connect(new InetSocketAddress("localhost", port), 100)
+ true
+ } catch {
+ case _: IOException => false
+ } finally {
+ socket.close()
+ }
+ }
+
+ def selectRandomPort(): Int = synchronized {
+ val port = Utils.selectRandomPort()
+ val portUsed = usedPorts.contains(port) || portBounded(port)
+ usedPorts.add(port)
+ if (portUsed) {
+ selectRandomPort()
+ } else {
+ port
+ }
+ }
+}
diff --git
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
index 01a16a31ee..6038ee4e15 100644
---
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
+++
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
@@ -119,6 +119,19 @@ public class RatisMasterStatusSystemSuiteJ {
while (!serversStarted) {
try {
+ // Re-point each server to a fresh storage directory on retry. 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. Skip
this on the first
+ // attempt: callers already configure a fresh directory when building
conf1/2/3, so
+ // reconfiguring here would orphan that just-created (empty) directory.
+ if (retryCount > 0) {
+ configureServerConf(conf1, 1);
+ configureServerConf(conf2, 2);
+ configureServerConf(conf3, 3);
+ }
+
STATUSSYSTEM1 = new HAMasterMetaManager(mockRpcEnv, conf1);
STATUSSYSTEM2 = new HAMasterMetaManager(mockRpcEnv, conf2);
STATUSSYSTEM3 = new HAMasterMetaManager(mockRpcEnv, conf3);
@@ -131,7 +144,8 @@ public class RatisMasterStatusSystemSuiteJ {
String id2 = UUID.randomUUID().toString();
String id3 = UUID.randomUUID().toString();
- int ratisPort1 = Utils$.MODULE$.selectRandomInt(1024, 65535);
+ int ratisPort1 =
+ Utils$.MODULE$.selectRandomInt(1024,
Utils$.MODULE$.MAX_SELECTABLE_PORT() - 2);
int ratisPort2 = ratisPort1 + 1;
int ratisPort3 = ratisPort2 + 1;
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterClusterFeature.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterClusterFeature.scala
index 65995bfc3d..5f6bd60f1c 100644
---
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterClusterFeature.scala
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterClusterFeature.scala
@@ -18,14 +18,15 @@
package org.apache.celeborn.service.deploy.master
import java.io.IOException
-import java.net.{BindException, InetSocketAddress, Socket}
+import java.net.BindException
import java.util.concurrent.TimeUnit
+import org.apache.celeborn.RandomPortSupport
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
-trait MasterClusterFeature extends Logging {
+trait MasterClusterFeature extends Logging with RandomPortSupport {
var masterInfo: (Master, Thread) = _
val maxRetries = 3
@@ -37,29 +38,6 @@ trait MasterClusterFeature extends Logging {
}
}
- val usedPorts = new java.util.HashSet[Integer]()
- def portBounded(port: Int): Boolean = {
- val socket = new Socket()
- try {
- socket.connect(new InetSocketAddress("localhost", port), 100)
- true
- } catch {
- case _: IOException => false
- } finally {
- socket.close()
- }
- }
- def selectRandomPort(): Int = synchronized {
- val port = Utils.selectRandomInt(1024, 65535)
- val portUsed = usedPorts.contains(port) || portBounded(port)
- usedPorts.add(port)
- if (portUsed) {
- selectRandomPort()
- } else {
- port
- }
- }
-
def withRetryOnPortBindException(f: () => Unit): Unit = {
var retryCount = 0
var pass = false
diff --git a/pom.xml b/pom.xml
index 2e608aa1c0..f8738df863 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1315,6 +1315,24 @@
<goals>
<goal>prepare-agent</goal>
</goals>
+ <configuration>
+ <!--
+ Netty 4.2 fires JFR events (e.g.
io.netty.buffer.FreeBufferEvent,
+ FreeChunkEvent) by lazily loading event classes that extend
+ jdk.jfr.Event. Those classes are already bytecode-enhanced,
so
+ JaCoCo's on-the-fly agent rejects them with
+ "Cannot process instrumented class ..." and throws an
+ IllegalClassFormatException from defineClass. When this
happens on
+ a Netty I/O thread mid-write (e.g. during buffer
deallocation while
+ the worker streams shuffle data to the Flink read client),
the
+ channel is torn down and the client sees "Client is lost",
flaking
+ the Flink integration tests. Netty is a third-party
dependency we do
+ not measure coverage for, so exclude it from instrumentation.
+ -->
+ <excludes>
+ <exclude>io/netty/**</exclude>
+ </excludes>
+ </configuration>
</execution>
<execution>
<id>report</id>
diff --git
a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/HybridShuffleWordCountTest.scala
b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/HybridShuffleWordCountTest.scala
index 6794f23dfc..ef71860d8c 100644
---
a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/HybridShuffleWordCountTest.scala
+++
b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/HybridShuffleWordCountTest.scala
@@ -27,7 +27,9 @@ import org.apache.flink.runtime.jobgraph.JobType
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator
import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.Eventually._
import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.time.SpanSugar._
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
@@ -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)) {
+ workers.map(worker => {
+ worker.storageManager.workingDirWriters.values().asScala.map(writers
=> {
+ writers.forEach((fileName, fileWriter) => {
+ assert(new File(fileName).length() ==
fileWriter.getDiskFileInfo.getFileLength)
+ })
})
})
- })
+ }
}
}
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
index 3c0303f406..6e4b081230 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
@@ -42,6 +42,7 @@ class ChangePartitionManagerUpdateWorkersSuite extends
WithShuffleClientSuite
}
override def beforeEach(): Unit = {
+ super.beforeEach()
val testConf = Map(
s"${CelebornConf.CLIENT_PUSH_MAX_REVIVE_TIMES.key}" -> "3")
val (master, _) = setupMiniClusterWithRandomPorts(testConf, testConf,
workerNum = 1)
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerUnregisterShuffleSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerUnregisterShuffleSuite.scala
index d689ead7e0..c5082bd732 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerUnregisterShuffleSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerUnregisterShuffleSuite.scala
@@ -36,6 +36,13 @@ class LifecycleManagerUnregisterShuffleSuite extends
WithShuffleClientSuite
celebornConf
.set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
+ // The default expired-check interval is 60s. removeExpiredShuffle only
+ // unregisters a shuffle once `unregisterTime < now - checkInterval` and
runs
+ // on a fixed-rate timer at that interval, so with 60s the master side
cannot
+ // be cleared until the second tick (~120s) -- exactly the eventually()
window
+ // below, leaving no margin and no retry if an RPC briefly fails under
load.
+ // Use a short interval so the unregister runs promptly with ample retries.
+ .set(CelebornConf.SHUFFLE_EXPIRED_CHECK_INTERVAL.key, "5s")
override def beforeAll(): Unit = {
super.beforeAll()
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
index 4f22982bb9..b4c8e61529 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
@@ -24,7 +24,6 @@ import org.apache.spark.sql.SparkSession
import org.scalatest.concurrent.Eventually._
import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
-import org.apache.celeborn.client.ShuffleClient
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.protocol.ShuffleMode
import org.apache.celeborn.service.deploy.worker.Worker
@@ -43,10 +42,11 @@ class CelebornHashCheckDiskSuite extends SparkTestBase {
}
override def beforeEach(): Unit = {
- ShuffleClient.reset()
+ stopActiveSparkSessions()
}
override def afterEach(): Unit = {
+ stopActiveSparkSessions()
System.gc()
}
@@ -59,7 +59,7 @@ class CelebornHashCheckDiskSuite extends SparkTestBase {
val combineResult = combine(sparkSession)
val groupByResult = groupBy(sparkSession)
val repartitionResult = repartition(sparkSession)
- sparkSession.stop()
+ stopActiveSparkSessions()
val sparkSessionEnableCeleborn = SparkSession.builder()
.config(updateSparkConf(sparkConf, ShuffleMode.HASH))
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
index ef382d75c5..daea52ac1d 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
@@ -110,8 +110,28 @@ class CelebornIntegrityCheckSuite extends AnyFunSuite
// verify that the app fails
case e: Throwable => {
logger.error("Expected exception, logging the full exception", e)
+ // The contract under test is that corrupted data makes the app abort
+ // rather than silently return wrong results.
assert(e.getMessage.contains("Job aborted"))
- assert(e.getMessage.contains("CommitMetadata mismatch"))
+ // A single-bit flip is planted at a random position within a random
+ // data record, so the corruption is caught in one of two ways
+ // depending on where it lands:
+ // 1. Celeborn's integrity CRC, which is validated only at the end
of
+ // the partition (CelebornInputStream#validateIntegrity),
surfaces
+ // as "CommitMetadata mismatch".
+ // 2. The downstream deserializer (Kryo) choking on a structurally
+ // invalid byte (e.g. a bogus reference id or length) before the
+ // partition end is reached, so validateIntegrity() never runs
and
+ // a deserialization error (e.g. IndexOutOfBoundsException) is
the
+ // most-recent failure reported by Spark.
+ // Both are valid "fail instead of returning wrong data" outcomes and
+ // which one surfaces is non-deterministic, so accept either rather
than
+ // asserting on the integrity-specific message (which made this
flaky).
+ assert(
+ e.getMessage.contains("CommitMetadata mismatch") ||
+ e.getMessage.contains("IndexOutOfBounds") ||
+ e.getMessage.contains("com.esotericsoftware.kryo"),
+ s"App aborted for an unexpected reason: ${e.getMessage}")
}
} finally {
sparkSession.stop()
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
index 059e7d0337..a1c6468c2f 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
@@ -51,7 +51,7 @@ class ShuffleFallbackSuite extends AnyFunSuite
}
test(s"celeborn spark integration test - fallback") {
- setupMiniClusterWithRandomPorts(workerNum = 5)
+ setupMiniClusterWithRandomPorts(workerNum = 3)
val sparkConf = new SparkConf().setAppName("celeborn-demo")
.setMaster("local[2]")
.set(s"spark.${CelebornConf.SPARK_SHUFFLE_FORCE_FALLBACK_ENABLED.key}",
"true")
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
index c857bd67b0..58b94fef1f 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.scalatest.funsuite.AnyFunSuite
+import org.apache.celeborn.client.ShuffleClient
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.CelebornConf._
import org.apache.celeborn.common.internal.Logging
@@ -48,14 +49,35 @@ trait SparkTestBase extends AnyFunSuite
override def beforeAll(): Unit = {
logInfo("test initialized , setup Celeborn mini cluster")
- setupMiniClusterWithRandomPorts(workerNum = 5)
+ // Use 3 workers (the MiniClusterFeature default) rather than 5. The
spark-it suites run
+ // serially in a single JVM (scalatest forkMode=once), so every extra
worker multiplies the
+ // long-lived thread/CPU footprint across the whole module. Under CPU
contention on CI runners
+ // that surplus starves RPC handlers long enough to blow the 240s network
timeout, which then
+ // amplifies through read retries and Spark stage reattempts into
multi-minute hangs. 3 workers
+ // still exercises replication and slot spreading while cutting that
footprint.
+ setupMiniClusterWithRandomPorts(workerNum = 3)
}
override def afterAll(): Unit = {
logInfo("all test complete , stop Celeborn mini cluster")
+ // Tear down any SparkSession/SparkContext still alive in this JVM before
the next suite runs.
+ // Spark integration suites run sequentially (parallelExecution = false),
but a context that is
+ // not stopped keeps its LifecycleManager and the process-wide static
ShuffleClient alive. A
+ // straggler task from such a leaked context can then bind to a later
suite's LifecycleManager
+ // through the shared client and corrupt celebornShuffleId 0
(ArrayIndexOutOfBoundsException or
+ // CommitMetadata CRC mismatch). Stopping the context here triggers
SparkShuffleManager.stop(),
+ // which shuts the client down and stops the LifecycleManager.
+ stopActiveSparkSessions()
shutdownMiniCluster()
}
+ protected def stopActiveSparkSessions(): Unit = {
+
SparkSession.getActiveSession.orElse(SparkSession.getDefaultSession).foreach(_.stop())
+ SparkSession.clearActiveSession()
+ SparkSession.clearDefaultSession()
+ ShuffleClient.reset()
+ }
+
var workerDirs: Seq[String] = Seq.empty
def getOneWorker(): Worker = {
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/fetch/failure/ShuffleReaderGetHooks.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/fetch/failure/ShuffleReaderGetHooks.scala
index adac14242b..09d9efe623 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/fetch/failure/ShuffleReaderGetHooks.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/fetch/failure/ShuffleReaderGetHooks.scala
@@ -38,17 +38,15 @@ class ShuffleReaderGetHooks(
val lock = new Object
private def deleteDataFile(appUniqueId: String, celebornShuffleId: Int):
Unit = {
- val datafile =
+ val dataFiles =
workerDirs.map(dir => {
new
File(s"$dir/celeborn-worker/shuffle_data/$appUniqueId/$celebornShuffleId")
}).filter(_.exists())
- .flatMap(_.listFiles().iterator).headOption
- datafile match {
- case Some(file) => {
- file.delete()
- }
- case None => throw new RuntimeException("unexpected, there must be some
data file")
+ .flatMap(_.listFiles().iterator)
+ if (dataFiles.isEmpty) {
+ throw new RuntimeException("unexpected, there must be some data file")
}
+ dataFiles.foreach(_.delete())
}
override def exec(
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/memory/MemorySparkTestBase.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/memory/MemorySparkTestBase.scala
index 6d4b1e7119..6770d21e2f 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/memory/MemorySparkTestBase.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/memory/MemorySparkTestBase.scala
@@ -35,11 +35,15 @@ 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 = {
logInfo("all test complete , stop Celeborn mini cluster")
+ stopActiveSparkSessions()
shutdownMiniCluster()
}
diff --git
a/tests/spark-it/src/test/scala/org/apache/spark/shuffle/celeborn/SparkUtilsSuite.scala
b/tests/spark-it/src/test/scala/org/apache/spark/shuffle/celeborn/SparkUtilsSuite.scala
index b86244636b..24e604fd27 100644
---
a/tests/spark-it/src/test/scala/org/apache/spark/shuffle/celeborn/SparkUtilsSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/spark/shuffle/celeborn/SparkUtilsSuite.scala
@@ -70,12 +70,12 @@ class SparkUtilsSuite extends AnyFunSuite
val jobThread = new Thread {
override def run(): Unit = {
try {
- val value = Range(1, 10000).mkString(",")
+ val value = Range(1, 100).mkString(",")
sc.parallelize(1 to 10000, 2)
.map { i => (i, value) }
- .groupByKey(10)
+ .groupByKey(2)
.mapPartitions { iter =>
- Thread.sleep(3000)
+ Thread.sleep(500)
iter
}.collect()
} catch {
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
index 477b1ad2f7..822ba5faa0 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
@@ -91,9 +91,19 @@ class JVMQuake(conf: CelebornConf, uniqueId: String =
UUID.randomUUID().toString
val runTimeTicks = currentExitTime - lastExitTime - gcTimeTicks
// JVMStat time monitors are reported in ticks. Convert deltas to nanos
before comparing
// them against JVMQuake thresholds, which are stored as nanos.
- val gcTime = ticksToNanos(gcTimeTicks)
- val runTime = ticksToNanos(runTimeTicks)
+ checkAndDump(ticksToNanos(gcTimeTicks), ticksToNanos(runTimeTicks))
+ lastExitTime = currentExitTime
+ lastGCTime = currentGCTime
+ }
+ /**
+ * Updates the GC "deficit" bucket with the latest GC and execution time
deltas (in nanos) and
+ * heap dumps or kills the JVM once the configured thresholds are crossed.
Separated from the
+ * jvmstat counter reads in [[run]] so the threshold logic can be exercised
deterministically
+ * without inducing real GC pressure.
+ */
+ @VisibleForTesting
+ private[monitor] def checkAndDump(gcTime: Long, runTime: Long): Unit = {
bucket = Math.max(0, bucket + gcTime - (BigDecimal(runTime) *
BigDecimal(runtimeWeight)).toLong)
logDebug(s"Time: (gc time: ${Utils.nanoDurationToString(gcTime)},
execution time: ${Utils.nanoDurationToString(runTime)})")
logDebug(
@@ -110,8 +120,6 @@ class JVMQuake(conf: CelebornConf, uniqueId: String =
UUID.randomUUID().toString
System.exit(exitCode)
}
}
- lastExitTime = currentExitTime
- lastGCTime = currentGCTime
}
def shouldHeapDump: Boolean = {
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
index 2fab21e718..e4b26a1c51 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
@@ -18,7 +18,7 @@
package org.apache.celeborn.service.deploy
import java.io.IOException
-import java.net.{BindException, InetSocketAddress, Socket}
+import java.net.BindException
import java.nio.file.Files
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock
@@ -27,6 +27,7 @@ import scala.collection.mutable
import org.apache.commons.lang3.StringUtils
+import org.apache.celeborn.RandomPortSupport
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
@@ -34,7 +35,7 @@ import org.apache.celeborn.service.deploy.master.{Master,
MasterArguments}
import org.apache.celeborn.service.deploy.worker.{Worker, WorkerArguments}
import org.apache.celeborn.service.deploy.worker.memory.MemoryManager
-trait MiniClusterFeature extends Logging {
+trait MiniClusterFeature extends Logging with RandomPortSupport {
var masterInfo: (Master, Thread) = _
val workerInfos = new mutable.HashMap[Worker, Thread]()
@@ -51,29 +52,6 @@ trait MiniClusterFeature extends Logging {
}
}
- val usedPorts = new java.util.HashSet[Integer]()
- def portBounded(port: Int): Boolean = {
- val socket = new Socket()
- try {
- socket.connect(new InetSocketAddress("localhost", port), 100)
- true
- } catch {
- case _: IOException => false
- } finally {
- socket.close()
- }
- }
- def selectRandomPort(): Int = synchronized {
- val port = Utils.selectRandomInt(1024, 65535)
- val portUsed = usedPorts.contains(port) || portBounded(port)
- usedPorts.add(port)
- if (portUsed) {
- selectRandomPort()
- } else {
- port
- }
- }
-
def setupMiniClusterWithRandomPorts(
masterConf: Map[String, String] = Map(),
workerConf: Map[String, String] = Map(),
@@ -213,7 +191,7 @@ trait MiniClusterFeature extends Logging {
val workers = new Array[Worker](workerNum)
val flagUpdateLock = new ReentrantLock()
val threads = (1 to workerNum).map { i =>
- val worker = createWorker(workerConf)
+ var worker = createWorker(workerConf)
val workerThread = new RunnerWrap({
var workerStartRetry = 0
var workerStarted = false
@@ -225,10 +203,20 @@ 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()
- }
+ // Tear the failed worker down locally, mirroring the
InterruptedException branch
+ // above. Do NOT call exitImmediately() here: a worker that
failed to start usually
+ // never registered, and exitImmediately() issues a blocking
WorkerLost RPC to the
+ // master and adds this worker's host:ports to the master's
excluded list. In this
+ // retry loop that blocks up to the ask timeout per attempt and
leaks stale exclusions
+ // as each recreated worker re-registers under new random ports.
+
Utils.tryLogNonFatalError(worker.stop(CelebornExitKind.EXIT_IMMEDIATELY))
+ Utils.tryLogNonFatalError(worker.rpcEnv.shutdown())
workerStarted = false
workerStartRetry += 1
logError(s"cannot start worker $i, retrying: ", ex)
@@ -236,6 +224,14 @@ trait MiniClusterFeature extends Logging {
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)
}
}
})
@@ -250,14 +246,20 @@ trait MiniClusterFeature extends Logging {
try {
(0 until workerNum).foreach { i =>
{
- if (workers(i) == null) {
+ // Snapshot the slot under the same lock that guards its write, so
a worker reassigned
+ // on retry is visible here.
+ flagUpdateLock.lock()
+ val worker = workers(i)
+ flagUpdateLock.unlock()
+ if (worker == null) {
throw new IllegalStateException(s"worker $i hasn't been
initialized")
- } else if (!workerInfos.contains(workers(i))) {
- workerInfos.put(workers(i), threads(i))
}
- if (!workers(i).registered.get()) {
+ if (!worker.registered.get()) {
throw new IllegalStateException(s"worker $i hasn't been
registered")
}
+ if (!workerInfos.contains(worker)) {
+ workerInfos.put(worker, threads(i))
+ }
}
}
allWorkersStarted = true
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
index fa3b7f36be..94b86811c0 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
@@ -18,8 +18,7 @@
package org.apache.celeborn.service.deploy.worker.monitor
import java.io.File
-
-import scala.collection.mutable.ArrayBuffer
+import java.util.concurrent.TimeUnit
import org.junit.Assert.assertTrue
@@ -30,13 +29,6 @@ import org.apache.celeborn.common.util.JavaUtils
class JVMQuakeSuite extends CelebornFunSuite {
- private val allocation = new ArrayBuffer[Array[Byte]]()
-
- override def afterEach(): Unit = {
- allocation.clear()
- System.gc()
- }
-
test("Convert JVMStat timer ticks to nanoseconds") {
assert(JVMQuake.ticksToNanos(1L, 1000000000L) === 1L)
assert(JVMQuake.ticksToNanos(1000L, 1000L) === 1000000000L)
@@ -53,9 +45,14 @@ 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)
assertTrue(quake.heapDumped)
val heapDump = new File(s"${quake.getHeapDumpSavePath}/${quake.dumpFile}")
@@ -64,17 +61,10 @@ class JVMQuakeSuite extends CelebornFunSuite {
JavaUtils.deleteRecursively(new File(quake.getHeapDumpLinkPath))
}
- def allocateMemory(quake: JVMQuake): Unit = {
- val capacity = 1024 * 100
- while (allocation.size * capacity < Runtime.getRuntime.maxMemory / 4) {
- val bytes = new Array[Byte](capacity)
- allocation.append(bytes)
- }
- while (quake.shouldHeapDump) {
- for (index <- allocation.indices) {
- val bytes = new Array[Byte](capacity)
- allocation(index) = bytes
- }
- }
+ test("start() schedules monitoring and stop() tears it down without
dumping") {
+ val quake = new JVMQuake(new
CelebornConf().set(WORKER_JVM_QUAKE_ENABLED.key, "true"))
+ quake.start()
+ quake.stop()
+ assert(!quake.heapDumped)
}
}
