wait for throwable and let the span be finished by shutdown hook

Author: aboitreaud

dd-java-agent/instrumentation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/SparkLauncherListener.java

@@ -97,7 +97,12 @@ public void stateChanged(SparkAppHandle handle) {
         if (state == SparkAppHandle.State.FAILED
             || state == SparkAppHandle.State.KILLED
             || state == SparkAppHandle.State.LOST) {
-          finishSpan(true, "Application " + state);
+          // Set error tags but don't finish yet — RunMainAdvice may add the throwable
+          // with the full stack trace. The span will be finished by RunMainAdvice or
+          // the shutdown hook.
+          span.setError(true);
+          span.setTag(DDTags.ERROR_TYPE, "Spark Launcher Failed");
+          span.setTag(DDTags.ERROR_MSG, "Application " + state);
         } else {
           finishSpan(false, null);
         }

dd-java-agent/instrumentation/spark/spark-common/src/test/groovy/datadog/trace/instrumentation/spark/SparkLauncherTest.groovy

@@ -130,7 +130,7 @@ class SparkLauncherTest extends InstrumentationSpecification {
     }
   }
 
-  def "SparkLauncherListener finishes span with error on FAILED state"() {
+  def "SparkLauncherListener sets error tags on FAILED state but does not finish span"() {
     setup:
     SparkLauncherListener.launcherSpan = null
     def tracer = AgentTracer.get()
@@ -150,6 +150,40 @@ class SparkLauncherTest extends InstrumentationSpecification {
     handle.getAppId() >> "app-456"
     listener.stateChanged(handle)
 
+    then:
+    // Span stays open so RunMainAdvice can add the throwable
+    SparkLauncherListener.launcherSpan != null
+    SparkLauncherListener.launcherSpan.isError()
+    SparkLauncherListener.launcherSpan.getTags()["error.type"] == "Spark Launcher Failed"
+    SparkLauncherListener.launcherSpan.getTags()["error.message"] == "Application FAILED"
+    SparkLauncherListener.launcherSpan.getTags()["spark.app_id"] == "app-456"
+
+    cleanup:
+    SparkLauncherListener.finishSpan(false, null)
+  }
+
+  def "finishSpanWithThrowable adds stack trace after FAILED state"() {
+    setup:
+    SparkLauncherListener.launcherSpan = null
+    def tracer = AgentTracer.get()
+    SparkLauncherListener.launcherSpan = tracer
+      .buildSpan("spark.launcher.launch")
+      .withSpanType("spark")
+      .withResourceName("SparkLauncher.startApplication")
+      .start()
+    SparkLauncherListener.launcherSpan.setSamplingPriority(
+      PrioritySampling.USER_KEEP,
+      SamplingMechanism.DATA_JOBS)
+    def listener = new SparkLauncherListener()
+    def handle = Mock(SparkAppHandle)
+
+    when:
+    // Simulate: listener sets error tags, then RunMainAdvice finishes with throwable
+    handle.getState() >> SparkAppHandle.State.FAILED
+    handle.getAppId() >> "app-456"
+    listener.stateChanged(handle)
+    SparkLauncherListener.finishSpanWithThrowable(new RuntimeException("job crashed"))
+
     then:
     SparkLauncherListener.launcherSpan == null
     assertTraces(1) {
@@ -158,8 +192,9 @@ class SparkLauncherTest extends InstrumentationSpecification {
           operationName "spark.launcher.launch"
           spanType "spark"
           errored true
-          assert span.tags["error.type"] == "Spark Launcher Failed"
-          assert span.tags["error.message"] == "Application FAILED"
+          assert span.tags["error.type"] == "java.lang.RuntimeException"
+          assert span.tags["error.message"] == "job crashed"
+          assert span.tags["error.stack"] != null
           assert span.tags["spark.app_id"] == "app-456"
         }
       }