Update variable name to accumulatorToStageID

Author: charlesmyu

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

@@ -128,7 +128,7 @@ public abstract class AbstractDatadogSparkListener extends SparkListener {
   protected final HashMap<Long, SparkPlanInfo> sqlPlans = new HashMap<>();
   private final HashMap<String, SparkListenerExecutorAdded> liveExecutors = new HashMap<>();
 
-  private final Map<Long, Integer> acc2stage = new HashMap<>();
+  private final Map<Long, Integer> accumulatorToStageID = new HashMap<>();
 
   private volatile boolean isStreamingJob = false;
   private final boolean isRunningOnDatabricks;
@@ -644,7 +644,7 @@ public synchronized void onStageCompleted(SparkListenerStageCompleted stageCompl
 
     for (AccumulableInfo info :
         JavaConverters.asJavaCollection(stageInfo.accumulables().values())) {
-      acc2stage.put(info.id(), stageId);
+      accumulatorToStageID.put(info.id(), stageId);
     }
 
     Properties prop = stageProperties.remove(stageSpanKey);
@@ -676,7 +676,7 @@ public synchronized void onStageCompleted(SparkListenerStageCompleted stageCompl
 
     SparkPlanInfo sqlPlan = sqlPlans.get(sqlExecutionId);
     if (sqlPlan != null) {
-      SparkSQLUtils.addSQLPlanToStageSpan(span, sqlPlan, acc2stage, stageMetric, stageId);
+      SparkSQLUtils.addSQLPlanToStageSpan(span, sqlPlan, accumulatorToStageID, stageMetric, stageId);
     }
 
     span.finish(completionTimeMs * 1000);

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

@@ -23,12 +23,12 @@ public class SparkSQLUtils {
   public static void addSQLPlanToStageSpan(
       AgentSpan span,
       SparkPlanInfo plan,
-      Map<Long, Integer> accumulators,
+      Map<Long, Integer> accumulatorToStageID,
       SparkAggregatedTaskMetrics stageMetric,
       int stageId) {
     Set<Integer> parentStageIds = new HashSet<>();
     SparkPlanInfoForStage planForStage =
-        computeStageInfoForStage(plan, accumulators, stageId, parentStageIds, false);
+        computeStageInfoForStage(plan, accumulatorToStageID, stageId, parentStageIds, false);
 
     span.setTag("_dd.spark.sql_parent_stage_ids", parentStageIds.toString());
 
@@ -40,11 +40,11 @@ public static void addSQLPlanToStageSpan(
 
   public static SparkPlanInfoForStage computeStageInfoForStage(
       SparkPlanInfo plan,
-      Map<Long, Integer> accumulators,
+      Map<Long, Integer> accumulatorToStageID,
       int stageId,
       Set<Integer> parentStageIds,
       boolean foundStage) {
-    Set<Integer> stageIds = stageIdsForPlan(plan, accumulators);
+    Set<Integer> stageIds = stageIdsForPlan(plan, accumulatorToStageID);
 
     boolean hasStageInfo = !stageIds.isEmpty();
     boolean isForStage = stageIds.contains(stageId);
@@ -64,7 +64,7 @@ public static SparkPlanInfoForStage computeStageInfoForStage(
       List<SparkPlanInfoForStage> childrenForStage = new ArrayList<>();
       for (SparkPlanInfo child : children) {
         SparkPlanInfoForStage planForStage =
-            computeStageInfoForStage(child, accumulators, stageId, parentStageIds, true);
+            computeStageInfoForStage(child, accumulatorToStageID, stageId, parentStageIds, true);
 
         if (planForStage != null) {
           childrenForStage.add(planForStage);
@@ -76,7 +76,7 @@ public static SparkPlanInfoForStage computeStageInfoForStage(
       // The expected stage was not found yet, searching in the children nodes
       for (SparkPlanInfo child : children) {
         SparkPlanInfoForStage planForStage =
-            computeStageInfoForStage(child, accumulators, stageId, parentStageIds, false);
+            computeStageInfoForStage(child, accumulatorToStageID, stageId, parentStageIds, false);
 
         if (planForStage != null) {
           // Early stopping if the stage was found, no need to keep searching
@@ -88,14 +88,14 @@ public static SparkPlanInfoForStage computeStageInfoForStage(
     return null;
   }
 
-  private static Set<Integer> stageIdsForPlan(SparkPlanInfo info, Map<Long, Integer> accumulators) {
+  private static Set<Integer> stageIdsForPlan(SparkPlanInfo info, Map<Long, Integer> accumulatorToStageID) {
     Set<Integer> stageIds = new HashSet<>();
 
     Collection<SQLMetricInfo> metrics =
         AbstractDatadogSparkListener.listener.getPlanInfoMetrics(info);
     for (SQLMetricInfo metric : metrics) {
       // Using the accumulators to associate a plan with its stage
-      Integer stageId = accumulators.get(metric.accumulatorId());
+      Integer stageId = accumulatorToStageID.get(metric.accumulatorId());
 
       if (stageId != null) {
         stageIds.add(stageId);