[incubator-celeborn] branch main updated: [CELEBORN-830] Add spark integration test to verify fallback with workers unavailable

[zhouky]

This is an automated email from the ASF dual-hosted git repository.

zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new 4918b8e21 [CELEBORN-830] Add spark integration test to verify fallback 
with workers unavailable
4918b8e21 is described below

commit 4918b8e21690667508c5d3478b116d15f3de4f84
Author: SteNicholas <programg...@163.com>
AuthorDate: Tue Aug 29 21:35:20 2023 +0800

    [CELEBORN-830] Add spark integration test to verify fallback with workers 
unavailable
    
    ### What changes were proposed in this pull request?
    
    Adds spark integration test to verify fallback with workers unavailable for 
#1814.
    
    ### Why are the changes needed?
    
    None.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    - `ShuffleFallbackSuite` adds `eleborn spark integration test - fallback 
with workers unavailable` to verify whether to fallback with workers 
unavailable.
    
    Closes #1856 from SteNicholas/CELEBORN-830.
    
    Authored-by: SteNicholas <programg...@163.com>
    Signed-off-by: zky.zhoukeyong <zky.zhoukey...@alibaba-inc.com>
---
 .../tests/spark/ShuffleFallbackSuite.scala         | 28 +++++++++++++++++++---
 1 file changed, 25 insertions(+), 3 deletions(-)

diff --git 
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
 
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
index 70c0655b9..cbf973b92 100644
--- 
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
+++ 
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
@@ -17,8 +17,6 @@
 
 package org.apache.celeborn.tests.spark
 
-import scala.util.Random
-
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.SparkSession
 import org.scalatest.BeforeAndAfterEach
@@ -26,12 +24,19 @@ import org.scalatest.funsuite.AnyFunSuite
 
 import org.apache.celeborn.client.ShuffleClient
 import org.apache.celeborn.common.CelebornConf
-import org.apache.celeborn.common.protocol.CompressionCodec
 
 class ShuffleFallbackSuite extends AnyFunSuite
   with SparkTestBase
   with BeforeAndAfterEach {
 
+  override def beforeAll(): Unit = {
+    logInfo("test initialized")
+  }
+
+  override def afterAll(): Unit = {
+    logInfo("all test complete")
+  }
+
   override def beforeEach(): Unit = {
     ShuffleClient.reset()
   }
@@ -46,6 +51,7 @@ class ShuffleFallbackSuite extends AnyFunSuite
   }
 
   test(s"celeborn spark integration test - fallback") {
+    setUpMiniCluster(workerNum = 5)
     val sparkConf = new SparkConf().setAppName("celeborn-demo")
       .setMaster("local[2]")
       .set(s"spark.${CelebornConf.SPARK_SHUFFLE_FORCE_FALLBACK_ENABLED.key}", 
"true")
@@ -57,5 +63,21 @@ class ShuffleFallbackSuite extends AnyFunSuite
       .repartition(100)
     df.collect()
     sparkSession.stop()
+    shutdownMiniCluster()
+  }
+
+  test(s"celeborn spark integration test - fallback with workers unavailable") 
{
+    setUpMiniCluster(workerNum = 0)
+    val sparkConf = new SparkConf().setAppName("celeborn-demo")
+      .setMaster("local[2]")
+
+    enableCeleborn(sparkConf)
+
+    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
+    val df = sparkSession.sparkContext.parallelize(1 to 120000, 8)
+      .repartition(100)
+    df.collect()
+    sparkSession.stop()
+    shutdownMiniCluster()
   }
 }