[DEV-76] Fix ClientTelemetry injection logic forcing JSON content type (#281)

* Fix ClientTelemetry injection logic forcing JSON content type

* Revert temp change to speed up testing

* Improve name of const

* Revert whitespace changes

Author: josephcummings

db-client-java/build.gradle

@@ -179,7 +179,7 @@ protobuf {
             // options.  Note the braces cannot be omitted, otherwise the
             // plugin will not be added. This is because of the implicit way
             // NamedDomainObjectContainer binds the methods.
-            grpc { }
+            grpc {}
         }
     }
 }

db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetry.java

@@ -12,6 +12,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.function.BiFunction;
@@ -29,11 +30,14 @@ private static Tracer getTracer() {
 
     private static List<EventData> tryInjectTracingContext(Span span, List<EventData> events) {
         List<EventData> injectedEvents = new ArrayList<>();
-        for (EventData event : events)
+        for (EventData event : events) {
+            boolean isJsonEvent = Objects.equals(event.getContentType(), ContentType.JSON);
+
             injectedEvents.add(EventDataBuilder
-                    .json(event.getEventId(), event.getEventType(), event.getEventData())
+                    .binary(event.getEventId(), event.getEventType(), event.getEventData(), isJsonEvent)
                     .metadataAsBytes(tryInjectTracingContext(span, event.getUserMetadata()))
                     .build());
+        }
         return injectedEvents;
     }
 

db-client-java/src/main/java/com/eventstore/dbclient/ContentType.java

@@ -0,0 +1,6 @@
+package com.eventstore.dbclient;
+
+class ContentType {
+    public static final String JSON = "application/json";
+    public static final String BYTES = "application/octet-stream";
+}

db-client-java/src/main/java/com/eventstore/dbclient/EventDataBuilder.java

@@ -55,6 +55,7 @@ public static <A> EventDataBuilder json(UUID id, String eventType, A eventData)
     public static EventDataBuilder json(String eventType, byte[] eventData) {
         return json(null, eventType, eventData);
     }
+
     /**
      * Configures an event data builder to host a JSON payload.
      * @param id event's id.
@@ -63,13 +64,7 @@ public static EventDataBuilder json(String eventType, byte[] eventData) {
      * @return an event data builder.
      */
     public static EventDataBuilder json(UUID id, String eventType, byte[] eventData) {
-        EventDataBuilder self = new EventDataBuilder();
-        self.eventData = eventData;
-        self.eventType = eventType;
-        self.isJson = true;
-        self.id = id;
-
-        return self;
+        return binary(id, eventType, eventData, true);
     }
 
     /**
@@ -90,11 +85,23 @@ public static EventDataBuilder binary(String eventType, byte[] eventData) {
      * @return an event data builder.
      */
     public static EventDataBuilder binary(UUID id, String eventType, byte[] eventData) {
+        return binary(id, eventType, eventData, false);
+    }
+
+    /**
+     * Configures an event data builder to host a binary payload.
+     * @param id event's id.
+     * @param eventType event's type.
+     * @param eventData event's payload.
+     * @param isJson whether the payload is JSON or not.
+     * @return an event data builder.
+     */
+    public static EventDataBuilder binary(UUID id, String eventType, byte[] eventData, boolean isJson) {
         EventDataBuilder self = new EventDataBuilder();
 
         self.eventData = eventData;
         self.eventType = eventType;
-        self.isJson = false;
+        self.isJson = isJson;
         self.id = id;
 
         return self;
@@ -134,11 +141,12 @@ public EventDataBuilder metadataAsBytes(byte[] value) {
 
     /**
      * Builds an event ready to be sent to EventStoreDB.
+     *
      * @see EventData
      */
     public EventData build() {
         UUID eventId = this.id == null ? UUID.randomUUID() : this.id;
-        String contentType = this.isJson ? "application/json" : "application/octet-stream";
+        String contentType = this.isJson ? ContentType.JSON : ContentType.BYTES;
         return new EventData(eventId, this.eventType, contentType, this.eventData, this.metadata);
     }
 }

db-client-java/src/main/java/com/eventstore/dbclient/WorkItemArgs.java

@@ -49,7 +49,7 @@ public boolean supportFeature(int feature) {
     public <A> HttpURLConnection getHttpConnection(OptionsBase<A> options, EventStoreDBClientSettings settings, String path) {
         try {
             HttpURLConnection conn = (HttpURLConnection) getURL(settings.isTls(), this.endpoint, path).openConnection();
-            conn.setRequestProperty("Accept", "application/json");
+            conn.setRequestProperty("Accept", ContentType.JSON);
             String creds = options.getHttpCredentialString();
 
             if (creds == null && settings.getDefaultCredentials() != null) {

db-client-java/src/test/java/com/eventstore/dbclient/TelemetryTests.java

@@ -3,6 +3,7 @@
 import com.eventstore.dbclient.telemetry.PersistentSubscriptionsTracingInstrumentationTests;
 import com.eventstore.dbclient.telemetry.SpanProcessorSpy;
 import com.eventstore.dbclient.telemetry.StreamsTracingInstrumentationTests;
+import com.eventstore.dbclient.telemetry.TracingContextInjectionTests;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
@@ -20,7 +21,7 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
-public class TelemetryTests implements StreamsTracingInstrumentationTests, PersistentSubscriptionsTracingInstrumentationTests {
+public class TelemetryTests implements StreamsTracingInstrumentationTests, PersistentSubscriptionsTracingInstrumentationTests, TracingContextInjectionTests {
     static private Database database;
     static private Logger logger;
 

db-client-java/src/test/java/com/eventstore/dbclient/telemetry/TracingContextInjectionTests.java

@@ -0,0 +1,69 @@
+package com.eventstore.dbclient.telemetry;
+
+import com.eventstore.dbclient.*;
+import org.junit.Assert;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public interface TracingContextInjectionTests extends TelemetryAware {
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testTracingContextInjectionDoesNotAffectEventBody() throws Throwable {
+        EventStoreDBClient streamsClient = getDefaultClient();
+        EventStoreDBPersistentSubscriptionsClient psClient = getDefaultPersistentSubscriptionClient();
+
+        String streamName = generateName();
+        String groupName = "aGroup";
+
+        EventData[] events = {
+                EventData.builderAsJson("JsonEvent", mapper.writeValueAsBytes(new Foo()))
+                        .eventId(UUID.randomUUID())
+                        .build(),
+                EventData.builderAsBinary("ProtoEvent", mapper.writeValueAsBytes(new Foo()))
+                        .eventId(UUID.randomUUID())
+                        .build()
+        };
+
+        Exceptions exceptions = new Exceptions().registerGoAwayError();
+        flaky(10, exceptions, () -> psClient.createToStream(streamName, groupName).get());
+
+        streamsClient.appendToStream(streamName, events).get();
+
+        CountDownLatch subscribeSpansLatch = new CountDownLatch(events.length);
+        onOperationSpanEnded(ClientTelemetryConstants.Operations.SUBSCRIBE, span -> subscribeSpansLatch.countDown());
+
+        ArrayList<RecordedEvent> receivedEvents = new ArrayList<>();
+        PersistentSubscription subscription = psClient.subscribeToStream(
+                streamName,
+                groupName,
+                SubscribePersistentSubscriptionOptions.get().bufferSize(32),
+                new PersistentSubscriptionListener() {
+                    @Override
+                    public void onEvent(PersistentSubscription subscription, int retryCount, ResolvedEvent event) {
+                        receivedEvents.add(event.getEvent());
+                    }
+                }
+        ).get();
+
+        subscribeSpansLatch.await();
+        subscription.stop();
+
+        for (RecordedEvent receivedEvent : receivedEvents) {
+            EventData sentEvent = Arrays.stream(events)
+                    .filter(e -> e.getEventId().equals(receivedEvent.getEventId()))
+                    .findFirst()
+                    .orElse(null);
+
+            Assertions.assertNotNull(sentEvent);
+            Assertions.assertArrayEquals(sentEvent.getEventData(), receivedEvent.getEventData());
+            Assertions.assertEquals(sentEvent.getContentType(), receivedEvent.getContentType());
+        }
+    }
+}