[SPARK-52548][CORE][TESTS] Add a test case for when shuffle manager is overridden by a SparkPlugin

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

As title, adding a test case for when shuffle manager is overridden by a SparkPlugin.

### Why are the changes needed?

PR #43627 introduced a change to allow the shuffle manager specified in Spark configuration to be overridden in SparkPlugin, this change was not guarded by test though.

The change in PR #43627 also allows Spark plugins to configure a shuffle manager automatically for user, so user won't have to configure both `spark.plugins` and `spark.shuffle.manager` at the same time.

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

No.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #51246 from zhztheplayer/wip-test-plugin-shuffle.

Lead-authored-by: Hongze Zhang <[email protected]>
Co-authored-by: Kent Yao <[email protected]>
Signed-off-by: Kent Yao <[email protected]>

Author: zhztheplayer

core/src/test/scala/org/apache/spark/internal/plugin/PluginContainerSuite.scala

@@ -42,6 +42,7 @@ import org.apache.spark.resource.ResourceInformation
 import org.apache.spark.resource.ResourceUtils.GPU
 import org.apache.spark.resource.TestResourceIDs.{DRIVER_GPU_ID, EXECUTOR_GPU_ID, WORKER_GPU_ID}
 import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
+import org.apache.spark.shuffle.sort.SortShuffleManager
 import org.apache.spark.util.Utils
 
 class PluginContainerSuite extends SparkFunSuite with LocalSparkContext {
@@ -292,6 +293,20 @@ class PluginContainerSuite extends SparkFunSuite with LocalSparkContext {
     // If the listener bus is stopped before the plugin is shutdown,
     // then the event will be dropped and won't be delivered to the listener.
   }
+
+  test("SPARK-52548: override shuffle manager in plugin") {
+    val conf = new SparkConf()
+      .setAppName(getClass().getName())
+      .set(SparkLauncher.SPARK_MASTER, "local[1]")
+      .set(SHUFFLE_MANAGER, "sort")
+      .set(PLUGINS, Seq(classOf[SetShuffleManagerPlugin].getName()))
+
+    sc = new SparkContext(conf)
+
+    // Ensures the shuffle manager specified in configuration was
+    // overridden by the Spark plugin.
+    assert(sc.env.shuffleManager.isInstanceOf[SetShuffleManagerPlugin.MyShuffleManager])
+  }
 }
 
 class MemoryOverridePlugin extends SparkPlugin {
@@ -391,6 +406,26 @@ object NonLocalModeSparkPlugin {
   }
 }
 
+class SetShuffleManagerPlugin extends SparkPlugin {
+  import SetShuffleManagerPlugin._
+  override def driverPlugin(): DriverPlugin = {
+    new DriverPlugin {
+      override def init(sc: SparkContext, ctx: PluginContext): JMap[String, String] = {
+        sc.conf.set(SHUFFLE_MANAGER, classOf[MyShuffleManager].getName)
+        Map.empty[String, String].asJava
+      }
+    }
+  }
+
+  override def executorPlugin(): ExecutorPlugin = {
+    new ExecutorPlugin {}
+  }
+}
+
+private object SetShuffleManagerPlugin {
+  class MyShuffleManager(conf: SparkConf) extends SortShuffleManager(conf)
+}
+
 class TestSparkPlugin extends SparkPlugin {
 
   override def driverPlugin(): DriverPlugin = {