Copilot commented on code in PR #3737:
URL: https://github.com/apache/celeborn/pull/3737#discussion_r3421382504
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -102,12 +104,21 @@ 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);
+ _appUniqueId = appUniqueId;
+ _instance = newInstance;
initialized = true;
}
Review Comment:
In the appUniqueId mismatch branch, `_appUniqueId` is written before
`_instance`. Because both are `volatile` and reads happen outside the
synchronized block, another thread can observe the new `_appUniqueId` and
return the *old* `_instance` (outer `if` becomes false) until `_instance` is
assigned. Also, the old instance is replaced without calling `shutdown()`,
which can leak threads/connections.
Publish `_instance` before `_appUniqueId` and shut down the old instance
when swapping to a new app id.
##########
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) ]; then
+ echo "exec: curl ${curl_opts} ${remote_tarball}" 1>&2
+ curl ${curl_opts} "${remote_tarball}" > "${local_tarball}"
+ elif [ $(command -v wget) ]; then
+ echo "exec: wget ${wget_opts} ${remote_tarball}" 1>&2
+ wget ${wget_opts} -O "${local_tarball}" "${remote_tarball}"
Review Comment:
`if [ $(command -v curl) ]; then` is unsafe: when `curl` is not installed,
the command substitution is empty and the test becomes `[ ]`, which triggers a
shell error. Use `command -v ... >/dev/null 2>&1` for a reliable existence
check (same for `wget`).
##########
worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala:
##########
@@ -225,17 +236,24 @@ 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
}
+ TimeUnit.SECONDS.sleep(Math.pow(2, workerStartRetry).toLong)
+ worker = createWorker(workerConf)
Review Comment:
The retry backoff uses `TimeUnit.SECONDS.sleep(...)` inside the `catch {
case ex: Exception => ... }` body. If the thread is interrupted during this
sleep, `InterruptedException` will escape the catch block with the interrupt
flag cleared, which makes teardown/interruption handling unreliable.
Re-interrupt the thread before rethrowing.
--
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]
