[SPARK-32575][CORE][TESTS] Bump up timeouts in BlockManagerDecommissionIntegrationSuite to reduce flakyness

### What changes were proposed in this pull request?

As reported by HyukjinKwon, BlockManagerDecommissionIntegrationSuite test is apparently still flaky (even after #29226): #29226 (comment).

The new flakyness is because the executors are not launching in the 6 seconds time out I had given them when run under github checks.

Bumped up the timeouts.

### Why are the changes needed?

To make this test not flaky so that it can give us high signal if decommissioning regresses.

### Does this PR introduce _any_ user-facing change?

No, unit test only check.

### How was this patch tested?

No new tests. Just github and jenkins.

Closes #29388 from agrawaldevesh/more_bm_harden.

Authored-by: Devesh Agrawal <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dagrawal3409

core/src/test/scala/org/apache/spark/storage/BlockManagerDecommissionIntegrationSuite.scala

@@ -40,7 +40,7 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
   val TaskEnded = "TASK_ENDED"
   val JobEnded = "JOB_ENDED"
 
-  test(s"verify that an already running task which is going to cache data succeeds " +
+  testRetry(s"verify that an already running task which is going to cache data succeeds " +
     s"on a decommissioned executor after task start") {
     runDecomTest(true, false, TaskStarted)
   }
@@ -89,7 +89,7 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
 
     val sleepIntervalMs = whenToDecom match {
       // Increase the window of time b/w task started and ended so that we can decom within that.
-      case TaskStarted => 2000
+      case TaskStarted => 10000
       // Make one task take a really short time so that we can decommission right after it is
       // done but before its peers are done.
       case TaskEnded =>
@@ -176,11 +176,11 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
       } else {
         10.milliseconds
       }
-      eventually(timeout(6.seconds), interval(intervalMs)) {
+      eventually(timeout(20.seconds), interval(intervalMs)) {
         assert(getCandidateExecutorToDecom.isDefined)
       }
     } else {
-      ThreadUtils.awaitResult(asyncCount, 15.seconds)
+      ThreadUtils.awaitResult(asyncCount, 1.minute)
     }
 
     // Decommission one of the executors.
@@ -194,7 +194,7 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
     val decomTime = new SystemClock().getTimeMillis()
 
     // Wait for job to finish.
-    val asyncCountResult = ThreadUtils.awaitResult(asyncCount, 15.seconds)
+    val asyncCountResult = ThreadUtils.awaitResult(asyncCount, 1.minute)
     assert(asyncCountResult === numParts)
     // All tasks finished, so accum should have been increased numParts times.
     assert(accum.value === numParts)
@@ -226,7 +226,7 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
     }
 
     // Wait for our respective blocks to have migrated
-    eventually(timeout(30.seconds), interval(10.milliseconds)) {
+    eventually(timeout(1.minute), interval(10.milliseconds)) {
       if (persist) {
         // One of our blocks should have moved.
         val rddUpdates = blocksUpdated.filter { update =>