Avoid ArrayList copying from TraceInterceptors

dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java

@@ -1202,7 +1202,7 @@ public AgentDataStreamsMonitoring getDataStreamsMonitoring() {
    *
    * @param trace a list of the spans related to the same trace
    */
-  void write(final List<DDSpan> trace) {
+  void write(final TraceList trace) {
     if (trace.isEmpty() || !trace.get(0).traceConfig().isTraceEnabled()) {
       return;
     }
@@ -1239,9 +1239,22 @@ void write(final List<DDSpan> trace) {
     }
   }
 
-  private List<DDSpan> interceptCompleteTrace(List<DDSpan> trace) {
-    if (!interceptors.isEmpty() && !trace.isEmpty()) {
-      Collection<? extends MutableSpan> interceptedTrace = new ArrayList<>(trace);
+  private List<DDSpan> interceptCompleteTrace(TraceList originalTrace) {
+    if (!interceptors.isEmpty() && !originalTrace.isEmpty()) {
+      // Using TraceList to optimize the common case where the interceptors,
+      // don't alter the list.  If the interceptors just return the provided
+      // List, then no need to copy to another List.
+
+      // As an extra precaution, also check the modCount before and after on
+      // the TraceList, since TraceInterceptor could put some other type of
+      // object into the List.
+
+      // There is still a risk that a TraceInterceptor holds onto the provided
+      // List and modifies it later on, but we cannot safeguard against 
+      // every possible misuse.
+      Collection<? extends MutableSpan> interceptedTrace = originalTrace;
+      int originalModCount = originalTrace.modCount();
+
       for (final TraceInterceptor interceptor : interceptors) {
         try {
           // If one TraceInterceptor throws an exception, then continue with the next one
@@ -1255,14 +1268,20 @@ private List<DDSpan> interceptCompleteTrace(List<DDSpan> trace) {
         }
       }
 
-      trace = new ArrayList<>(interceptedTrace.size());
-      for (final MutableSpan span : interceptedTrace) {
-        if (span instanceof DDSpan) {
-          trace.add((DDSpan) span);
+      if (interceptedTrace != originalTrace || originalTrace.modCount() != originalModCount) {
+        TraceList trace = new TraceList(interceptedTrace.size());
+        for (final MutableSpan span : interceptedTrace) {
+          if (span instanceof DDSpan) {
+            trace.add((DDSpan) span);
+          }
         }
+        return trace;
+      } else {
+        return originalTrace;
       }
+    } else {
+      return originalTrace;
     }
-    return trace;
   }
 
   @Override

dd-trace-core/src/main/java/datadog/trace/core/PendingTrace.java

@@ -327,7 +327,7 @@ private int write(boolean isPartial) {
     if (!spans.isEmpty()) {
       try (Recording recording = tracer.writeTimer()) {
         // Only one writer at a time
-        final List<DDSpan> trace;
+        final TraceList trace;
         int completedSpans = 0;
         synchronized (this) {
           if (!isPartial) {
@@ -346,10 +346,10 @@ private int write(boolean isPartial) {
           // count(s) will be incremented, and any new spans added during the period that the count
           // was negative will be written by someone even if we don't write them right now.
           if (size > 0 && (!isPartial || size >= tracer.getPartialFlushMinSpans())) {
-            trace = new ArrayList<>(size);
+            trace = new TraceList(size);
             completedSpans = enqueueSpansToWrite(trace, writeRunningSpans);
           } else {
-            trace = EMPTY;
+            trace = TraceList.EMPTY;
           }
         }
         if (!trace.isEmpty()) {

dd-trace-core/src/main/java/datadog/trace/core/StreamingTraceCollector.java

@@ -4,7 +4,6 @@
 import datadog.trace.api.time.TimeSource;
 import datadog.trace.bootstrap.instrumentation.api.AgentScope;
 import datadog.trace.core.monitor.HealthMetrics;
-import java.util.Collections;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import javax.annotation.Nonnull;
 
@@ -72,7 +71,7 @@ PublishState onPublish(DDSpan span) {
       tracer.onRootSpanPublished(rootSpan);
     }
     healthMetrics.onFinishSpan();
-    tracer.write(Collections.singletonList(span));
+    tracer.write(TraceList.of(span));
     return PublishState.WRITTEN;
   }
 

dd-trace-core/src/main/java/datadog/trace/core/TraceList.java

@@ -0,0 +1,22 @@
+package datadog.trace.core;
+
+import java.util.ArrayList;
+
+/** ArrayList that exposes modCount to allow for an optimization in TraceInterceptor handling */
+final class TraceList extends ArrayList<DDSpan> {
+  static final TraceList EMPTY = new TraceList(0);
+
+  static final TraceList of(DDSpan span) {
+    TraceList list = new TraceList(1);
+    list.add(span);
+    return list;
+  }
+
+  TraceList(int capacity) {
+    super(capacity);
+  }
+
+  int modCount() {
+    return this.modCount;
+  }
+}