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

Author: charlesmyu

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

@@ -14,6 +14,8 @@
 import org.slf4j.LoggerFactory;
 import scala.Predef;
 import scala.collection.JavaConverters;
+import scala.collection.Map;
+import scala.collection.Seq;
 
 @AutoService(InstrumenterModule.class)
 public class Spark212Instrumentation extends AbstractSparkInstrumentation {
@@ -101,14 +103,26 @@ public static void exit(
           && (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());
+        scala.collection.immutable.Map<String, String> meta =
+            JavaConverters.mapAsScalaMap(planUtils.extractFormattedProduct(plan))
+                .toMap(Predef.$conforms());
+        try {
+          Class<?> spiClass = Class.forName("org.apache.spark.sql.execution.SparkPlanInfo");
+          java.lang.reflect.Constructor<?> targetCtor =
+              spiClass.getConstructor(
+                  new Class[] {String.class, String.class, Seq.class, Map.class, Seq.class});
+          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

@@ -14,6 +14,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.collection.JavaConverters;
+import scala.collection.Map;
+import scala.collection.Seq;
 import scala.collection.immutable.HashMap;
 
 @AutoService(InstrumenterModule.class)
@@ -108,13 +110,9 @@ public static void exit(
             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()) {
-            if (c.getParameterCount() == 5) {
-              targetCtor = c;
-              break;
-            }
-          }
+          java.lang.reflect.Constructor<?> targetCtor =
+              spiClass.getConstructor(
+                  new Class[] {String.class, String.class, Seq.class, Map.class, Seq.class});
           if (targetCtor != null) {
             Object newInst =
                 targetCtor.newInstance(