SteNicholas commented on code in PR #3737:
URL: https://github.com/apache/celeborn/pull/3737#discussion_r3428201756
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -102,12 +104,24 @@ 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 (!appUniqueId.equals(_appUniqueId)) {
Review Comment:
Bug: The `!initialized` branch (line 110) calls `_instance.shutdown()`
before replacing the instance, but this new `appUniqueId`-mismatch branch
silently drops the old instance without shutdown. This leaks the old
`ShuffleClientImpl`'s `RpcEnv`, `TransportClientFactory`, `pushDataRetryPool`
thread pool, and `ReviveManager` resources.
In YARN container-reuse or Spark dynamic-allocation where an executor is
reassigned to a new app, these resources accumulate until the JVM exits.
Suggest adding `_instance.shutdown();` before creating `newInstance`,
matching the pattern in the `!initialized` branch.
##########
build/mvn:
##########
@@ -77,19 +100,20 @@ 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" ;;
Review Comment:
Minor: When `APACHE_MIRROR` is explicitly set to a `closer.lua` URL, the new
code appends `?action=download` but retries 3 times against the same
potentially-broken mirror without falling back to `archive.apache.org`. The old
code had an explicit HEAD-request probe that detected HTML responses and fell
back. The retry-with-tarball-validation partially mitigates this, but the
explicit fallback to a known-good URL is lost.
Consider adding a fallback: after exhausting retries, try
`archive.apache.org` as a last resort before exiting.
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -102,12 +104,24 @@ 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 (!appUniqueId.equals(_appUniqueId)) {
+ ShuffleClientImpl newInstance = new ShuffleClientImpl(appUniqueId,
conf, userIdentifier);
+ newInstance.setupLifecycleManagerRef(driverHost, port);
+ newInstance.setExtension(extension);
+ // Publish _instance before _appUniqueId. The outer guard reads both
volatiles without
+ // holding the lock, so writing _appUniqueId first would let another
thread observe the
+ // new id while _instance is still stale and return the old instance.
+ _instance = newInstance;
+ _appUniqueId = appUniqueId;
initialized = true;
}
}
Review Comment:
Cleanup: The three inner branches (`_instance == null`, `!initialized`,
`!appUniqueId.equals(_appUniqueId)`) now share nearly identical 4-line setup
sequences (create `ShuffleClientImpl`, `setupLifecycleManagerRef`,
`setExtension`, set `_appUniqueId`, set `initialized`). The only meaningful
difference is whether to call `shutdown()` first.
Consider extracting the common setup into a helper to avoid the copy-paste
hazard — a future change added to one branch but missed in another would
silently misconfigure the client only under the specific race that triggers
that branch.
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala:
##########
@@ -43,13 +43,21 @@ class CelebornHashCheckDiskSuite extends SparkTestBase {
}
override def beforeEach(): Unit = {
- ShuffleClient.reset()
+ resetSparkState()
}
override def afterEach(): Unit = {
+ resetSparkState()
System.gc()
}
+ private def resetSparkState(): Unit = {
+
SparkSession.getActiveSession.orElse(SparkSession.getDefaultSession).foreach(_.stop())
Review Comment:
Cleanup (duplication): This `resetSparkState()` method is a line-for-line
duplicate of `stopActiveSparkSessions()` added to the parent trait
`SparkTestBase` in this same PR. Since `CelebornHashCheckDiskSuite` extends
`SparkTestBase`, it could call the inherited method instead.
If the cleanup logic ever changes, two copies would need to be updated in
lockstep.
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -90,7 +92,7 @@ public static ShuffleClient get(
CelebornConf conf,
UserIdentifier userIdentifier,
byte[] extension) {
- if (null == _instance || !initialized) {
+ if (null == _instance || !initialized ||
!appUniqueId.equals(_appUniqueId)) {
Review Comment:
Bug (defensive): `appUniqueId.equals(_appUniqueId)` will throw
`NullPointerException` if the `appUniqueId` **parameter** is null. The old
guard (`null == _instance || !initialized`) never dereferenced `appUniqueId`.
While current Spark callers always pass non-null, MapReduce and Tez callers
read `appUniqueId` from Hadoop `conf.get(key)` which returns null when the key
is absent.
Consider using `Objects.equals(appUniqueId, _appUniqueId)` or guarding with
a null check.
--
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]
