Skip Dataproc shuffle manager auto-configuration

docs/additional-functionality/advanced_configs.md

@@ -56,7 +56,7 @@ Name | Description | Default Value | Applicable at
 <a name="python.memory.gpu.allocFraction"></a>spark.rapids.python.memory.gpu.allocFraction|The fraction of total GPU memory that should be initially allocated for pooled memory for all the Python workers. It supposes to be less than (1 - $(spark.rapids.memory.gpu.allocFraction)), since the executor will share the GPU with its owning Python workers. Half of the rest will be used if not specified|None|Runtime
 <a name="python.memory.gpu.maxAllocFraction"></a>spark.rapids.python.memory.gpu.maxAllocFraction|The fraction of total GPU memory that limits the maximum size of the RMM pool for all the Python workers. It supposes to be less than (1 - $(spark.rapids.memory.gpu.maxAllocFraction)), since the executor will share the GPU with its owning Python workers. when setting to 0 it means no limit.|0.0|Runtime
 <a name="python.memory.gpu.pooling.enabled"></a>spark.rapids.python.memory.gpu.pooling.enabled|Should RMM in Python workers act as a pooling allocator for GPU memory, or should it just pass through to CUDA memory allocation directly. When not specified, It will honor the value of config 'spark.rapids.memory.gpu.pool', but now only 'DEFAULT' and 'NONE' are supported. If 'ASYNC' or 'ARENA' is specified, it will fall back to 'DEFAULT'.|None|Runtime
-<a name="shuffle.enabled"></a>spark.rapids.shuffle.enabled|Enable or disable the RAPIDS Shuffle Manager implementation at runtime. On supported Spark versions, including Spark 4.0.0 and later, the [RAPIDS Shuffle Manager](https://docs.nvidia.com/spark-rapids/user-guide/latest/additional-functionality/rapids-shuffle.html) is configured automatically unless spark.shuffle.manager is explicitly set. On earlier Spark versions, the RAPIDS Shuffle Manager must already be configured. When set to `false`, the built-in Spark shuffle implementation will be used. |true|Runtime
+<a name="shuffle.enabled"></a>spark.rapids.shuffle.enabled|Enable or disable the RAPIDS Shuffle Manager implementation at runtime. On supported Spark versions, including Spark 4.0.0 and later, the [RAPIDS Shuffle Manager](https://docs.nvidia.com/spark-rapids/user-guide/latest/additional-functionality/rapids-shuffle.html) is configured automatically unless spark.shuffle.manager is explicitly set. This automatic configuration is skipped on Dataproc runtimes that set spark.dataproc.engine, including Lightning Engine runtimes; on those runtimes, spark.shuffle.manager remains unset unless explicitly configured. On earlier Spark versions, the RAPIDS Shuffle Manager must already be configured. When set to `false`, the built-in Spark shuffle implementation will be used. |true|Runtime
 <a name="shuffle.mode"></a>spark.rapids.shuffle.mode|RAPIDS Shuffle Manager mode. "MULTITHREADED": shuffle file writes and reads are parallelized using a thread pool. "UCX": (requires UCX installation) uses accelerated transports for transferring shuffle blocks. "CACHE_ONLY": use when running a single executor, for short-circuit cached shuffle (for testing purposes).|MULTITHREADED|Startup
 <a name="shuffle.multiThreaded.maxBytesInFlight"></a>spark.rapids.shuffle.multiThreaded.maxBytesInFlight|The size limit, in bytes, that the RAPIDS shuffle manager configured in "MULTITHREADED" mode will allow to be serialized or deserialized concurrently per task. This is also the maximum amount of memory that will be used per task. This should be set larger than Spark's default maxBytesInFlight (48MB). The larger this setting is, the more compressed shuffle chunks are processed concurrently. In practice, care needs to be taken to not go over the amount of off-heap memory that Netty has available. See https://github.com/NVIDIA/cudf-spark/issues/9153.|134217728|Startup
 <a name="shuffle.multiThreaded.reader.threads"></a>spark.rapids.shuffle.multiThreaded.reader.threads|The number of threads to use for reading shuffle blocks per executor in the RAPIDS shuffle manager configured in "MULTITHREADED" mode. There are two special values: 0 = feature is disabled, falls back to Spark built-in shuffle reader; 1 = our implementation of Spark's built-in shuffle reader with extra metrics.|20|Startup

sql-plugin/src/main/scala/com/nvidia/spark/rapids/Plugin.scala

@@ -59,9 +59,11 @@ object RapidsShuffleManagerAutoConfigurator {
   private val SHUFFLE_MANAGER_KEY = "spark.shuffle.manager"
   private val SHUFFLE_DATA_IO_PLUGIN_KEY = "spark.shuffle.sort.io.plugin.class"
   private val RAPIDS_SHUFFLE_DATA_IO_CLASS_SUFFIX = "RapidsLocalDiskShuffleDataIO"
+  private val DATAPROC_ENGINE_KEY = "spark.dataproc.engine"
 
   def configure(conf: SparkConf): Unit = {
     if (ShuffleManagerShimUtils.supportsAutoConfiguration &&
+        !conf.contains(DATAPROC_ENGINE_KEY) &&
         !conf.contains(SHUFFLE_MANAGER_KEY) &&
         conf.getOption(SHUFFLE_DATA_IO_PLUGIN_KEY)
           .forall(_.endsWith(RAPIDS_SHUFFLE_DATA_IO_CLASS_SUFFIX))) {

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

@@ -2200,9 +2200,11 @@ val GPU_COREDUMP_PIPE_PATTERN = conf("spark.rapids.gpu.coreDump.pipePattern")
       "Spark versions, including Spark 4.0.0 and later, the " +
       "[RAPIDS Shuffle Manager](https://docs.nvidia.com/spark-rapids/user-guide/latest" +
       "/additional-functionality/rapids-shuffle.html) is configured automatically unless " +
-      "spark.shuffle.manager is explicitly set. On earlier Spark versions, the RAPIDS Shuffle " +
-      "Manager must already be configured. When set to `false`, the built-in Spark shuffle " +
-      "implementation will be used. ")
+      "spark.shuffle.manager is explicitly set. This automatic configuration is skipped on " +
+      "Dataproc runtimes that set spark.dataproc.engine, including Lightning Engine runtimes; " +
+      "on those runtimes, spark.shuffle.manager remains unset unless explicitly configured. " +
+      "On earlier Spark versions, the RAPIDS Shuffle Manager must already be configured. When " +
+      "set to `false`, the built-in Spark shuffle implementation will be used. ")
     .booleanConf
     .createWithDefault(true)
 

sql-plugin/src/test/scala/com/nvidia/spark/rapids/RapidsPluginUtilsSuite.scala

@@ -66,6 +66,15 @@ class RapidsPluginUtilsSuite extends AnyFunSuite {
     assert(!conf.contains("spark.shuffle.manager"))
   }
 
+  test("shuffle manager is not auto-configured on Dataproc") {
+    val conf = new SparkConf(false)
+      .set("spark.dataproc.engine", "lightningEngine")
+
+    RapidsShuffleManagerAutoConfigurator.configure(conf)
+
+    assert(!conf.contains("spark.shuffle.manager"))
+  }
+
   test("shuffle manager runtime setting does not control auto-configuration") {
     val conf = new SparkConf(false)
       .set(RapidsConf.SHUFFLE_MANAGER_ENABLED.key, "false")