Repository: EventStore/EventStoreDB-Client-Java
Pull request #270 (2024-04-19 15:53:58)

DEV-344 - Tracing Instrumentation
Changed: Added tracing instrumentation of Append and Subscribe (Catchup and Persistent) operations using the Open Telemetry APIs.

linear[bot]: <p><a href="https://linear.app/eventstore/issue/DEV-344/tracing-instrumentation-java-client">DEV-344 Tracing Instrumentation Java Client</a></p>

Patch:
diff --git a/README.md b/README.md
index 5d7de08..f58c194 100644
--- a/README.md
+++ b/README.md
@@ -40,6 +40,12 @@ Tests are written using [TestContainers](https://www.testcontainers.org/) and re
 
 Specific docker images can be specified via the enviroment variable `EVENTSTORE_IMAGE`.
 
+## Open Telemetry
+
+ Tracing is the only telemetry currently exported, specifically for the `Append` and `Subscribe` (Catchup and Persistent) operations.
+
+ For more information about Open Telemetry, refer to the [official documentation](https://opentelemetry.io/docs/what-is-opentelemetry/).
+
 ## EventStoreDB Server Compatibility
 
 This client is compatible with version `20.6.1` upwards.
diff --git a/db-client-java/build.gradle b/db-client-java/build.gradle
index 4cd5a5e..91cbd3d 100644
--- a/db-client-java/build.gradle
+++ b/db-client-java/build.gradle
@@ -50,6 +50,11 @@ dependencies {
     implementation "org.bouncycastle:bcprov-jdk18on:1.77"
     implementation "org.bouncycastle:bcpkix-jdk18on:1.77"
 
+    implementation platform("io.opentelemetry:opentelemetry-bom:${openTelemetryVersion}")
+    implementation "io.opentelemetry:opentelemetry-api"
+    implementation "io.opentelemetry.semconv:opentelemetry-semconv:${openTelemetrySemConvVersion}"
+    implementation "io.opentelemetry.semconv:opentelemetry-semconv-incubating:${openTelemetrySemConvVersion}"
+
     testImplementation "org.junit.jupiter:junit-jupiter-api:${junitVersion}"
     testImplementation "org.junit.jupiter:junit-jupiter-params:${junitVersion}"
     testImplementation "org.junit.jupiter:junit-jupiter-engine:${junitVersion}"
@@ -62,6 +67,8 @@ dependencies {
     testImplementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
     testImplementation "com.github.javafaker:javafaker:1.0.2"
     testImplementation 'org.slf4j:slf4j-simple:2.0.9'
+    testImplementation "io.opentelemetry:opentelemetry-sdk"
+    testImplementation "io.opentelemetry:opentelemetry-sdk-testing"
 }
 
 tasks.withType(Test).configureEach {
@@ -106,6 +113,7 @@ tasks.register("singleNodeTests", Test) {
     useJUnitPlatform {
         include("**/StreamsTests.class")
         include("**/PersistentSubscriptionsTests.class")
+        include("**/TelemetryTests.class")
         include("**/ExpectationsTests.class")
     }
 }
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
index e0b47c9..3b0886f 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractRegularSubscription.java
@@ -47,7 +47,7 @@ abstract class AbstractRegularSubscription {
                     .setOptions(createOptions())
                     .build();
 
-            StreamsGrpc.StreamsStub client = GrpcUtils.configureStub(StreamsGrpc.newStub(channel), this.client.getSettings(), this.options);
+            StreamsGrpc.StreamsStub streamsClient = GrpcUtils.configureStub(StreamsGrpc.newStub(channel), this.client.getSettings(), this.options);
 
             CompletableFuture<Subscription> future = new CompletableFuture<>();
             ClientResponseObserver<StreamsOuterClass.ReadReq, StreamsOuterClass.ReadResp> observer = new ClientResponseObserver<StreamsOuterClass.ReadReq, StreamsOuterClass.ReadResp>() {
@@ -106,7 +106,14 @@ abstract class AbstractRegularSubscription {
                     }
 
                     try {
-                        listener.onEvent(this._subscription, ResolvedEvent.fromWire(readResp.getEvent()));
+                        ResolvedEvent resolvedEvent = ResolvedEvent.fromWire(readResp.getEvent());
+                        ClientTelemetry.traceSubscribe(
+                                () -> listener.onEvent(this._subscription, resolvedEvent),
+                                _subscription.getSubscriptionId(),
+                                channel,
+                                client.getSettings(),
+                                options.getCredentials(),
+                                resolvedEvent.getOriginalEvent());
                     } catch (Exception e) {
                         onError(e);
                     }
@@ -144,7 +151,7 @@ abstract class AbstractRegularSubscription {
                 }
             };
 
-            client.read(readReq, observer);
+            streamsClient.read(readReq, observer);
 
             return future;
         });
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java
index 8df5f63..8ae4450 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/AbstractSubscribePersistentSubscription.java
@@ -38,7 +38,7 @@ abstract class AbstractSubscribePersistentSubscription {
 
     public CompletableFuture<PersistentSubscription> execute() {
         return this.client.runWithArgs(args -> {
-            PersistentSubscriptionsGrpc.PersistentSubscriptionsStub client =
+            PersistentSubscriptionsGrpc.PersistentSubscriptionsStub persistentSubscriptionsClient =
                     GrpcUtils.configureStub(PersistentSubscriptionsGrpc.newStub(args.getChannel()), this.client.getSettings(), this.options);
 
             final CompletableFuture<PersistentSubscription> result = new CompletableFuture<>();
@@ -91,7 +91,14 @@ abstract class AbstractSubscribePersistentSubscription {
                         int retryCount = readResp.getEvent().hasNoRetryCount() ? 0 : readResp.getEvent().getRetryCount();
 
                         try {
-                            listener.onEvent(this._subscription, retryCount, ResolvedEvent.fromWire(readResp.getEvent()));
+                            ResolvedEvent resolvedEvent = ResolvedEvent.fromWire(readResp.getEvent());
+                            ClientTelemetry.traceSubscribe(
+                                    () -> listener.onEvent(this._subscription, retryCount, resolvedEvent),
+                                    _subscription.getSubscriptionId(),
+                                    args.getChannel(),
+                                    client.getSettings(),
+                                    options.getCredentials(),
+                                    resolvedEvent.getOriginalEvent());
                         } catch (Exception e) {
                             onError(e);
                         }
@@ -129,7 +136,7 @@ abstract class AbstractSubscribePersistentSubscription {
                     }
                 };
 
-                StreamObserver<Persistent.ReadReq> wireStream = client.read(observer);
+                StreamObserver<Persistent.ReadReq> wireStream = persistentSubscriptionsClient.read(observer);
                 wireStream.onNext(req);
             }
 
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/AppendToStream.java b/db-client-java/src/main/java/com/eventstore/dbclient/AppendToStream.java
index 1bcae90..2f3a814 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/AppendToStream.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/AppendToStream.java
@@ -4,6 +4,7 @@ import com.eventstore.dbclient.proto.shared.Shared;
 import com.eventstore.dbclient.proto.streams.StreamsGrpc;
 import com.eventstore.dbclient.proto.streams.StreamsOuterClass;
 import com.google.protobuf.ByteString;
+import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
@@ -30,93 +31,101 @@ class AppendToStream {
     }
 
     public CompletableFuture<WriteResult> execute() {
-        return this.client.run(channel -> {
-            CompletableFuture<WriteResult> result = new CompletableFuture<>();
-            StreamsOuterClass.AppendReq.Options.Builder options = this.options.getExpectedRevision().applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
-                    .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
-                            .setStreamName(ByteString.copyFromUtf8(streamName))
-                            .build()));
-            StreamsGrpc.StreamsStub client = GrpcUtils.configureStub(StreamsGrpc.newStub(channel), this.client.getSettings(), this.options);
-
-            StreamObserver<StreamsOuterClass.AppendReq> requestStream = client.append(GrpcUtils.convertSingleResponse(result, resp -> {
-                if (resp.hasSuccess()) {
-                    StreamsOuterClass.AppendResp.Success success = resp.getSuccess();
-
-                    Position logPosition = null;
-                    if (success.getPositionOptionCase() == StreamsOuterClass.AppendResp.Success.PositionOptionCase.POSITION) {
-                        StreamsOuterClass.AppendResp.Position p = success.getPosition();
-                        logPosition = new Position(p.getCommitPosition(), p.getPreparePosition());
-                    }
-
-                    ExpectedRevision nextExpectedRevision = success.hasNoStream() ? ExpectedRevision.noStream()
-                            : ExpectedRevision.expectedRevision(success.getCurrentRevision());
-
-                    return new WriteResult(nextExpectedRevision, logPosition);
-                }
-                if (resp.hasWrongExpectedVersion()) {
-                    StreamsOuterClass.AppendResp.WrongExpectedVersion wev = resp.getWrongExpectedVersion();
-
-                    ExpectedRevision expectedRevision;
-                    if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_ANY) {
-                        expectedRevision = ExpectedRevision.any();
-                    } else if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_STREAM_EXISTS) {
-                        expectedRevision = ExpectedRevision.streamExists();
-                    } else {
-                        expectedRevision = ExpectedRevision.expectedRevision(wev.getExpectedRevision());
-                    }
-
-                    ExpectedRevision currentRevision;
-                    if (wev.getCurrentRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.CurrentRevisionOptionCase.CURRENT_NO_STREAM) {
-                        currentRevision = ExpectedRevision.noStream();
-                    } else {
-                        currentRevision = ExpectedRevision.expectedRevision(wev.getCurrentRevision());
-                    }
-
-                    String streamName = options.getStreamIdentifier().getStreamName().toStringUtf8();
-
-                    throw new WrongExpectedVersionException(streamName, expectedRevision, currentRevision);
+        return this.client.run(channel -> ClientTelemetry.traceAppend(
+                this::append,
+                channel,
+                events,
+                this.streamName,
+                this.client.getSettings(),
+                this.options.getCredentials()));
+    }
+
+    private CompletableFuture<WriteResult> append(ManagedChannel channel, List<EventData> events) {
+        CompletableFuture<WriteResult> result = new CompletableFuture<>();
+        StreamsOuterClass.AppendReq.Options.Builder options = this.options.getExpectedRevision().applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
+                .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
+                        .setStreamName(ByteString.copyFromUtf8(streamName))
+                        .build()));
+        StreamsGrpc.StreamsStub client = GrpcUtils.configureStub(StreamsGrpc.newStub(channel), this.client.getSettings(), this.options);
+
+        StreamObserver<StreamsOuterClass.AppendReq> requestStream = client.append(GrpcUtils.convertSingleResponse(result, resp -> {
+            if (resp.hasSuccess()) {
+                StreamsOuterClass.AppendResp.Success success = resp.getSuccess();
+
+                Position logPosition = null;
+                if (success.getPositionOptionCase() == StreamsOuterClass.AppendResp.Success.PositionOptionCase.POSITION) {
+                    StreamsOuterClass.AppendResp.Position p = success.getPosition();
+                    logPosition = new Position(p.getCommitPosition(), p.getPreparePosition());
                 }
 
-                throw new IllegalStateException("AppendResponse has neither Success or WrongExpectedVersion variants");
-            }));
-
-            try {
-                requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder().setOptions(options).build());
-
-                for (EventData e : this.events) {
-                    StreamsOuterClass.AppendReq.ProposedMessage.Builder msgBuilder = StreamsOuterClass.AppendReq.ProposedMessage.newBuilder()
-                            .setId(Shared.UUID.newBuilder()
-                                    .setStructured(Shared.UUID.Structured.newBuilder()
-                                            .setMostSignificantBits(e.getEventId().getMostSignificantBits())
-                                            .setLeastSignificantBits(e.getEventId().getLeastSignificantBits())))
-                            .setData(ByteString.copyFrom(e.getEventData()))
-                            .putMetadata(SystemMetadataKeys.CONTENT_TYPE, e.getContentType())
-                            .putMetadata(SystemMetadataKeys.TYPE, e.getEventType());
-
-                    if (e.getUserMetadata() != null) {
-                        msgBuilder.setCustomMetadata(ByteString.copyFrom(e.getUserMetadata()));
-                    }
-
-                    requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder()
-                            .setProposedMessage(msgBuilder)
-                            .build());
+                ExpectedRevision nextExpectedRevision = success.hasNoStream() ? ExpectedRevision.noStream()
+                        : ExpectedRevision.expectedRevision(success.getCurrentRevision());
+
+                return new WriteResult(nextExpectedRevision, logPosition);
+            }
+            if (resp.hasWrongExpectedVersion()) {
+                StreamsOuterClass.AppendResp.WrongExpectedVersion wev = resp.getWrongExpectedVersion();
+
+                ExpectedRevision expectedRevision;
+                if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_ANY) {
+                    expectedRevision = ExpectedRevision.any();
+                } else if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_STREAM_EXISTS) {
+                    expectedRevision = ExpectedRevision.streamExists();
+                } else {
+                    expectedRevision = ExpectedRevision.expectedRevision(wev.getExpectedRevision());
                 }
-                requestStream.onCompleted();
-            } catch (StatusRuntimeException e) {
-                String leaderHost = e.getTrailers().get(Metadata.Key.of("leader-endpoint-host", Metadata.ASCII_STRING_MARSHALLER));
-                String leaderPort = e.getTrailers().get(Metadata.Key.of("leader-endpoint-port", Metadata.ASCII_STRING_MARSHALLER));
-
-                if (leaderHost != null && leaderPort != null) {
-                    NotLeaderException reason = new NotLeaderException(leaderHost, Integer.valueOf(leaderPort));
-                    result.completeExceptionally(reason);
+
+                ExpectedRevision currentRevision;
+                if (wev.getCurrentRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.CurrentRevisionOptionCase.CURRENT_NO_STREAM) {
+                    currentRevision = ExpectedRevision.noStream();
                 } else {
-                    result.completeExceptionally(e);
+                    currentRevision = ExpectedRevision.expectedRevision(wev.getCurrentRevision());
+                }
+
+                String streamName = options.getStreamIdentifier().getStreamName().toStringUtf8();
+
+                throw new WrongExpectedVersionException(streamName, expectedRevision, currentRevision);
+            }
+
+            throw new IllegalStateException("AppendResponse has neither Success or WrongExpectedVersion variants");
+        }));
+
+        try {
+            requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder().setOptions(options).build());
+
+            for (EventData e : events) {
+                StreamsOuterClass.AppendReq.ProposedMessage.Builder msgBuilder = StreamsOuterClass.AppendReq.ProposedMessage.newBuilder()
+                        .setId(Shared.UUID.newBuilder()
+                                .setStructured(Shared.UUID.Structured.newBuilder()
+                                        .setMostSignificantBits(e.getEventId().getMostSignificantBits())
+                                        .setLeastSignificantBits(e.getEventId().getLeastSignificantBits())))
+                        .setData(ByteString.copyFrom(e.getEventData()))
+                        .putMetadata(SystemMetadataKeys.CONTENT_TYPE, e.getContentType())
+                        .putMetadata(SystemMetadataKeys.TYPE, e.getEventType());
+
+                if (e.getUserMetadata() != null) {
+                    msgBuilder.setCustomMetadata(ByteString.copyFrom(e.getUserMetadata()));
                 }
-            } catch (RuntimeException e) {
+
+                requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder()
+                        .setProposedMessage(msgBuilder)
+                        .build());
+            }
+            requestStream.onCompleted();
+        } catch (StatusRuntimeException e) {
+            String leaderHost = e.getTrailers().get(Metadata.Key.of("leader-endpoint-host", Metadata.ASCII_STRING_MARSHALLER));
+            String leaderPort = e.getTrailers().get(Metadata.Key.of("leader-endpoint-port", Metadata.ASCII_STRING_MARSHALLER));
+
+            if (leaderHost != null && leaderPort != null) {
+                NotLeaderException reason = new NotLeaderException(leaderHost, Integer.valueOf(leaderPort));
+                result.completeExceptionally(reason);
+            } else {
                 result.completeExceptionally(e);
             }
+        } catch (RuntimeException e) {
+            result.completeExceptionally(e);
+        }
 
-            return result;
-        });
+        return result;
     }
 }
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetry.java b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetry.java
new file mode 100644
index 0000000..8b23ce8
--- /dev/null
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetry.java
@@ -0,0 +1,166 @@
+package com.eventstore.dbclient;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.grpc.ManagedChannel;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.*;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.function.BiFunction;
+
+class ClientTelemetry {
+    private static final ClientTelemetryTags DEFAULT_ATTRIBUTES = new ClientTelemetryTags() {{
+        put(ClientTelemetryAttributes.Database.SYSTEM, ClientTelemetryConstants.INSTRUMENTATION_NAME);
+    }};
+
+    private static Tracer getTracer() {
+        return GlobalOpenTelemetry.getTracer(
+                ClientTelemetry.class.getPackage().getName(),
+                ClientTelemetry.class.getPackage().getImplementationVersion());
+    }
+
+    private static List<EventData> tryInjectTracingContext(Span span, List<EventData> events) {
+        List<EventData> injectedEvents = new ArrayList<>();
+        for (EventData event : events)
+            injectedEvents.add(EventDataBuilder
+                    .json(event.getEventId(), event.getEventType(), event.getEventData())
+                    .metadataAsBytes(tryInjectTracingContext(span, event.getUserMetadata()))
+                    .build());
+        return injectedEvents;
+    }
+
+    private static byte[] tryInjectTracingContext(Span span, byte[] userMetadataBytes) {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            ObjectNode userMetadata = userMetadataBytes != null
+                    ? objectMapper.readValue(userMetadataBytes, ObjectNode.class)
+                    : objectMapper.createObjectNode();
+
+            userMetadata.put(ClientTelemetryConstants.Metadata.TRACE_ID, span.getSpanContext().getTraceId());
+            userMetadata.put(ClientTelemetryConstants.Metadata.SPAN_ID, span.getSpanContext().getSpanId());
+
+            return objectMapper.writeValueAsBytes(userMetadata);
+        } catch (Throwable t) {
+            // User metadata may not be a valid JSON object, or not JSON altogether.
+            return userMetadataBytes;
+        }
+    }
+
+    private static SpanContext tryExtractTracingContext(byte[] userMetadataBytes) {
+        if (userMetadataBytes == null)
+            return null;
+
+        try {
+            ObjectNode userMetadata = new ObjectMapper().readValue(userMetadataBytes, ObjectNode.class);
+
+            JsonNode traceIdNode = userMetadata.get(ClientTelemetryConstants.Metadata.TRACE_ID);
+            JsonNode spanIdNode = userMetadata.get(ClientTelemetryConstants.Metadata.SPAN_ID);
+
+            if (traceIdNode == null || spanIdNode == null)
+                return null;
+
+            String traceId = traceIdNode.asText();
+            String spanId = spanIdNode.asText();
+
+            if (!TraceId.isValid(traceId) || !SpanId.isValid(spanId))
+                return null;
+
+            return SpanContext.createFromRemoteParent(traceId, spanId, TraceFlags.getSampled(), TraceState.getDefault());
+        } catch (Throwable t) {
+            return null;
+        }
+    }
+
+    static CompletableFuture<WriteResult> traceAppend(BiFunction<ManagedChannel, List<EventData>, CompletableFuture<WriteResult>> appendOperation, ManagedChannel channel,
+                                                      List<EventData> events, String streamId, EventStoreDBClientSettings settings, UserCredentials optionalCallCredentials) {
+        Span span = createSpan(
+                ClientTelemetryConstants.Operations.APPEND,
+                SpanKind.CLIENT,
+                null,
+                ClientTelemetryTags.builder()
+                        .withRequiredTag(ClientTelemetryAttributes.EventStore.STREAM, streamId)
+                        .withServerTagsFromGrpcChannel(channel)
+                        .withServerTagsFromClientSettings(settings)
+                        .withOptionalDatabaseUserTag(settings.getDefaultCredentials())
+                        .withOptionalDatabaseUserTag(optionalCallCredentials)
+                        .build());
+
+        try (Scope scope = span.makeCurrent()) {
+            return appendOperation
+                    .apply(channel, tryInjectTracingContext(span, events))
+                    .handle(((writeResult, throwable) -> {
+                        if (throwable != null) {
+                            span.setStatus(StatusCode.ERROR);
+                            span.recordException(throwable);
+                            span.end();
+                            throw new CompletionException(throwable);
+                        } else {
+                            span.setStatus(StatusCode.OK);
+                            span.end();
+                            return writeResult;
+                        }
+                    }));
+        }
+    }
+
+    static void traceSubscribe(Runnable tracedOperation, String subscriptionId, ManagedChannel channel, EventStoreDBClientSettings settings,
+                               UserCredentials optionalCallCredentials, RecordedEvent event) {
+        SpanContext remoteParentContext = tryExtractTracingContext(event.getUserMetadata());
+        Span span = createSpan(
+                ClientTelemetryConstants.Operations.SUBSCRIBE,
+                SpanKind.CONSUMER,
+                remoteParentContext,
+                ClientTelemetryTags.builder()
+                        .withRequiredTag(ClientTelemetryAttributes.EventStore.STREAM, event.getStreamId())
+                        .withRequiredTag(ClientTelemetryAttributes.EventStore.SUBSCRIPTION_ID, subscriptionId)
+                        .withRequiredTag(ClientTelemetryAttributes.EventStore.EVENT_ID, event.getEventId().toString())
+                        .withRequiredTag(ClientTelemetryAttributes.EventStore.EVENT_TYPE, event.getEventType())
+                        .withServerTagsFromGrpcChannel(channel)
+                        .withServerTagsFromClientSettings(settings)
+                        .withOptionalDatabaseUserTag(settings.getDefaultCredentials())
+                        .withOptionalDatabaseUserTag(optionalCallCredentials)
+                        .build());
+
+        try (Scope scope = span.makeCurrent()) {
+            tracedOperation.run();
+            span.setStatus(StatusCode.OK);
+        } catch (Throwable t) {
+            span.recordException(t);
+            span.setStatus(StatusCode.ERROR);
+            throw t;
+        } finally {
+            span.end();
+        }
+    }
+
+    static Span createSpan(String operationName, SpanKind spanKind, SpanContext parentContext, ClientTelemetryTags customAttributes) {
+        SpanBuilder spanBuilder = getTracer().spanBuilder(operationName).setSpanKind(spanKind);
+
+        if (parentContext != null)
+            spanBuilder.setParent(Context.current().with(Span.wrap(parentContext)));
+
+        ClientTelemetryTags attributes = new ClientTelemetryTags(DEFAULT_ATTRIBUTES) {{
+            put(ClientTelemetryAttributes.Database.OPERATION, operationName);
+
+            if (customAttributes != null)
+                putAll(customAttributes);
+        }};
+
+        for (Map.Entry<String, String> entry : attributes.entrySet()) {
+            String value = entry.getValue();
+            if (value == null) continue;
+
+            spanBuilder.setAttribute(entry.getKey(), value);
+        }
+
+        return spanBuilder.startSpan();
+    }
+}
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryAttributes.java b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryAttributes.java
new file mode 100644
index 0000000..7fd4cc1
--- /dev/null
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryAttributes.java
@@ -0,0 +1,32 @@
+package com.eventstore.dbclient;
+
+import io.opentelemetry.semconv.ExceptionAttributes;
+import io.opentelemetry.semconv.OtelAttributes;
+import io.opentelemetry.semconv.ServerAttributes;
+import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
+
+public class ClientTelemetryAttributes {
+    public static class Database {
+        public static final String USER = DbIncubatingAttributes.DB_USER.getKey();
+        public static final String SYSTEM = DbIncubatingAttributes.DB_SYSTEM.getKey();
+        public static final String OPERATION = DbIncubatingAttributes.DB_OPERATION.getKey();
+    }
+
+    public static class Server {
+        public static final String ADDRESS = ServerAttributes.SERVER_ADDRESS.getKey();
+        public static final String PORT = ServerAttributes.SERVER_PORT.getKey();
+    }
+
+    public static class Exceptions {
+        public static final String TYPE = ExceptionAttributes.EXCEPTION_TYPE.getKey();
+        public static final String MESSAGE = ExceptionAttributes.EXCEPTION_MESSAGE.getKey();
+        public static final String STACK_TRACE = ExceptionAttributes.EXCEPTION_STACKTRACE.getKey();
+    }
+
+    public static class EventStore {
+        public static final String STREAM = "db.eventstoredb.stream";
+        public static final String SUBSCRIPTION_ID = "db.eventstoredb.subscription.id";
+        public static final String EVENT_ID = "db.eventstoredb.event.id";
+        public static final String EVENT_TYPE = "db.eventstoredb.event.type";
+    }
+}
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryConstants.java b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryConstants.java
new file mode 100644
index 0000000..e7de5ac
--- /dev/null
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryConstants.java
@@ -0,0 +1,15 @@
+package com.eventstore.dbclient;
+
+public class ClientTelemetryConstants {
+    public static final String INSTRUMENTATION_NAME = "eventstoredb";
+
+    public static class Metadata {
+        public static final String TRACE_ID = "$traceId";
+        public static final String SPAN_ID = "$spanId";
+    }
+
+    public static class Operations {
+        public static final String APPEND = "streams.append";
+        public static final String SUBSCRIBE = "streams.subscribe";
+    }
+}
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryTags.java b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryTags.java
new file mode 100644
index 0000000..76548d1
--- /dev/null
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/ClientTelemetryTags.java
@@ -0,0 +1,72 @@
+package com.eventstore.dbclient;
+
+import io.grpc.ManagedChannel;
+
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+
+class ClientTelemetryTags extends HashMap<String, String> {
+    ClientTelemetryTags() {
+        super();
+    }
+
+    ClientTelemetryTags(ClientTelemetryTags tags) {
+        super(tags);
+    }
+
+    static Builder builder() {
+        return new Builder();
+    }
+
+    static class Builder {
+        final ClientTelemetryTags tags;
+
+        Builder() {
+            tags = new ClientTelemetryTags();
+        }
+
+        Builder withServerTagsFromGrpcChannel(ManagedChannel channel) {
+            if (channel == null) return this;
+
+            String[] authorityParts = channel.authority().split(":");
+
+            return withServerTags(authorityParts[0], authorityParts[1]);
+        }
+
+        Builder withServerTagsFromClientSettings(EventStoreDBClientSettings settings) {
+            if (settings == null || !settings.isDnsDiscover()) return this;
+
+            InetSocketAddress dns = settings.getHosts()[0];
+
+            return withServerTags(dns.getAddress().toString(), String.valueOf(dns.getPort()));
+        }
+
+        private Builder withServerTags(String address, String port) {
+            return withRequiredTag(ClientTelemetryAttributes.Server.ADDRESS, address)
+                    .withRequiredTag(ClientTelemetryAttributes.Server.PORT, String.valueOf(port));
+        }
+
+        Builder withOptionalDatabaseUserTag(UserCredentials userCredentials) {
+            if (userCredentials == null) return this;
+
+            return withOptionalTag(ClientTelemetryAttributes.Database.USER, userCredentials.getUsername());
+        }
+
+        Builder withRequiredTag(String key, String value) {
+            if (key == null) throw new NullPointerException("Required tag key is null.");
+            if (value == null) throw new NullPointerException("Required tag value is null.");
+
+            return withOptionalTag(key, value);
+        }
+
+        Builder withOptionalTag(String key, String value) {
+            if (value == null) return this;
+            tags.put(key, value);
+            return this;
+        }
+
+        ClientTelemetryTags build() {
+            return tags;
+        }
+    }
+}
diff --git a/db-client-java/src/main/java/com/eventstore/dbclient/UserCredentials.java b/db-client-java/src/main/java/com/eventstore/dbclient/UserCredentials.java
index 7168803..3a2529e 100644
--- a/db-client-java/src/main/java/com/eventstore/dbclient/UserCredentials.java
+++ b/db-client-java/src/main/java/com/eventstore/dbclient/UserCredentials.java
@@ -7,13 +7,18 @@ import java.util.Base64;
  * Holds a login and a password for authenticated requests.
  */
 public final class UserCredentials {
+    private final String username;
     private final String base64Encoded;
 
     public UserCredentials(String username, String password) {
+        this.username = username;
+
         byte[] credentialsBytes = String.format("%s:%s", username, password).getBytes(StandardCharsets.US_ASCII);
         base64Encoded = String.format("Basic %s", Base64.getEncoder().encodeToString(credentialsBytes));
     }
 
+    String getUsername() { return username; }
+
     String basicAuthHeader() {
         return base64Encoded;
     }
diff --git a/gradle.properties b/gradle.properties
index 5ab1c3c..f962537 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -9,6 +9,8 @@ annotationApiVersion=1.3.2
 validationApiVersion=2.0.1.Final
 reactiveStreamsApiVersion=1.0.4
 junitVersion=5.10.0
+openTelemetryVersion=1.37.0
+openTelemetrySemConvVersion=1.25.0-alpha
 
 # Test Dependencies
 slf4jNopVersion=1.7.29


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 0ab02e6..7fd8e78 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -14,7 +14,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test: [Streams, PersistentSubscriptions, Expectations]
+        test: [Streams, PersistentSubscriptions, Expectations, Telemetry]
 
     runs-on: ubuntu-latest
     steps:
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/ConnectionAware.java b/db-client-java/src/test/java/com/eventstore/dbclient/ConnectionAware.java
index ee1773d..bce8c5b 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/ConnectionAware.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/ConnectionAware.java
@@ -3,18 +3,13 @@ package com.eventstore.dbclient;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.json.JsonMapper;
 import com.github.javafaker.Faker;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
 import org.slf4j.Logger;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ExecutionException;
-import java.util.function.Consumer;
-import java.util.function.Supplier;
 
 public interface ConnectionAware {
     Faker faker = new Faker();
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/TelemetryTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/TelemetryTests.java
new file mode 100644
index 0000000..0460370
--- /dev/null
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/TelemetryTests.java
@@ -0,0 +1,81 @@
+package com.eventstore.dbclient;
+
+import com.eventstore.dbclient.telemetry.PersistentSubscriptionsTracingInstrumentationTests;
+import com.eventstore.dbclient.telemetry.SpanProcessorSpy;
+import com.eventstore.dbclient.telemetry.StreamsTracingInstrumentationTests;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+public class TelemetryTests implements StreamsTracingInstrumentationTests, PersistentSubscriptionsTracingInstrumentationTests {
+    static private Database database;
+    static private Logger logger;
+
+    private final List<Consumer<ReadableSpan>> spanEndedHooks = new ArrayList<>();
+    private final List<ReadableSpan> recordedSpans = new ArrayList<>();
+
+    @BeforeAll
+    public static void setup() {
+        database = DatabaseFactory.spawn();
+        logger = LoggerFactory.getLogger(StreamsTests.class);
+    }
+
+    @BeforeEach
+    public void beforeEach() {
+        GlobalOpenTelemetry.resetForTest();
+        spanEndedHooks.add(recordedSpans::add);
+
+        OpenTelemetrySdk.builder()
+                .setTracerProvider(SdkTracerProvider
+                        .builder()
+                        .addSpanProcessor(new SpanProcessorSpy(spanEndedHooks))
+                        .build())
+                .buildAndRegisterGlobal();
+    }
+
+    @Override
+    public void onOperationSpanEnded(String operation, Consumer<ReadableSpan> onSpanEnded) {
+        spanEndedHooks.add(span -> {
+            if(Objects.equals(span.getAttribute(AttributeKey.stringKey(ClientTelemetryAttributes.Database.OPERATION)), operation))
+                onSpanEnded.accept(span);
+        });
+    }
+
+    @Override
+    public List<ReadableSpan> getSpansForOperation(String operation) {
+        return recordedSpans.stream()
+                .filter(span -> {
+                    String spanOperation = span.getAttribute(AttributeKey.stringKey(ClientTelemetryAttributes.Database.OPERATION));
+                    return spanOperation != null && spanOperation.equals(operation);
+                })
+                .collect(Collectors.toList());
+    }
+
+    @Override
+    public Database getDatabase() {
+        return database;
+    }
+
+    @Override
+    public Logger getLogger() {
+        return logger;
+    }
+
+    @AfterAll
+    public static void cleanup() {
+        database.dispose();
+    }
+}
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/expectations/ProjectionManagementTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/expectations/ProjectionManagementTests.java
index 55f956d..b53b5e7 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/expectations/ProjectionManagementTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/expectations/ProjectionManagementTests.java
@@ -27,10 +27,6 @@ public interface ProjectionManagementTests extends ConnectionAware {
     String COUNT_EVENTS_PARTITIONED_PROJECTION = loadResourceAsString(COUNT_EVENTS_PARTITIONED_PROJECTION_FILENAME);
     String UNKNOWN_KEYNAMES_PROJECTION = loadResourceAsString(UNKNOWN_KEYNAMES_PROJECTION_FILENAME);
 
-    default EventStoreDBProjectionManagementClient projectionClient() {
-        return EventStoreDBProjectionManagementClient.from(getDefaultClient());
-    }
-
     static String loadResourceAsString(String fileName) {
 
         try (BufferedReader reader = new BufferedReader(new InputStreamReader(Objects.requireNonNull(ProjectionManagementTests.class.getClassLoader()
@@ -45,7 +41,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(1)
     default void testCreateAndGetContinuousProjection() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         String name = generateName();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
@@ -63,7 +59,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(2)
     default void testDeserializingBasedOnJavaType() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         String name = generateName();
         projectionClient
             .create(name, UNKNOWN_KEYNAMES_PROJECTION)
@@ -79,7 +75,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(3)
     default void testEnablingProjection() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         String name = generateName();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
@@ -96,7 +92,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(4)
     default void testDisablingProjection() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         String name = generateName();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
@@ -118,7 +114,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(5)
     default void testAbortingProjection() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         String name = generateName();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
@@ -134,7 +130,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(6)
     default void testResettingProjection() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         String name = generateName();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
@@ -163,7 +159,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Order(7)
     default void testDeletingProjection() throws ExecutionException, InterruptedException, TimeoutException {
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
             .get();
@@ -197,7 +193,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     default void testGetProjectionStatistics() throws Exception {
         Exceptions exceptions = new Exceptions().registerUnknownError();
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
 
         flaky(10, exceptions, () -> projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
@@ -220,7 +216,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Order(9)
     default void testUpdateProjection() throws ExecutionException, InterruptedException {
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
             .get();
@@ -240,7 +236,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Order(10)
     default void testGetProjectionState() throws ExecutionException, InterruptedException {
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
             .get();
@@ -258,7 +254,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Order(11)
     default void testGetProjectionResultByPartition() throws ExecutionException, InterruptedException {
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient
             .create(name, COUNT_EVENTS_PARTITIONED_PROJECTION)
             .get();
@@ -297,7 +293,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Order(12)
     default void testGetProjectionStateByPartition() throws ExecutionException, InterruptedException {
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient
             .create(name, COUNT_EVENTS_PARTITIONED_PROJECTION)
             .get();
@@ -333,7 +329,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Order(13)
     default void testGetProjectionStatus() throws ExecutionException, InterruptedException {
         String name = generateName();
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient
             .create(name, COUNT_EVENTS_PROJECTION)
             .get();
@@ -356,7 +352,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(14)
     default void testListProjections() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         List<ProjectionDetails> projections = projectionClient.list().get();
         Assertions.assertTrue(projections.size() > 1);
     }
@@ -375,12 +371,12 @@ public interface ProjectionManagementTests extends ConnectionAware {
     @Test
     @Order(15)
     default void testRestartingProjectionSubsystem() throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         projectionClient.restartSubsystem().get();
     }
 
     default Map<String, Item> getResultOfUnknownKeyNamesProjection(String name) throws ExecutionException, InterruptedException {
-        EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
 
         return projectionClient
             .<Map<String, Item>>getResult(name, factory -> factory.constructMapType(HashMap.class, String.class, Item.class))
@@ -424,7 +420,7 @@ public interface ProjectionManagementTests extends ConnectionAware {
     }
 
     default void waitUntilProjectionStatusIs(String name, String... statuses) throws ExecutionException, InterruptedException {
-        final EventStoreDBProjectionManagementClient projectionClient = projectionClient();
+        final EventStoreDBProjectionManagementClient projectionClient = getDefaultProjectionClient();
         CompletableFuture.runAsync(() -> {
             String last = "";
             for (int i = 0; i < 6; i++) {
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/CreatePersistentSubscriptionTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/CreatePersistentSubscriptionTests.java
index aeea971..10a58a5 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/CreatePersistentSubscriptionTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/CreatePersistentSubscriptionTests.java
@@ -6,7 +6,7 @@ import org.junit.jupiter.api.Test;
 public interface CreatePersistentSubscriptionTests extends ConnectionAware {
     @Test
     default void testCreatePersistentSub() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
 
         client.createToStream(generateName(), generateName())
                 .get();
@@ -17,7 +17,7 @@ public interface CreatePersistentSubscriptionTests extends ConnectionAware {
 
     @Test
     default void testCreatePersistentSubToAll() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
 
         client.createToAll(generateName())
                 .get();
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/DeletePersistentSubscriptionToStreamTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/DeletePersistentSubscriptionToStreamTests.java
index 4f1938b..a070379 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/DeletePersistentSubscriptionToStreamTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/DeletePersistentSubscriptionToStreamTests.java
@@ -10,7 +10,7 @@ import java.util.concurrent.ExecutionException;
 public interface DeletePersistentSubscriptionToStreamTests extends ConnectionAware {
     @Test
     default void testDeletePersistentSub() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String streamName = generateName();
         String groupName = generateName();
 
@@ -23,7 +23,7 @@ public interface DeletePersistentSubscriptionToStreamTests extends ConnectionAwa
 
     @Test
     default void testDeletePersistentSubToAll() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String groupName = generateName();
 
         client.createToAll(groupName)
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionManagementTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionManagementTests.java
index 219f138..9818b78 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionManagementTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionManagementTests.java
@@ -18,7 +18,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Test
     @Order(1)
     default void testListPersistentSubscriptions() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String groupName = generateName();
         String streamA = generateName();
         String streamB = generateName();
@@ -46,7 +46,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     default void testListPersistentSubscriptionsForStream() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
 
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String streamName = generateName();
         String groupName = generateName();
 
@@ -63,7 +63,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Test
     @Order(3)
     default void testListPersistentSubscriptionsToAll() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String groupName = generateName();
         client.createToAll(groupName)
                 .get();
@@ -87,7 +87,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Test
     @Order(4)
     default void testGetPersistentSubscriptionInfo() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String streamName = generateName();
         String groupName = generateName();
         client.createToStream(streamName, groupName)
@@ -110,7 +110,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Test
     @Order(5)
     default void testGetPersistentSubscriptionInfoToAll() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String groupName = generateName();
 
         client.createToAll(groupName)
@@ -126,7 +126,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Test
     @Order(6)
      default void testGetPersistentSubscriptionInfoNotExisting() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         Optional<PersistentSubscriptionToStreamInfo> result = client.getInfoToStream(generateName(), generateName()).get();
 
         Assertions.assertFalse(result.isPresent());
@@ -137,7 +137,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Timeout(value = 2, unit = TimeUnit.MINUTES)
     default void testReplayParkedMessages() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         final EventStoreDBClient streamClient = getDatabase().defaultClient();
         final JsonMapper jsonMapper = new JsonMapper();
         final String streamName = generateName();
@@ -196,7 +196,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Timeout(value = 2, unit = TimeUnit.MINUTES)
     default void testReplayParkedMessagesToAll() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         final EventStoreDBClient streamClient = getDatabase().defaultClient();
         final JsonMapper jsonMapper = new JsonMapper();
         String streamName = generateName();
@@ -254,7 +254,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Order(9)
     default void testEncoding() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         String streamName = String.format("/foo/%s/stream", generateName());
         String groupName = String.format("/foo/%s/group", generateName());
 
@@ -279,7 +279,7 @@ public interface PersistentSubscriptionManagementTests extends ConnectionAware {
     @Test
     @Order(10)
     default void testRestartSubsystem() throws Throwable {
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         client.restartSubsystem().get();
     }
 }
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java
index 36eb120..8c765e5 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/PersistentSubscriptionToAllWithFilterTests.java
@@ -12,7 +12,7 @@ public interface PersistentSubscriptionToAllWithFilterTests extends ConnectionAw
     @Test
     default void testPersistentSubscriptionToAllWithFilter() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         EventStoreDBClient streamsClient = getDatabase().defaultClient();
         String groupName = generateName();
         int filteredEventTypeCount = 10;
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java
index 0107ea6..ac3f601 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/SubscribePersistentSubscriptionToStreamTests.java
@@ -13,7 +13,7 @@ public interface SubscribePersistentSubscriptionToStreamTests extends Connection
     default void testSubscribePersistentSub() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
         String streamName = generateName();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         EventStoreDBClient streamsClient = getDatabase().defaultClient();
         JsonMapper jsonMapper = new JsonMapper();
 
@@ -66,7 +66,7 @@ public interface SubscribePersistentSubscriptionToStreamTests extends Connection
     default void testSubscribePersistentSubToAll() throws Throwable {
         Exceptions exceptions = new Exceptions().registerGoAwayError();
         String streamName = generateName();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         EventStoreDBClient streamsClient = getDatabase().defaultClient();
         final JsonMapper jsonMapper = new JsonMapper();
 
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/UpdatePersistentSubscriptionToStreamTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/UpdatePersistentSubscriptionToStreamTests.java
index a839752..ed85a4c 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/UpdatePersistentSubscriptionToStreamTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/persistentsubscriptions/UpdatePersistentSubscriptionToStreamTests.java
@@ -10,7 +10,7 @@ public interface UpdatePersistentSubscriptionToStreamTests extends ConnectionAwa
         String streamName = generateName();
         String groupName = generateName();
 
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
         client.createToStream(streamName, groupName)
                 .get();
 
@@ -25,7 +25,7 @@ public interface UpdatePersistentSubscriptionToStreamTests extends ConnectionAwa
     @Test
     default void testUpdatePersistentSubToAll() throws Throwable {
         String groupName = generateName();
-        EventStoreDBPersistentSubscriptionsClient client = EventStoreDBPersistentSubscriptionsClient.from(getDatabase().defaultClient());
+        EventStoreDBPersistentSubscriptionsClient client = getDefaultPersistentSubscriptionClient();
 
         client.createToAll(groupName)
                     .get();
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/plugins/ClientCertificateAuthenticationTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/plugins/ClientCertificateAuthenticationTests.java
index 19cf686..6294a77 100644
--- a/db-client-java/src/test/java/com/eventstore/dbclient/plugins/ClientCertificateAuthenticationTests.java
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/plugins/ClientCertificateAuthenticationTests.java
@@ -13,7 +13,7 @@ public interface ClientCertificateAuthenticationTests extends ConnectionAware {
             EventStoreDBClient client = getDatabase()
                     .createClient(getDatabase()
                             .defaultSettingsBuilder()
-                            .defaultClientCertificate(ClientCertificate("admin"), userKey("admin"))
+                            .defaultClientCertificate(clientCertificate("admin"), userKey("admin"))
                             .defaultCredentials(null)
                             .buildConnectionSettings());
 
@@ -21,7 +21,7 @@ public interface ClientCertificateAuthenticationTests extends ConnectionAware {
         });
     }
 
-    static String ClientCertificate(String user) {
+    static String clientCertificate(String user) {
         return buildCertPath(user, "crt");
     }
 
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/PersistentSubscriptionsTracingInstrumentationTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/PersistentSubscriptionsTracingInstrumentationTests.java
new file mode 100644
index 0000000..ce22569
--- /dev/null
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/PersistentSubscriptionsTracingInstrumentationTests.java
@@ -0,0 +1,122 @@
+package com.eventstore.dbclient.telemetry;
+
+import com.eventstore.dbclient.*;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public interface PersistentSubscriptionsTracingInstrumentationTests extends TelemetryAware {
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testPersistentSubscriptionIsInstrumentedWithTracingAndRestoresRemoteContextAsExpected() throws Throwable {
+        EventStoreDBClient streamsClient = getDefaultClient();
+        EventStoreDBPersistentSubscriptionsClient psClient = getDefaultPersistentSubscriptionClient();
+
+        String streamName = generateName();
+        String groupName = "aGroup";
+
+        EventData[] events = {
+                EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                        .eventId(UUID.randomUUID())
+                        .build(),
+                EventData.builderAsJson("TestEvent", 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());
+
+        PersistentSubscription subscription = psClient.subscribeToStream(
+                streamName,
+                groupName,
+                SubscribePersistentSubscriptionOptions.get().bufferSize(32),
+                new PersistentSubscriptionListener() {
+                }
+        ).get();
+
+        subscribeSpansLatch.await();
+        subscription.stop();
+
+        List<ReadableSpan> appendSpans = getSpansForOperation(ClientTelemetryConstants.Operations.APPEND);
+        Assertions.assertEquals(1, appendSpans.size());
+
+        List<ReadableSpan> subscribeSpans = getSpansForOperation(ClientTelemetryConstants.Operations.SUBSCRIBE);
+        Assertions.assertEquals(events.length, subscribeSpans.size());
+
+        SpanContext appendSpanContext = appendSpans.get(0).getSpanContext();
+        for (int i = 0; i < subscribeSpans.size(); i++) {
+            ReadableSpan subscribeSpan = subscribeSpans.get(i);
+            SpanContext parentSpanContext = subscribeSpan.getParentSpanContext();
+
+            Assertions.assertNotNull(parentSpanContext);
+            Assertions.assertEquals(appendSpanContext.getTraceId(), parentSpanContext.getTraceId());
+            Assertions.assertEquals(appendSpanContext.getSpanId(), parentSpanContext.getSpanId());
+            Assertions.assertTrue(parentSpanContext.isRemote());
+
+            assertSubscriptionActivityHasExpectedAttributes(
+                    subscribeSpan,
+                    streamName,
+                    subscription.getSubscriptionId(),
+                    events[i].getEventId().toString(),
+                    events[i].getEventType());
+        }
+    }
+
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testPersistentSubscriptionInstrumentationHandlesExceptionsAsExpected() throws Throwable {
+        EventStoreDBClient streamsClient = getDefaultClient();
+        EventStoreDBPersistentSubscriptionsClient psClient = getDefaultPersistentSubscriptionClient();
+
+        String streamName = generateName();
+        String groupName = generateName();
+
+        Exceptions exceptions = new Exceptions().registerGoAwayError();
+        flaky(10, exceptions, () -> psClient.createToStream(streamName, groupName).get());
+
+        streamsClient.appendToStream(
+                        streamName,
+                        EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                                .eventId(UUID.randomUUID())
+                                .build())
+                .get();
+
+        CountDownLatch subscribeSpansLatch = new CountDownLatch(1);
+        onOperationSpanEnded(ClientTelemetryConstants.Operations.SUBSCRIBE, span -> subscribeSpansLatch.countDown());
+
+        RuntimeException expectedException = new RuntimeException("Oops! something went wrong...");
+        PersistentSubscription subscription = psClient.subscribeToStream(streamName, groupName, new PersistentSubscriptionListener() {
+            @Override
+            public void onEvent(PersistentSubscription subscription, int retryCount, ResolvedEvent event) {
+                throw expectedException;
+            }
+
+            @Override
+            public void onCancelled(PersistentSubscription subscription, Throwable throwable) {
+                if (throwable != null && !throwable.equals(expectedException))
+                    Assertions.fail(throwable);
+            }
+        }).get();
+
+        subscribeSpansLatch.await();
+        subscription.stop();
+
+        List<ReadableSpan> subscribeSpans = getSpansForOperation(ClientTelemetryConstants.Operations.SUBSCRIBE);
+        Assertions.assertEquals(1, subscribeSpans.size());
+
+        assertErroneousSpanHasExpectedAttributes(subscribeSpans.get(0), expectedException);
+    }
+}
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/SpanProcessorSpy.java b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/SpanProcessorSpy.java
new file mode 100644
index 0000000..b64548f
--- /dev/null
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/SpanProcessorSpy.java
@@ -0,0 +1,38 @@
+package com.eventstore.dbclient.telemetry;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.ReadWriteSpan;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.SpanProcessor;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.List;
+import java.util.function.Consumer;
+
+public class SpanProcessorSpy implements SpanProcessor {
+    private final List<Consumer<ReadableSpan>> spanEndedHooks;
+
+    public SpanProcessorSpy(List<Consumer<ReadableSpan>> spanEndedHooks) {
+        this.spanEndedHooks = spanEndedHooks;
+    }
+
+    @Override
+    public void onStart(@NotNull Context context, @NotNull ReadWriteSpan readWriteSpan) {
+        // Do nothing.
+    }
+
+    @Override
+    public boolean isStartRequired() {
+        return false;
+    }
+
+    @Override
+    public void onEnd(@NotNull ReadableSpan readableSpan) {
+        spanEndedHooks.forEach(hook -> hook.accept(readableSpan));
+    }
+
+    @Override
+    public boolean isEndRequired() {
+        return true;
+    }
+}
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/StreamsTracingInstrumentationTests.java b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/StreamsTracingInstrumentationTests.java
new file mode 100644
index 0000000..52ccee7
--- /dev/null
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/StreamsTracingInstrumentationTests.java
@@ -0,0 +1,212 @@
+package com.eventstore.dbclient.telemetry;
+
+import com.eventstore.dbclient.*;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+public interface StreamsTracingInstrumentationTests extends TelemetryAware {
+    @Test
+    default void testAppendIsInstrumentedWithTracingAsExpected() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        String streamName = generateName();
+
+        client.appendToStream(
+                        streamName,
+                        AppendToStreamOptions.get().expectedRevision(ExpectedRevision.noStream()),
+                        EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                                .eventId(UUID.randomUUID())
+                                .build())
+                .get();
+
+        List<ReadableSpan> spans = getSpansForOperation(ClientTelemetryConstants.Operations.APPEND);
+        Assertions.assertEquals(1, spans.size());
+
+        assertAppendSpanHasExpectedAttributes(spans.get(0), streamName);
+    }
+
+    @Test
+    default void testTracingContextIsInjectedAsExpectedWhenUserMetadataIsJsonObject() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        String streamName = generateName();
+
+        client.appendToStream(
+                        streamName,
+                        AppendToStreamOptions.get().expectedRevision(ExpectedRevision.noStream()),
+                        EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                                .metadataAsBytes(mapper.writeValueAsBytes(new Foo()))
+                                .eventId(UUID.randomUUID())
+                                .build())
+                .get();
+
+        ReadResult readResult = client.readStream(streamName, ReadStreamOptions.get()).get();
+
+        ResolvedEvent resolvedEvent = readResult.getEvents().get(0);
+        Assertions.assertNotNull(resolvedEvent);
+
+        ObjectNode userMetadata = mapper.readValue(resolvedEvent.getEvent().getUserMetadata(), ObjectNode.class);
+
+        JsonNode traceIdNode = userMetadata.get(ClientTelemetryConstants.Metadata.TRACE_ID);
+        JsonNode spanIdNode = userMetadata.get(ClientTelemetryConstants.Metadata.SPAN_ID);
+
+        Assertions.assertNotNull(traceIdNode);
+        Assertions.assertNotNull(spanIdNode);
+    }
+
+    @Test
+    default void testTracingContextInjectionIsIgnoredAsExpectedWhenUserMetadataIsNonNullAndNotAJsonObject() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        String streamName = generateName();
+        byte[] userMetadata = mapper.writeValueAsBytes("clearlynotvalidjson");
+
+        client.appendToStream(
+                        streamName,
+                        AppendToStreamOptions.get().expectedRevision(ExpectedRevision.noStream()),
+                        EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                                .metadataAsBytes(userMetadata)
+                                .eventId(UUID.randomUUID())
+                                .build())
+                .get();
+
+        ReadResult readResult = client.readStream(streamName, ReadStreamOptions.get()).get();
+
+        ResolvedEvent resolvedEvent = readResult.getEvents().get(0);
+        Assertions.assertNotNull(resolvedEvent);
+
+        // Assert unchanged
+        Assertions.assertArrayEquals(userMetadata, resolvedEvent.getEvent().getUserMetadata());
+    }
+
+    @Test
+    default void testAppendInstrumentationHandlesExceptionsAsExpected() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        String streamName = generateName();
+
+        WrongExpectedVersionException actualException = null;
+        try {
+            client.appendToStream(
+                            streamName,
+                            // Force WrongExpectedVersionException to be thrown.
+                            AppendToStreamOptions.get().expectedRevision(ExpectedRevision.streamExists()),
+                            EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                                    .eventId(UUID.randomUUID())
+                                    .build())
+                    .get();
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof WrongExpectedVersionException)
+                actualException = (WrongExpectedVersionException) e.getCause();
+        }
+
+        // Ensure WrongExpectedVersionException was thrown.
+        Assertions.assertNotNull(actualException);
+
+        List<ReadableSpan> spans = getSpansForOperation(ClientTelemetryConstants.Operations.APPEND);
+        Assertions.assertEquals(1, spans.size());
+
+        assertErroneousSpanHasExpectedAttributes(spans.get(0), actualException);
+    }
+
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testCatchupSubscriptionIsInstrumentedWithTracingAndRestoresRemoteContextAsExpected() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        String streamName = generateName();
+
+        EventData[] events = {
+                EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                        .eventId(UUID.randomUUID())
+                        .build(),
+                EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                        .eventId(UUID.randomUUID())
+                        .build()
+        };
+
+        client.appendToStream(streamName, events).get();
+
+        CountDownLatch subscribeSpansLatch = new CountDownLatch(events.length);
+        onOperationSpanEnded(ClientTelemetryConstants.Operations.SUBSCRIBE, span -> subscribeSpansLatch.countDown());
+
+        Subscription subscription = client.subscribeToStream(
+                streamName,
+                new SubscriptionListener() {
+                }
+        ).get();
+
+        subscribeSpansLatch.await();
+        subscription.stop();
+
+        List<ReadableSpan> appendSpans = getSpansForOperation(ClientTelemetryConstants.Operations.APPEND);
+        Assertions.assertEquals(1, appendSpans.size());
+
+        List<ReadableSpan> subscribeSpans = getSpansForOperation(ClientTelemetryConstants.Operations.SUBSCRIBE);
+        Assertions.assertEquals(events.length, subscribeSpans.size());
+
+        SpanContext appendSpanContext = appendSpans.get(0).getSpanContext();
+        for (int i = 0; i < subscribeSpans.size(); i++) {
+            ReadableSpan subscribeSpan = subscribeSpans.get(i);
+            SpanContext parentSpanContext = subscribeSpan.getParentSpanContext();
+
+            Assertions.assertNotNull(parentSpanContext);
+            Assertions.assertEquals(appendSpanContext.getTraceId(), parentSpanContext.getTraceId());
+            Assertions.assertEquals(appendSpanContext.getSpanId(), parentSpanContext.getSpanId());
+            Assertions.assertTrue(parentSpanContext.isRemote());
+
+            assertSubscriptionActivityHasExpectedAttributes(
+                    subscribeSpan,
+                    streamName,
+                    subscription.getSubscriptionId(),
+                    events[i].getEventId().toString(),
+                    events[i].getEventType());
+        }
+    }
+
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testCatchupSubscriptionInstrumentationHandlesExceptionsAsExpected() throws Throwable {
+        EventStoreDBClient client = getDefaultClient();
+        String streamName = generateName();
+
+        client.appendToStream(
+                        streamName,
+                        EventData.builderAsJson("TestEvent", mapper.writeValueAsBytes(new Foo()))
+                                .eventId(UUID.randomUUID())
+                                .build())
+                .get();
+
+        RuntimeException expectedException = new RuntimeException("Oops! something went wrong...");
+
+        CountDownLatch subscribeSpansLatch = new CountDownLatch(1);
+        onOperationSpanEnded(ClientTelemetryConstants.Operations.SUBSCRIBE, span -> subscribeSpansLatch.countDown());
+
+        Subscription subscription = client.subscribeToStream(streamName, new SubscriptionListener() {
+            @Override
+            public void onEvent(Subscription subscription, ResolvedEvent event) {
+                throw expectedException;
+            }
+
+            @Override
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable != null && !throwable.equals(expectedException))
+                    Assertions.fail(throwable);
+            }
+        }).get();
+
+        subscribeSpansLatch.await();
+        subscription.stop();
+
+        List<ReadableSpan> subscribeSpans = getSpansForOperation(ClientTelemetryConstants.Operations.SUBSCRIBE);
+        Assertions.assertEquals(1, subscribeSpans.size());
+
+        assertErroneousSpanHasExpectedAttributes(subscribeSpans.get(0), expectedException);
+    }
+}
diff --git a/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/TelemetryAware.java b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/TelemetryAware.java
new file mode 100644
index 0000000..7353d0e
--- /dev/null
+++ b/db-client-java/src/test/java/com/eventstore/dbclient/telemetry/TelemetryAware.java
@@ -0,0 +1,56 @@
+package com.eventstore.dbclient.telemetry;
+
+import com.eventstore.dbclient.ClientTelemetryAttributes;
+import com.eventstore.dbclient.ClientTelemetryConstants;
+import com.eventstore.dbclient.ConnectionAware;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.internal.data.ExceptionEventData;
+import org.junit.jupiter.api.Assertions;
+
+import java.util.List;
+import java.util.function.Consumer;
+
+public interface TelemetryAware extends ConnectionAware {
+    void onOperationSpanEnded(String operation, Consumer<ReadableSpan> onSpanEnded);
+    List<ReadableSpan> getSpansForOperation(String operation);
+
+    default void assertAppendSpanHasExpectedAttributes(ReadableSpan span, String streamName) {
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.Database.SYSTEM, ClientTelemetryConstants.INSTRUMENTATION_NAME);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.Database.OPERATION, ClientTelemetryConstants.Operations.APPEND);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.EventStore.STREAM, streamName);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.Database.USER, "admin");
+        Assertions.assertEquals(StatusCode.OK, span.toSpanData().getStatus().getStatusCode());
+        Assertions.assertEquals(SpanKind.CLIENT, span.getKind());
+    }
+
+    default void assertErroneousSpanHasExpectedAttributes(ReadableSpan span, Throwable actualException) {
+        SpanData spanData = span.toSpanData();
+        List<io.opentelemetry.sdk.trace.data.EventData> spanEvents = spanData.getEvents();
+
+        Assertions.assertEquals(StatusCode.ERROR, spanData.getStatus().getStatusCode());
+        Assertions.assertEquals(1, spanEvents.size());
+
+        ExceptionEventData errorEvent = (ExceptionEventData) spanEvents.get(0);
+        Assertions.assertEquals(errorEvent.getException(), actualException);
+    }
+
+    default void assertSubscriptionActivityHasExpectedAttributes(ReadableSpan span, String streamName, String subscriptionId, String eventId, String eventType) {
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.Database.SYSTEM, ClientTelemetryConstants.INSTRUMENTATION_NAME);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.Database.OPERATION, ClientTelemetryConstants.Operations.SUBSCRIBE);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.Database.USER, "admin");
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.EventStore.STREAM, streamName);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.EventStore.SUBSCRIPTION_ID, subscriptionId);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.EventStore.EVENT_ID, eventId);
+        assertSpanAttributeEquals(span, ClientTelemetryAttributes.EventStore.EVENT_TYPE, eventType);
+        Assertions.assertEquals(StatusCode.OK, span.toSpanData().getStatus().getStatusCode());
+        Assertions.assertEquals(SpanKind.CONSUMER, span.getKind());
+    }
+
+    default void assertSpanAttributeEquals(ReadableSpan span, String attributeKey, String expectedValue) {
+        Assertions.assertEquals(expectedValue, span.getAttribute(AttributeKey.stringKey(attributeKey)));
+    }
+}