[DJM-974] Use reflection for constructor in Scala 2.12, lookup by parameter classes (#9886)

Author: charlesmyu

dd-java-agent/instrumentation/spark/spark_2.12/src/main/java/datadog/trace/instrumentation/spark/Spark212Instrumentation.java

@@ -6,6 +6,8 @@
 import com.google.auto.service.AutoService;
 import datadog.trace.agent.tooling.InstrumenterModule;
 import datadog.trace.api.Config;
+import de.thetaphi.forbiddenapis.SuppressForbidden;
+import java.lang.reflect.Constructor;
 import net.bytebuddy.asm.Advice;
 import org.apache.spark.SparkContext;
 import org.apache.spark.sql.execution.SparkPlan;
@@ -14,6 +16,7 @@
 import org.slf4j.LoggerFactory;
 import scala.Predef;
 import scala.collection.JavaConverters;
+import scala.collection.immutable.Map;
 
 @AutoService(InstrumenterModule.class)
 public class Spark212Instrumentation extends AbstractSparkInstrumentation {
@@ -94,21 +97,37 @@ public static void enter(@Advice.This SparkContext sparkContext) {
 
   public static class SparkPlanInfoAdvice {
     @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class)
+    @SuppressForbidden
     public static void exit(
         @Advice.Return(readOnly = false) SparkPlanInfo planInfo,
         @Advice.Argument(0) SparkPlan plan) {
       if (planInfo.metadata().size() == 0
           && (Config.get().isDataJobsParseSparkPlanEnabled()
               || Config.get().isDataJobsExperimentalFeaturesEnabled())) {
         Spark212PlanSerializer planUtils = new Spark212PlanSerializer();
-        planInfo =
-            new SparkPlanInfo(
-                planInfo.nodeName(),
-                planInfo.simpleString(),
-                planInfo.children(),
-                JavaConverters.mapAsScalaMap(planUtils.extractFormattedProduct(plan))
-                    .toMap(Predef.$conforms()),
-                planInfo.metrics());
+        Map<String, String> meta =
+            JavaConverters.mapAsScalaMap(planUtils.extractFormattedProduct(plan))
+                .toMap(Predef.$conforms());
+        try {
+          Constructor<?> targetCtor = null;
+          for (Constructor<?> c : SparkPlanInfo.class.getConstructors()) {
+            if (c.getParameterCount() == 5) {
+              targetCtor = c;
+              break;
+            }
+          }
+          if (targetCtor != null) {
+            Object newInst =
+                targetCtor.newInstance(
+                    planInfo.nodeName(),
+                    planInfo.simpleString(),
+                    planInfo.children(),
+                    meta,
+                    planInfo.metrics());
+            planInfo = (SparkPlanInfo) newInst;
+          }
+        } catch (Throwable ignored) {
+        }
       }
     }
   }

dd-java-agent/instrumentation/spark/spark_2.13/src/main/java/datadog/trace/instrumentation/spark/Spark213Instrumentation.java

@@ -7,6 +7,7 @@
 import datadog.trace.agent.tooling.InstrumenterModule;
 import datadog.trace.api.Config;
 import de.thetaphi.forbiddenapis.SuppressForbidden;
+import java.lang.reflect.Constructor;
 import net.bytebuddy.asm.Advice;
 import org.apache.spark.SparkContext;
 import org.apache.spark.sql.execution.SparkPlan;
@@ -15,6 +16,7 @@
 import org.slf4j.LoggerFactory;
 import scala.collection.JavaConverters;
 import scala.collection.immutable.HashMap;
+import scala.collection.immutable.Map;
 
 @AutoService(InstrumenterModule.class)
 public class Spark213Instrumentation extends AbstractSparkInstrumentation {
@@ -104,12 +106,11 @@ public static void exit(
           && (Config.get().isDataJobsParseSparkPlanEnabled()
               || Config.get().isDataJobsExperimentalFeaturesEnabled())) {
         Spark213PlanSerializer planUtils = new Spark213PlanSerializer();
-        scala.collection.immutable.Map<String, String> meta =
+        Map<String, String> meta =
             HashMap.from(JavaConverters.asScala(planUtils.extractFormattedProduct(plan)));
         try {
-          Class<?> spiClass = Class.forName("org.apache.spark.sql.execution.SparkPlanInfo");
-          java.lang.reflect.Constructor<?> targetCtor = null;
-          for (java.lang.reflect.Constructor<?> c : spiClass.getConstructors()) {
+          Constructor<?> targetCtor = null;
+          for (Constructor<?> c : SparkPlanInfo.class.getConstructors()) {
             if (c.getParameterCount() == 5) {
               targetCtor = c;
               break;