[SPARK-26507][CORE] Fix core tests for Java 11

## What changes were proposed in this pull request?

This should make tests in core modules pass for Java 11.

## How was this patch tested?

Existing tests, with modifications.

Closes apache#23419 from srowen/Java11.

Authored-by: Sean Owen <sean.owen@databricks.com>
Signed-off-by: Sean Owen <sean.owen@databricks.com>

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -2740,13 +2740,17 @@ private[spark] object Utils extends Logging {
 
   /**
    * Safer than Class obj's getSimpleName which may throw Malformed class name error in scala.
-   * This method mimicks scalatest's getSimpleNameOfAnObjectsClass.
+   * This method mimics scalatest's getSimpleNameOfAnObjectsClass.
    */
   def getSimpleName(cls: Class[_]): String = {
     try {
-      return cls.getSimpleName
+      cls.getSimpleName
     } catch {
-      case err: InternalError => return stripDollars(stripPackages(cls.getName))
+      // TODO: the value returned here isn't even quite right; it returns simple names
+      // like UtilsSuite$MalformedClassObject$MalformedClass instead of MalformedClass
+      // The exact value may not matter much as it's used in log statements
+      case _: InternalError =>
+        stripDollars(stripPackages(cls.getName))
     }
   }
 

core/src/test/scala/org/apache/spark/metrics/source/AccumulatorSourceSuite.scala

@@ -17,9 +17,8 @@
 
 package org.apache.spark.metrics.source
 
-import com.codahale.metrics.MetricRegistry
 import org.mockito.ArgumentCaptor
-import org.mockito.Mockito.{mock, never, spy, times, verify, when}
+import org.mockito.Mockito.{mock, times, verify, when}
 
 import org.apache.spark.{SparkContext, SparkEnv, SparkFunSuite}
 import org.apache.spark.metrics.MetricsSystem
@@ -37,7 +36,7 @@ class AccumulatorSourceSuite extends SparkFunSuite {
     val accs = Map("my-accumulator-1" -> acc1,
                    "my-accumulator-2" -> acc2)
     LongAccumulatorSource.register(mockContext, accs)
-    val captor = new ArgumentCaptor[AccumulatorSource]()
+    val captor = ArgumentCaptor.forClass(classOf[AccumulatorSource])
     verify(mockMetricSystem, times(1)).registerSource(captor.capture())
     val source = captor.getValue()
     val gauges = source.metricRegistry.getGauges()
@@ -59,7 +58,7 @@ class AccumulatorSourceSuite extends SparkFunSuite {
     val accs = Map("my-accumulator-1" -> acc1,
                    "my-accumulator-2" -> acc2)
     LongAccumulatorSource.register(mockContext, accs)
-    val captor = new ArgumentCaptor[AccumulatorSource]()
+    val captor = ArgumentCaptor.forClass(classOf[AccumulatorSource])
     verify(mockMetricSystem, times(1)).registerSource(captor.capture())
     val source = captor.getValue()
     val gauges = source.metricRegistry.getGauges()
@@ -81,7 +80,7 @@ class AccumulatorSourceSuite extends SparkFunSuite {
       "my-accumulator-1" -> acc1,
       "my-accumulator-2" -> acc2)
     DoubleAccumulatorSource.register(mockContext, accs)
-    val captor = new ArgumentCaptor[AccumulatorSource]()
+    val captor = ArgumentCaptor.forClass(classOf[AccumulatorSource])
     verify(mockMetricSystem, times(1)).registerSource(captor.capture())
     val source = captor.getValue()
     val gauges = source.metricRegistry.getGauges()

core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala

@@ -761,13 +761,13 @@ private[spark] object JsonProtocolSuite extends Assertions {
   }
 
   private def assertJsonStringEquals(expected: String, actual: String, metadata: String) {
-    val expectedJson = pretty(parse(expected))
-    val actualJson = pretty(parse(actual))
+    val expectedJson = parse(expected)
+    val actualJson = parse(actual)
     if (expectedJson != actualJson) {
       // scalastyle:off
       // This prints something useful if the JSON strings don't match
-      println("=== EXPECTED ===\n" + expectedJson + "\n")
-      println("=== ACTUAL ===\n" + actualJson + "\n")
+      println(s"=== EXPECTED ===\n${pretty(expectedJson)}\n")
+      println(s"=== ACTUAL ===\n${pretty(actualJson)}\n")
       // scalastyle:on
       throw new TestFailedException(s"$metadata JSON did not equal", 1)
     }
@@ -807,7 +807,13 @@ private[spark] object JsonProtocolSuite extends Assertions {
   }
 
   private def assertStackTraceElementEquals(ste1: StackTraceElement, ste2: StackTraceElement) {
-    assert(ste1 === ste2)
+    // This mimics the equals() method from Java 8 and earlier. Java 9 adds checks for
+    // class loader and module, which will cause them to be not equal, when we don't
+    // care about those
+    assert(ste1.getClassName === ste2.getClassName)
+    assert(ste1.getMethodName === ste2.getMethodName)
+    assert(ste1.getLineNumber === ste2.getLineNumber)
+    assert(ste1.getFileName === ste2.getFileName)
   }
 
   /** ----------------------------------- *

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -1156,22 +1156,6 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties with Logging {
     }
   }
 
-  object MalformedClassObject {
-    class MalformedClass
-  }
-
-  test("Safe getSimpleName") {
-    // getSimpleName on class of MalformedClass will result in error: Malformed class name
-    // Utils.getSimpleName works
-    val err = intercept[java.lang.InternalError] {
-      classOf[MalformedClassObject.MalformedClass].getSimpleName
-    }
-    assert(err.getMessage === "Malformed class name")
-
-    assert(Utils.getSimpleName(classOf[MalformedClassObject.MalformedClass]) ===
-      "UtilsSuite$MalformedClassObject$MalformedClass")
-  }
-
   test("stringHalfWidth") {
     // scalastyle:off nonascii
     assert(Utils.stringHalfWidth(null) == 0)

launcher/src/test/java/org/apache/spark/launcher/SparkSubmitCommandBuilderSuite.java

@@ -158,7 +158,7 @@ public void testPySparkLauncher() throws Exception {
 
     Map<String, String> env = new HashMap<>();
     List<String> cmd = buildCommand(sparkSubmitArgs, env);
-    assertEquals("python", cmd.get(cmd.size() - 1));
+    assertTrue(Arrays.asList("python", "python2", "python3").contains(cmd.get(cmd.size() - 1)));
     assertEquals(
       String.format("\"%s\" \"foo\" \"%s\" \"bar\" \"%s\"",
         parser.MASTER, parser.DEPLOY_MODE, SparkSubmitCommandBuilder.PYSPARK_SHELL_RESOURCE),

pom.xml

@@ -2060,6 +2060,8 @@
           </executions>
           <configuration>
             <scalaVersion>${scala.version}</scalaVersion>
+            <checkMultipleScalaVersions>true</checkMultipleScalaVersions>
+            <failOnMultipleScalaVersions>true</failOnMultipleScalaVersions>
             <recompileMode>incremental</recompileMode>
             <useZincServer>true</useZincServer>
             <args>