afterincomparableyum commented on code in PR #3737:
URL: https://github.com/apache/celeborn/pull/3737#discussion_r3570704818
##########
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:
I will leave a comment here explaining why we do this and why no change is
needed.
--
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]
