SteNicholas commented on code in PR #3834:
URL: https://github.com/apache/celeborn/pull/3834#discussion_r4258712688
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java:
##########
@@ -110,28 +127,47 @@ private Optional<CryptoHandler> getCryptoHandler() {
return cryptoHandler;
}
+ private static boolean isDecommissionShuffleBlocksEnabled(SparkConf conf) {
+ return DECOMMISSION_ENABLED != null
+ && STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED != null
+ && (Boolean) conf.get(DECOMMISSION_ENABLED)
+ && (Boolean) conf.get(STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED);
+ }
+
+ static boolean isUnsafeDraFallback(SparkConf conf) {
+ return conf.getBoolean("spark.dynamicAllocation.enabled", false)
+ && !conf.getBoolean("spark.shuffle.service.enabled", false)
+ && !(Boolean)
conf.get(package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED())
+ && !isDecommissionShuffleBlocksEnabled(conf);
+ }
+
public SparkShuffleManager(SparkConf conf, boolean isDriver) {
if (conf.getBoolean(SQLConf.LOCAL_SHUFFLE_READER_ENABLED().key(), true)) {
logger.warn(
"Detected {} (default is true) is enabled, it's highly recommended
to disable it when "
+ "use Celeborn as Remote Shuffle Service to avoid performance
degradation.",
SQLConf.LOCAL_SHUFFLE_READER_ENABLED().key());
}
- if ((Boolean)
conf.get(package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED())) {
- String key =
package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().key();
- Boolean defaultValue =
- (Boolean)
package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().defaultValue().get();
+ this.celebornConf = SparkUtils.fromSparkConf(conf);
+ boolean shuffleTrackingEnabled =
+ (Boolean)
conf.get(package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED());
+ boolean neverFallback =
FallbackPolicy.NEVER.equals(celebornConf.sparkShuffleFallbackPolicy());
+ if (!neverFallback && isUnsafeDraFallback(conf)) {
logger.warn(
- "Detected {} (default is {}) is enabled, "
- + "it's highly recommended to disable it when use Celeborn as
Remote Shuffle Service "
- + "to avoid performance degradation.",
- key,
- defaultValue);
+ "DRA is enabled without the external shuffle service and fallback
policy is not NEVER, "
+ + "but {} is disabled and shuffle decommissioning is not
enabled. Enable tracking so fallback "
+ + "shuffle output is not lost when idle executors are
reclaimed.",
+ package$.MODULE$.DYN_ALLOCATION_SHUFFLE_TRACKING_ENABLED().key());
+ } else if (neverFallback && shuffleTrackingEnabled) {
+ // Under NEVER all shuffle stays on Celeborn, so tracking only delays
releasing executors.
+ logger.warn(
+ "{} is enabled with fallback policy NEVER; it is unnecessary and
delays releasing idle "
+ + "executors, so it is recommended to disable it.",
Review Comment:
[P2] Qualify the recommendation to disable shuffle tracking
`NEVER` alone does not make shuffle tracking unnecessary for Spark's
dynamic-allocation validation. For example, with DRA enabled on Spark 3.4 and
neither an external shuffle service nor shuffle decommissioning configured,
following this warning makes the application fail at startup. On Spark 3.5+,
disabling tracking also requires an IO plugin that reports reliable storage,
such as `CelebornShuffleDataIO`, unless one of those other mechanisms is
configured. Please guard this recommendation with the required capability or
state these prerequisites explicitly, consistent with the README and deployment
guide.
##########
client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleManagerSuite.scala:
##########
@@ -92,6 +92,70 @@ class SparkShuffleManagerSuite extends Logging {
sc.stop()
}
+ @junit.Test
+ def testReliableStorageFollowsFallbackPolicy(): Unit = {
+ // supportsReliableStorage must be true only for NEVER; AUTO/ALWAYS may
fall back to local-disk
+ // shuffle, so reporting reliable storage would let DRA reclaim executors
holding fallback data.
+ def reliable(policy: String): Boolean = {
+ val conf = new SparkConf(false)
+ .set(s"spark.${CelebornConf.SPARK_SHUFFLE_FALLBACK_POLICY.key}",
policy)
+ new CelebornShuffleDataIO(conf).driver()
+
.asInstanceOf[CelebornShuffleDriverComponents].supportsReliableStorage()
+ }
+ Assert.assertTrue(reliable("NEVER"))
+ Assert.assertFalse(reliable("AUTO"))
+ Assert.assertFalse(reliable("ALWAYS"))
+ Assert.assertFalse(
+ new CelebornShuffleDataIO(new SparkConf(false)).driver()
+
.asInstanceOf[CelebornShuffleDriverComponents].supportsReliableStorage())
+ }
+
+ @junit.Test
+ def testUnsafeDraFallback(): Unit = {
+ for {
+ dra <- Seq(false, true)
+ shuffleService <- Seq(false, true)
+ tracking <- Seq(false, true)
+ decommission <- Seq(false, true)
+ shuffleBlocks <- Seq(false, true)
+ } {
+ val conf = new SparkConf(false)
+ .set("spark.dynamicAllocation.enabled", dra.toString)
+ .set("spark.shuffle.service.enabled", shuffleService.toString)
+ .set("spark.dynamicAllocation.shuffleTracking.enabled",
tracking.toString)
+ .set("spark.decommission.enabled", decommission.toString)
+ .set("spark.storage.decommission.shuffleBlocks.enabled",
shuffleBlocks.toString)
+ val decommissionSupported = SparkVersionUtil.isGreaterThan(3, 0)
+ val expected = dra && !shuffleService && !tracking &&
+ !(decommissionSupported && decommission && shuffleBlocks)
+ Assert.assertEquals(conf.toDebugString, expected,
SparkShuffleManager.isUnsafeDraFallback(conf))
Review Comment:
[P2] Apply the formatter to unblock CI
This assertion fails the project's Spotless check: the formatter expects the
call to be split across multiple lines. The current [style
check](https://github.com/apache/celeborn/actions/runs/34889029285/job/115149935707)
and [Spark 3.5 Maven
job](https://github.com/apache/celeborn/actions/runs/34889029379/job/115149937317)
both stop on this violation before the Spark tests run. Please apply the
project formatter and rerun the formatting 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]
