delete most of tests

Author: CodingCat

tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureDiskCleanSuite.scala

@@ -22,9 +22,7 @@ import org.apache.celeborn.tests.spark.fetch.failure.{FetchFailureDiskCleanBase,
 
 class CelebornFetchFailureDiskCleanSuite extends FetchFailureDiskCleanBase {
 
-  // 1. for single level 1-1 lineage, the old disk space is cleaned before the spark application
-  // finish
-  test("celeborn spark integration test - (1-1 dep with, single level lineage) the failed shuffle file is cleaned up correctly") {
+  test("celeborn spark integration test - the failed shuffle file is cleaned up correctly") {
     if (Spark3OrNewer) {
       val sparkSession = createSparkSession(enableFailedShuffleCleaner = true)
       val celebornConf = SparkUtils.fromSparkConf(sparkSession.sparkContext.getConf)
@@ -47,90 +45,4 @@ class CelebornFetchFailureDiskCleanSuite extends FetchFailureDiskCleanBase {
       sparkSession.stop()
     }
   }
-
-  // 2. for multiple level 1-1 lineage, the old disk space is cleaned one by one
-  test("celeborn spark integration test - (1-1 dep with, multi-level lineage) the failed shuffle file is cleaned up correctly") {
-    if (Spark3OrNewer) {
-      val sparkSession = createSparkSession(enableFailedShuffleCleaner = true)
-      val celebornConf = SparkUtils.fromSparkConf(sparkSession.sparkContext.getConf)
-      val hook =
-        new FileDeletionShuffleReaderGetHook(
-          celebornConf,
-          workerDirs,
-          shuffleIdToBeDeleted = Seq(0, 1),
-          triggerStageId = Some(2))
-      TestCelebornShuffleManager.registerReaderGetHook(hook)
-      val checkingThread = triggerStorageCheckThread(
-        Seq(0, 1),
-        Seq(2, 3, 4),
-        sparkSession,
-        forStableStatusChecking = false)
-      val tuples = sparkSession.sparkContext.parallelize(1 to 10000, 2)
-        .map { i => (i, i) }.groupByKey(16).map {
-          case (k, elements) => (k, elements.map(i => i))
-        }.groupByKey(4).groupByKey(2).collect()
-      checkStorageValidation(checkingThread)
-      // verify result
-      assert(hook.executed.get())
-      assert(tuples.length == 10000)
-      for (elem <- tuples) {
-        elem._2.flatten.flatten.foreach(s => s.equals(elem._1))
-      }
-      sparkSession.stop()
-    }
-  }
-
-  // 3. for single level M-1 lineage, the single failed disk space is cleaned
-  test(
-    "celeborn spark integration test - (M-1 dep with single level lineage) the single failed shuffle file is cleaned up correctly") {
-    if (Spark3OrNewer) {
-      val sparkSession = createSparkSession(enableFailedShuffleCleaner = true)
-      val celebornConf = SparkUtils.fromSparkConf(sparkSession.sparkContext.getConf)
-      val hook = new FileDeletionShuffleReaderGetHook(
-        celebornConf,
-        workerDirs,
-        shuffleIdToBeDeleted = Seq(0))
-      TestCelebornShuffleManager.registerReaderGetHook(hook)
-      val checkingThread =
-        triggerStorageCheckThread(Seq(0), Seq(1, 2), sparkSession, forStableStatusChecking = false)
-      import sparkSession.implicits._
-      val df1 = Seq((1, "a"), (2, "b")).toDF("id", "data").groupBy("id").count()
-      val df2 = Seq((2, "c"), (2, "d")).toDF("id", "data").groupBy("id").count()
-      val tuples = df1.hint("merge").join(df2, "id").select("*").collect()
-      checkStorageValidation(checkingThread)
-      // verify result
-      assert(hook.executed.get())
-      val expect = "[2,1,2]"
-      assert(tuples.head.toString().equals(expect))
-      sparkSession.stop()
-    }
-  }
-
-  // 4. for single level M-1 lineage, all failed disk spaces are cleaned
-  test("celeborn spark integration test - (M-1 dep with single-level lineage) all failed disk spaces are cleaned") {
-    if (Spark3OrNewer) {
-      val sparkSession = createSparkSession(enableFailedShuffleCleaner = true)
-      val celebornConf = SparkUtils.fromSparkConf(sparkSession.sparkContext.getConf)
-      val hook = new FileDeletionShuffleReaderGetHook(
-        celebornConf,
-        workerDirs,
-        shuffleIdToBeDeleted = Seq(0, 1))
-      TestCelebornShuffleManager.registerReaderGetHook(hook)
-      val checkingThread = triggerStorageCheckThread(
-        Seq(0, 1),
-        Seq(2, 3),
-        sparkSession,
-        forStableStatusChecking = false)
-      import sparkSession.implicits._
-      val df1 = Seq((1, "a"), (2, "b")).toDF("id", "data").groupBy("id").count()
-      val df2 = Seq((2, "c"), (3, "d")).toDF("id", "data").groupBy("id").count()
-      val tuples = df1.hint("merge").join(df2, "id").select("*").collect()
-      checkStorageValidation(checkingThread)
-      // verify result
-      assert(hook.executed.get())
-      val expect = "[2,1,1]"
-      assert(tuples.head.toString().equals(expect))
-      sparkSession.stop()
-    }
-  }
 }