Re: [PR] [CELEBORN-2451] Make DRA safe under AUTO fallback by tying supportsReliableStorage to fallback policy [celeborn]

["SteNicholas (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org