Fix concurrent span event recording in OTel shim

dd-java-agent/agent-otel/otel-shim/src/main/java/datadog/opentelemetry/shim/trace/OtelSpan.java

@@ -37,8 +37,14 @@ public class OtelSpan implements Span, WithAgentSpan, SpanWrapper {
   private StatusCode statusCode;
   private boolean recording;
 
-  /** Span events ({@code null} until an event is added). */
-  private List<OtelSpanEvent> events;
+  /**
+   * Span events ({@code null} until an event is added). Mutations are guarded by synchronizing on
+   * {@code this}: events can be recorded from application threads while the span is finished (and
+   * its events serialized) from a different thread, so unsynchronized access would corrupt the
+   * backing list. Declared {@code volatile} so {@link #onSpanFinished()} can take a lock-free fast
+   * path for the common case of a span with no events.
+   */
+  private volatile List<OtelSpanEvent> events;
 
   public OtelSpan(AgentSpan delegate) {
     this.delegate = delegate;
@@ -85,25 +91,26 @@ public <T> Span setAttribute(AttributeKey<T> key, T value) {
   @Override
   public Span addEvent(String name, Attributes attributes) {
     if (this.recording) {
-      if (this.events == null) {
-        this.events = new ArrayList<>();
-      }
-      this.events.add(new OtelSpanEvent(name, attributes));
+      addEvent(new OtelSpanEvent(name, attributes));
     }
     return this;
   }
 
   @Override
   public Span addEvent(String name, Attributes attributes, long timestamp, TimeUnit unit) {
     if (this.recording) {
-      if (this.events == null) {
-        this.events = new ArrayList<>();
-      }
-      this.events.add(new OtelSpanEvent(name, attributes, timestamp, unit));
+      addEvent(new OtelSpanEvent(name, attributes, timestamp, unit));
     }
     return this;
   }
 
+  private synchronized void addEvent(OtelSpanEvent event) {
+    if (this.events == null) {
+      this.events = new ArrayList<>();
+    }
+    this.events.add(event);
+  }
+
   @Override
   public Span setStatus(StatusCode statusCode, String description) {
     if (this.recording) {
@@ -123,12 +130,9 @@ public Span setStatus(StatusCode statusCode, String description) {
   @Override
   public Span recordException(Throwable exception, Attributes additionalAttributes) {
     if (this.recording) {
-      if (this.events == null) {
-        this.events = new ArrayList<>();
-      }
       additionalAttributes = initializeExceptionAttributes(exception, additionalAttributes);
       applySpanEventExceptionAttributesAsTags(this.delegate, additionalAttributes);
-      this.events.add(new OtelSpanEvent(EXCEPTION_SPAN_EVENT_NAME, additionalAttributes));
+      addEvent(new OtelSpanEvent(EXCEPTION_SPAN_EVENT_NAME, additionalAttributes));
     }
     return this;
   }
@@ -179,7 +183,17 @@ public AgentSpan asAgentSpan() {
   @Override
   public void onSpanFinished() {
     applyNamingConvention(this.delegate);
-    setEventsAsTag(this.delegate, this.events);
+    // Fast path: most spans have no events, so avoid taking the lock entirely (single volatile
+    // read).
+    List<OtelSpanEvent> eventsSnapshot = this.events;
+    if (eventsSnapshot != null) {
+      // Copy under lock so serialization iterates a private snapshot that cannot be mutated
+      // concurrently by an application thread still recording events.
+      synchronized (this) {
+        eventsSnapshot = new ArrayList<>(this.events);
+      }
+    }
+    setEventsAsTag(this.delegate, eventsSnapshot);
   }
 
   private static class NoopSpan implements Span {

dd-java-agent/instrumentation/opentelemetry/opentelemetry-1.4/src/test/java/opentelemetry14/OpenTelemetry14Test.java

@@ -67,9 +67,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
 import opentelemetry14.context.propagation.TextMap;
+import org.json.JSONArray;
 import org.json.JSONException;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -289,6 +291,52 @@ void testAddMultipleSpanEvents() {
                     tag(SPAN_EVENTS, isJson(expectedEventTag)))));
   }
 
+  @Test
+  void testConcurrentAddEvents() throws Exception {
+    // Regression test for the concurrent recording of span events. Events used to be stored in a
+    // plain ArrayList mutated without synchronization; recording events from multiple threads (as
+    // GraphQL DataLoaders do on virtual threads) corrupted the backing list, leaving null holes
+    // that
+    // triggered a NullPointerException in OtelSpanEvent.toTag when the span was finished. See trace
+    // b8b8e4edde47f90a92e832b63251a577.
+    int threadCount = 8;
+    int eventsPerThread = 2000;
+    Span span = this.otelTracer.spanBuilder("some-name").startSpan();
+
+    CountDownLatch start = new CountDownLatch(1);
+    List<Thread> threads = new ArrayList<>();
+    for (int t = 0; t < threadCount; t++) {
+      Thread thread =
+          new Thread(
+              () -> {
+                try {
+                  start.await();
+                } catch (InterruptedException e) {
+                  Thread.currentThread().interrupt();
+                  return;
+                }
+                for (int i = 0; i < eventsPerThread; i++) {
+                  span.addEvent("event");
+                }
+              });
+      thread.start();
+      threads.add(thread);
+    }
+
+    start.countDown();
+    for (Thread thread : threads) {
+      thread.join();
+    }
+    span.end();
+
+    writer.waitForTraces(1);
+    List<DDSpan> firstTrace = writer.firstTrace();
+    assertEquals(1, firstTrace.size());
+    Object eventsTag = firstTrace.get(0).getTags().get(SPAN_EVENTS);
+    JSONArray events = new JSONArray((String) eventsTag);
+    assertEquals(threadCount * eventsPerThread, events.length());
+  }
+
   @Test
   void testSimpleSpanLinks() {
     String traceId = "1234567890abcdef1234567890abcdef";