Add Spark 4.0.0 test suite on JDK 17 to spark_2.13 (#12049)

Add Spark 4.0.0 test suite on JDK 17 to spark_2.13

Mirror Spark's full JDK 17 module opens in test_spark4

Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>

Author: aboitreaud

dd-java-agent/instrumentation/spark/spark-common/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSparkListenerTest.groovy

@@ -60,6 +60,15 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
       )
   }
 
+  protected applicationEndEvent(Long time) {
+    // Constructor of SparkListenerApplicationEnd gained an exitCode parameter starting spark 4.0
+    if (TestSparkComputation.getSparkVersion() < "4") {
+      return new SparkListenerApplicationEnd(time)
+    }
+
+    return new SparkListenerApplicationEnd(time, Option.empty())
+  }
+
   protected jobStartEvent(Integer jobId, Long time, ArrayList<Integer> stageIds) {
     def stageInfos = stageIds.collect { stageId ->
       createStageInfo(stageId)
@@ -214,7 +223,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     listener.onExecutorAdded(executorAddedEvent(3000L, "executor-2", 5))
     listener.onExecutorRemoved(executorRemovedEvent(4000L, "executor-2"))
 
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(5000L))
+    listener.onApplicationEnd(applicationEndEvent(5000L))
 
     expect:
     def expectedExecutorTime = (5000 - 2000) * 4 + (4000 - 3000) * 5
@@ -256,7 +265,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     listener.onStageCompleted(stageCompletedEvent(3, 3100L))
 
     listener.onJobEnd(jobEndEvent(1, 3100L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(3100L))
+    listener.onApplicationEnd(applicationEndEvent(3100L))
 
     expect:
     /*
@@ -339,7 +348,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     listener.onStageCompleted(stageCompletedEvent(2, 3000L))
 
     listener.onJobEnd(jobEndEvent(1, 3000L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(3000L))
+    listener.onApplicationEnd(applicationEndEvent(3000L))
 
     expect:
     assertTraces(1) {
@@ -384,7 +393,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     listener.onTaskEnd(taskEndEvent(1,1900L, 300))
     listener.onStageCompleted(stageCompletedEvent(1, 2200L))
     listener.onJobEnd(jobEndEvent(1, 17200L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(3100L))
+    listener.onApplicationEnd(applicationEndEvent(3100L))
 
     assertTraces(1) {
       trace(3) {
@@ -430,7 +439,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
 
     listener.onStageCompleted(stageCompletedEvent(2, 17200L))
     listener.onJobEnd(jobEndEvent(1, 17200L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(3100L))
+    listener.onApplicationEnd(applicationEndEvent(3100L))
 
     expect:
     def relativeAccuracy = 1/32D
@@ -480,7 +489,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
 
     when:
     listener.onApplicationStart(applicationStartEvent(1000L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(2000L))
+    listener.onApplicationEnd(applicationEndEvent(2000L))
 
     then:
     assertTraces(1) {
@@ -500,7 +509,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     listener.onApplicationStart(applicationStartEvent(1000L))
     listener.onJobStart(jobStartEvent(1, 1900L, [1]))
     listener.onJobEnd(jobFailedEvent(1, 2200L, "Job was cancelled by user"))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(2300L))
+    listener.onApplicationEnd(applicationEndEvent(2300L))
 
     expect:
     assertTraces(1) {
@@ -531,7 +540,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     def analysisException = new RuntimeException("[TABLE_OR_VIEW_NOT_FOUND] The table or view `missing_table` cannot be found.")
     listener.onSqlFailure(analysisException)
 
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(2000L))
+    listener.onApplicationEnd(applicationEndEvent(2000L))
 
     expect:
     assertTraces(1) {
@@ -563,7 +572,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     listener.onStageCompleted(stageCompletedEvent(1, 1800L))
     listener.onJobEnd(jobFailedEvent(1, 2000L, "Job aborted due to NullPointerException"))
 
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(3000L))
+    listener.onApplicationEnd(applicationEndEvent(3000L))
 
     expect:
     assertTraces(1) {
@@ -663,7 +672,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     injectSysConfig("dd.tags", ddTags)
     def listener = getTestDatadogSparkListener()
     listener.onApplicationStart(applicationStartEvent(1000L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(5000L))
+    listener.onApplicationEnd(applicationEndEvent(5000L))
 
     expect:
     assertTraces(1) {
@@ -729,7 +738,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
 
     when:
     listener.onApplicationStart(applicationStartEvent(1000L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(2000L))
+    listener.onApplicationEnd(applicationEndEvent(2000L))
 
     then:
     assertTraces(1) {
@@ -749,7 +758,7 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
 
     when:
     listener.onApplicationStart(applicationStartEvent(1000L))
-    listener.onApplicationEnd(new SparkListenerApplicationEnd(2000L))
+    listener.onApplicationEnd(applicationEndEvent(2000L))
 
     then:
     assertTraces(1) {

dd-java-agent/instrumentation/spark/spark-common/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSparkTest.groovy

@@ -157,7 +157,8 @@ abstract class AbstractSparkTest extends InstrumentationSpecification {
           errored true
           parent()
           assert span.tags["error.type"] == "Spark Application Failed"
-          assert span.tags["error.message"] =~ /^Job aborted due to stage failure.*java.lang.NullPointerException$/
+          // JDK 15+'s helpful NullPointerException messages (JEP 358) append extra detail after the exception class name.
+          assert span.tags["error.message"] =~ /^Job aborted due to stage failure.*java.lang.NullPointerException.*$/
           assert span.tags["error.stack"] =~ /(?s)^org.apache.spark.SparkException.*Caused by: java.lang.NullPointerException.*$/
         }
         span {
@@ -167,7 +168,8 @@ abstract class AbstractSparkTest extends InstrumentationSpecification {
           errored true
           childOf(span(0))
           assert span.tags["error.type"] == "Spark Job Failed"
-          assert span.tags["error.message"] =~ /^Job aborted due to stage failure.*java.lang.NullPointerException$/
+          // JDK 15+'s helpful NullPointerException messages (JEP 358) append extra detail after the exception class name.
+          assert span.tags["error.message"] =~ /^Job aborted due to stage failure.*java.lang.NullPointerException.*$/
           assert span.tags["error.stack"] =~ /(?s)^org.apache.spark.SparkException.*Caused by: java.lang.NullPointerException.*$/
         }
         span {
@@ -177,7 +179,8 @@ abstract class AbstractSparkTest extends InstrumentationSpecification {
           errored true
           childOf(span(1))
           assert span.tags["error.type"] == "Spark Stage Failed"
-          assert span.tags["error.message"] =~ /^Job aborted due to stage failure.*java.lang.NullPointerException$/
+          // JDK 15+'s helpful NullPointerException messages (JEP 358) append extra detail after the exception class name.
+          assert span.tags["error.message"] =~ /^Job aborted due to stage failure.*java.lang.NullPointerException.*$/
           assert span.tags["error.stack"] =~ /(?s).*\n\tat datadog.trace.instrumentation.spark.TestSparkComputation.{500,}\$/
         }
         span {
@@ -186,8 +189,9 @@ abstract class AbstractSparkTest extends InstrumentationSpecification {
           errored true
           childOf(span(2))
           assert span.tags["error.type"] == "Spark Task Failed"
-          assert span.tags["error.message"] == "java.lang.NullPointerException: null"
-          assert span.tags["error.stack"] =~ /(?s)^java.lang.NullPointerException\n\tat datadog.trace.instrumentation.spark.TestSparkComputation.{500,}\$/
+          // JDK 15+'s helpful NullPointerException messages (JEP 358) append extra detail after the exception class name.
+          assert span.tags["error.message"] =~ /^java\.lang\.NullPointerException: .*$/
+          assert span.tags["error.stack"] =~ /(?s)^java.lang.NullPointerException.*\n\tat datadog.trace.instrumentation.spark.TestSparkComputation.{500,}\$/
         }
       }
     }
@@ -829,6 +833,9 @@ abstract class AbstractSparkTest extends InstrumentationSpecification {
     def sparkSession = SparkSession.builder()
       .config("spark.master", "local[2]")
       .config("spark.sql.shuffle.partitions", "2")
+      // Under ANSI mode (default since Spark 4.0), casting the non-numeric "value" column
+      // content to a number for the filter predicate below throws instead of yielding null.
+      .config("spark.sql.ansi.enabled", "false")
       .getOrCreate()
 
     def df = generateSampleDataframe(sparkSession)

dd-java-agent/instrumentation/spark/spark_2.13/build.gradle

@@ -24,6 +24,7 @@ apply from: "$rootDir/gradle/java.gradle"
 
 addTestSuiteForDir('latestDepTest', 'test')
 addTestSuite('test_spark32')
+addTestSuiteForDir('test_spark4', 'test')
 
 testJvmConstraints {
   // Hadoop does not behave correctly with OpenJ9 https://issues.apache.org/jira/browse/HADOOP-18174
@@ -53,6 +54,10 @@ dependencies {
   test_spark32Implementation group: 'org.apache.spark', name: "spark-sql_$scalaVersion", version: "3.2.4"
   test_spark32Implementation group: 'org.apache.spark', name: "spark-yarn_$scalaVersion", version: "3.2.4"
 
+  test_spark4Implementation group: 'org.apache.spark', name: "spark-core_$scalaVersion", version: "4.0.0"
+  test_spark4Implementation group: 'org.apache.spark', name: "spark-sql_$scalaVersion", version: "4.0.0"
+  test_spark4Implementation group: 'org.apache.spark', name: "spark-yarn_$scalaVersion", version: "4.0.0"
+
   // FIXME: Currently not working on Spark 4.0.0 preview releases.
   latestDepTestImplementation group: 'org.apache.spark', name: "spark-core_$scalaVersion", version: '3.+'
   latestDepTestImplementation group: 'org.apache.spark', name: "spark-sql_$scalaVersion", version: '3.+'
@@ -61,4 +66,51 @@ dependencies {
 
 tasks.named("test", Test) {
   dependsOn "test_spark32"
+  dependsOn "test_spark4"
+}
+
+tasks.named("test_spark4", Test) {
+  testJvmConstraints {
+    // Override the module-wide maxJavaVersion = 11 cap (set above for the Spark 3.2 test JVM
+    // constraints) since Spark 4.0.0 requires JDK 17+.
+    minJavaVersion = JavaVersion.VERSION_17
+    // Hadoop's UserGroupInformation calls javax.security.auth.Subject.getSubject(), which
+    // throws UnsupportedOperationException once the SecurityManager is fully removed in JDK 24+
+    // (JEP 486).
+    maxJavaVersion = JavaVersion.VERSION_23
+  }
+
+  // Spark itself only opens these modules when launched through its own scripts
+  // (org.apache.spark.launcher.JavaModuleOptions); since this suite starts a SparkSession
+  // directly in-process, mirror that same list so Spark's own reflective access (Kryo shuffle
+  // serializers, Tungsten unsafe memory, Hadoop UGI, etc.) doesn't fail under JDK 16+ module encapsulation.
+  conditionalJvmArgs(
+    it,
+    JavaVersion.VERSION_16,
+    [
+      "--add-opens=java.base/java.lang=ALL-UNNAMED",
+      "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED",
+      "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED",
+      "--add-opens=java.base/java.io=ALL-UNNAMED",
+      "--add-opens=java.base/java.net=ALL-UNNAMED",
+      "--add-opens=java.base/java.nio=ALL-UNNAMED",
+      "--add-opens=java.base/java.util=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED",
+      "--add-opens=java.base/jdk.internal.ref=ALL-UNNAMED",
+      "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED",
+      "--add-opens=java.base/sun.nio.cs=ALL-UNNAMED",
+      "--add-opens=java.base/sun.security.action=ALL-UNNAMED",
+      "--add-opens=java.base/sun.util.calendar=ALL-UNNAMED",
+      "--add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED"
+    ]
+    )
+}
+
+// Spark 4.0.0 transitively pulls a newer protobuf-java that's binary-incompatible with the
+// pre-generated com.datadoghq.sketch.ddsketch.proto.DDSketch classes used by the test fixtures.
+['test_spark4CompileClasspath', 'test_spark4RuntimeClasspath'].each {
+  configurations.named(it) {
+    resolutionStrategy.force 'com.google.protobuf:protobuf-java:3.14.0'
+  }
 }