From 5658635362a739cd652b22545554951d66eef7f1 Mon Sep 17 00:00:00 2001 From: Chandni Singh Date: Mon, 28 Jun 2021 12:31:20 -0700 Subject: [PATCH 1/2] SPARK-35917 Disable push-based shuffle feature to prevent it from being used --- .../org/apache/spark/network/yarn/YarnShuffleService.java | 4 ++++ core/src/main/scala/org/apache/spark/util/Utils.scala | 7 ++++++- .../org/apache/spark/scheduler/DAGSchedulerSuite.scala | 2 ++ .../spark/shuffle/HostLocalShuffleReadingSuite.scala | 2 ++ core/src/test/scala/org/apache/spark/util/UtilsSuite.scala | 2 +- 5 files changed, 15 insertions(+), 2 deletions(-) diff --git a/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java b/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java index cb6d5d0ca2037..ddd16b5ac95de 100644 --- a/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java +++ b/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java @@ -220,6 +220,10 @@ protected void serviceInit(Configuration externalConf) throws Exception { TransportConf transportConf = new TransportConf("shuffle", new HadoopConfigProvider(_conf)); MergedShuffleFileManager shuffleMergeManager = newMergedShuffleFileManagerInstance( transportConf); + if (!(shuffleMergeManager instanceof ExternalBlockHandler.NoOpMergedShuffleFileManager)) { + // TODO: Remove this once push-based shuffle is fully supported. + throw new UnsupportedOperationException("Push-based shuffle is not yet supported."); + } blockHandler = new ExternalBlockHandler( transportConf, registeredExecutorFile, shuffleMergeManager); diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala index b1df7bd03d98c..fdf29da3f5886 100644 --- a/core/src/main/scala/org/apache/spark/util/Utils.scala +++ b/core/src/main/scala/org/apache/spark/util/Utils.scala @@ -2598,11 +2598,16 @@ private[spark] object Utils extends Logging { * Support push based shuffle with multiple app attempts */ def isPushBasedShuffleEnabled(conf: SparkConf): Boolean = { - conf.get(PUSH_BASED_SHUFFLE_ENABLED) && + val isPushBasedShuffleEnabled = conf.get(PUSH_BASED_SHUFFLE_ENABLED) && (conf.get(IS_TESTING).getOrElse(false) || (conf.get(SHUFFLE_SERVICE_ENABLED) && conf.get(SparkLauncher.SPARK_MASTER, null) == "yarn" && getYarnMaxAttempts(conf) == 1)) + if (isPushBasedShuffleEnabled && !conf.get(IS_TESTING).getOrElse(false)) { + // TODO: Remove this once push-based shuffle is fully supported. + throw new UnsupportedOperationException("Push-based shuffle is not yet supported.") + } + isPushBasedShuffleEnabled } /** diff --git a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala index f6e87ee38b130..34bc3c78cab07 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala @@ -35,6 +35,7 @@ import org.apache.spark._ import org.apache.spark.broadcast.BroadcastManager import org.apache.spark.executor.ExecutorMetrics import org.apache.spark.internal.config +import org.apache.spark.internal.config.Tests.IS_TESTING import org.apache.spark.rdd.{DeterministicLevel, RDD} import org.apache.spark.resource.{ExecutorResourceRequests, ResourceProfile, ResourceProfileBuilder, TaskResourceRequests} import org.apache.spark.resource.ResourceUtils.{FPGA, GPU} @@ -3426,6 +3427,7 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti private def initPushBasedShuffleConfs(conf: SparkConf) = { conf.set(config.SHUFFLE_SERVICE_ENABLED, true) conf.set(config.PUSH_BASED_SHUFFLE_ENABLED, true) + conf.set(IS_TESTING, true) conf.set("spark.master", "pushbasedshuffleclustermanager") } diff --git a/core/src/test/scala/org/apache/spark/shuffle/HostLocalShuffleReadingSuite.scala b/core/src/test/scala/org/apache/spark/shuffle/HostLocalShuffleReadingSuite.scala index 33f544a3911cd..661b8a52b3893 100644 --- a/core/src/test/scala/org/apache/spark/shuffle/HostLocalShuffleReadingSuite.scala +++ b/core/src/test/scala/org/apache/spark/shuffle/HostLocalShuffleReadingSuite.scala @@ -22,6 +22,7 @@ import org.scalatest.matchers.should.Matchers._ import org.apache.spark._ import org.apache.spark.internal.config._ +import org.apache.spark.internal.config.Tests.IS_TESTING import org.apache.spark.network.TransportContext import org.apache.spark.network.netty.{NettyBlockTransferService, SparkTransportConf} import org.apache.spark.network.server.TransportServer @@ -136,6 +137,7 @@ class HostLocalShuffleReadingSuite extends SparkFunSuite with Matchers with Loca test("Enable host local shuffle reading when push based shuffle is enabled") { val conf = new SparkConf() + .set(IS_TESTING, true) .set(SHUFFLE_SERVICE_ENABLED, true) .set("spark.yarn.maxAttempts", "1") .set(PUSH_BASED_SHUFFLE_ENABLED, true) diff --git a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala index dba7e39e93973..94abde1b3103f 100644 --- a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala +++ b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala @@ -1448,7 +1448,7 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties with Logging { conf.set(SHUFFLE_SERVICE_ENABLED, true) conf.set(SparkLauncher.SPARK_MASTER, "yarn") conf.set("spark.yarn.maxAttempts", "1") - assert(Utils.isPushBasedShuffleEnabled(conf) === true) + assertThrows[UnsupportedOperationException](Utils.isPushBasedShuffleEnabled(conf)) conf.set("spark.yarn.maxAttempts", "2") assert(Utils.isPushBasedShuffleEnabled(conf) === false) } From e3f75abe3ceae7e40a87b0ffdf1ebd36535009c8 Mon Sep 17 00:00:00 2001 From: Chandni Singh Date: Mon, 28 Jun 2021 16:12:55 -0700 Subject: [PATCH 2/2] Updated the config documentation for both the client and the server --- .../main/java/org/apache/spark/network/util/TransportConf.java | 2 ++ .../main/scala/org/apache/spark/internal/config/package.scala | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java index f051042a7adb4..4264a999a85dd 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java +++ b/common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java @@ -381,6 +381,8 @@ public boolean useOldFetchProtocol() { * 'org.apache.spark.network.shuffle.ExternalBlockHandler$NoOpMergedShuffleFileManager'. * To turn on push-based shuffle at a cluster level, set the configuration to * 'org.apache.spark.network.shuffle.RemoteBlockPushResolver'. + * + * Push-based shuffle is not yet supported. */ public String mergedShuffleFileManagerImpl() { return conf.get("spark.shuffle.server.mergedShuffleFileManagerImpl", diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 613a66d2d5aca..a6b5776e1ea8c 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -2079,7 +2079,7 @@ package object config { "conjunction with the server side flag spark.shuffle.server.mergedShuffleFileManagerImpl " + "which needs to be set with the appropriate " + "org.apache.spark.network.shuffle.MergedShuffleFileManager implementation for push-based " + - "shuffle to be enabled") + "shuffle to be enabled. Push-based shuffle is not yet supported.") .version("3.1.0") .booleanConf .createWithDefault(false)