Fixes thread safety issue in Tag.tag

The current codebase carefully synchronizes all access to MutableSpan
where there can be concurrent use. We missed a spot where Tag.tag is not
synchronizing access. The only other place we don't guard is ScopedSpan,
which is documented explicitly as not thread safe.

Fixes #1384

Signed-off-by: Adrian Cole <adrian@tetrate.io>

brave/src/main/java/brave/Tag.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 The OpenZipkin Authors
+ * Copyright 2013-2024 The OpenZipkin Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
  * in compliance with the License. You may obtain a copy of the License at
@@ -174,7 +174,10 @@ final void tag(Object span, I input, @Nullable TraceContext context) {
     if (span instanceof SpanCustomizer) {
       ((SpanCustomizer) span).tag(key, value);
     } else if (span instanceof MutableSpan) {
-      ((MutableSpan) span).tag(key, value);
+      MutableSpan mSpan = (MutableSpan) span;
+      synchronized (mSpan) {
+        mSpan.tag(key, value);
+      }
     }
   }
 

brave/src/test/java/brave/TagTest.java

@@ -15,7 +15,11 @@
 
 import brave.handler.MutableSpan;
 import brave.propagation.TraceContext;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
+import java.util.stream.IntStream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -24,6 +28,7 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
@@ -266,6 +271,27 @@ class TagTest {
     assertThat(mutableSpan).isEqualTo(expected);
   }
 
+  @Test
+  public void tag_mutableSpan_threadSafe() throws InterruptedException {
+    int numThreads = 1000;
+    ExecutorService service = Executors.newFixedThreadPool(numThreads);
+    try {
+      for (int i = 0; i < numThreads; i++) {
+        String val = String.valueOf(i);
+        Tag<Object> tag = new Tag<Object>("key"+i) {
+          @Override protected String parseValue(Object input, TraceContext context) {
+            return val;
+          }
+        };
+        service.submit(() -> tag.tag(input, context, mutableSpan));
+      }
+    } finally {
+      service.shutdown();
+      service.awaitTermination(1, TimeUnit.MINUTES);
+    }
+    assertThat(mutableSpan.tagCount()).isEqualTo(numThreads);
+  }
+
   @Test void tag_mutableSpan_nullContext() {
     when(parseValue.apply(eq(input), isNull())).thenReturn("value");