Thread safety (#383)

* Thread safety fixes

Author: atshaw43

aws-xray-recorder-sdk-core/src/main/java/com/amazonaws/xray/entities/SubsegmentImpl.java

@@ -20,8 +20,9 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import java.util.HashSet;
+import java.util.Collections;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -64,7 +65,7 @@ public SubsegmentImpl(AWSXRayRecorder creator,
         super(creator, name);
         this.parentSegment = parentSegment;
         parentSegment.incrementReferenceCount();
-        this.precursorIds = new HashSet<>();
+        this.precursorIds = Collections.newSetFromMap(new ConcurrentHashMap<>());
         this.shouldPropagate = true;
 
         this.isSampled = samplingStrategyOverride == SamplingStrategyOverride.DISABLED ?

aws-xray-recorder-sdk-core/src/test/java/com/amazonaws/xray/ConcurrencyTest.java

@@ -16,7 +16,13 @@
 package com.amazonaws.xray;
 
 import com.amazonaws.xray.emitters.Emitter;
+import com.amazonaws.xray.entities.Segment;
+import com.amazonaws.xray.entities.SegmentImpl;
+import com.amazonaws.xray.entities.SubsegmentImpl;
 import com.amazonaws.xray.strategy.sampling.LocalizedSamplingStrategy;
+import java.util.Iterator;
+import java.util.Random;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -64,4 +70,31 @@ void testManyThreads() throws InterruptedException {
         }
     }
 
+    @Test
+    public void testPrecursorIdConcurrency() throws Exception {
+        Segment seg = new SegmentImpl(AWSXRay.getGlobalRecorder(), "test");
+        SubsegmentImpl subseg = new SubsegmentImpl(AWSXRay.getGlobalRecorder(), "test", seg);
+
+        final long startTime = System.currentTimeMillis();
+
+        Thread thread1 = new Thread(() -> {
+            while (true) {
+                subseg.addPrecursorId("ID" + new Random().nextInt());
+            }
+        });
+        thread1.start();
+
+        Callable<Void> callable = (Callable) () -> {
+            while (System.currentTimeMillis() - startTime < TimeUnit.SECONDS.toMillis(1)) {
+                Iterator it = subseg.getPrecursorIds().iterator();
+                while (it.hasNext()) {
+                    it.next();
+                }
+            }
+            return null;
+        };
+        callable.call();
+
+        thread1.join();
+    }
 }

aws-xray-recorder-sdk-core/src/test/java/com/amazonaws/xray/entities/EntityTest.java

@@ -76,6 +76,18 @@ void testUnknownClassSerialization() {
         seg.serialize();  // Verify we don't crash here
     }
 
+    @Test
+    void testPrecursorIds() {
+        Segment seg = new SegmentImpl(AWSXRay.getGlobalRecorder(), "test");
+        SubsegmentImpl subseg = new SubsegmentImpl(AWSXRay.getGlobalRecorder(), "test", seg);
+        subseg.addPrecursorId("myId1");
+        subseg.addPrecursorId("myId2");
+        String serializedSubSeg = subseg.serialize();
+
+        String expected = "\"precursor_ids\":[\"myId1\",\"myId2\"]";
+        assertThat(serializedSubSeg).contains(expected);
+    }
+
     static class EmptyBean {
         String otherField = "cerealization";
     }