Avoid consuming ByteBuffers (#913)

A ByteBuffer is a pointer to a byte[] with a starting position, a current position, and a limit. Any function that reads from its contents updates the current position. Both TracingPropagator and WorkflowUtils copy the entirety of its contents, and in doing so they mutate the current position. WorkflowUtils resets it afterwards but this still isn't thread-safe as another thread may be trying to read it.

By duplicating the ByteBuffer (copying only the metadata, not the actual contents) we avoid modifying it. It doesn't seem likely that there's real impact in either of these cases beyond unit tests, where these ByteBuffers stick around in the workflow history and are repeatedly serialized/deserialized. Modifying them during serialization can create test flakiness as that can trigger exceptions.

Author: natemort

src/main/java/com/uber/cadence/internal/tracing/TracingPropagator.java

@@ -161,7 +161,7 @@ private SpanContext extract(Header header) {
                         Map.Entry::getKey,
                         e -> {
                           byte[] bytes = new byte[e.getValue().remaining()];
-                          e.getValue().get(bytes);
+                          e.getValue().duplicate().get(bytes);
                           return new String(bytes);
                         }))));
   }

src/main/java/com/uber/cadence/workflow/WorkflowUtils.java

@@ -35,11 +35,9 @@ public static <T> T getValueFromSearchAttributes(
   }
 
   private static byte[] getValueBytes(SearchAttributes searchAttributes, String key) {
-    ByteBuffer byteBuffer = searchAttributes.getIndexedFields().get(key);
+    ByteBuffer byteBuffer = searchAttributes.getIndexedFields().get(key).duplicate();
     final byte[] valueBytes = new byte[byteBuffer.remaining()];
-    byteBuffer.mark();
     byteBuffer.get(valueBytes);
-    byteBuffer.reset();
     return valueBytes;
   }
 }

src/test/java/com/uber/cadence/internal/tracing/TracingPropagatorTest.java

@@ -0,0 +1,50 @@
+package com.uber.cadence.internal.tracing;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import com.google.common.collect.ImmutableMap;
+import com.uber.cadence.Header;
+import com.uber.cadence.PollForActivityTaskResponse;
+import io.opentracing.Span;
+import io.opentracing.mock.MockSpan;
+import io.opentracing.mock.MockTracer;
+import java.nio.ByteBuffer;
+import java.util.List;
+import org.junit.Test;
+
+public class TracingPropagatorTest {
+
+  private final MockTracer mockTracer = new MockTracer();
+  private final TracingPropagator propagator = new TracingPropagator(mockTracer);
+
+  @Test
+  public void testSpanForExecuteActivity_allowReusingHeaders() {
+    Header header =
+        new Header()
+            .setFields(
+                ImmutableMap.of(
+                    "traceid",
+                    ByteBuffer.wrap("100".getBytes()),
+                    "spanid",
+                    ByteBuffer.wrap("200".getBytes())));
+
+    Span span =
+        propagator.spanForExecuteActivity(
+            new PollForActivityTaskResponse().setHeader(header).setActivityId("id"));
+    span.finish();
+    Span span2 =
+        propagator.spanForExecuteActivity(
+            new PollForActivityTaskResponse().setHeader(header).setActivityId("id2"));
+    span2.finish();
+
+    for (MockSpan mockSpan : mockTracer.finishedSpans()) {
+      assertEquals("100", mockSpan.context().toTraceId());
+      List<MockSpan.Reference> references = mockSpan.references();
+      assertFalse(references.isEmpty());
+      MockSpan.Reference from = references.get(0);
+      assertEquals("200", from.getContext().toSpanId());
+      assertEquals("follows_from", from.getReferenceType());
+    }
+  }
+}