Track schema registry usage

Author: piochelepiotr

.gitignore

@@ -46,6 +46,10 @@ out/
 ######################
 .vscode
 
+# Cursor #
+##########
+.cursor
+
 # Others #
 ##########
 /logs/*

dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/datastreams/RecordingDatastreamsPayloadWriter.groovy

@@ -19,6 +19,9 @@ class RecordingDatastreamsPayloadWriter implements DatastreamsPayloadWriter {
   @SuppressWarnings('UnusedPrivateField')
   private final Set<DataStreamsTags> backlogs = []
 
+  @SuppressWarnings('UnusedPrivateField')
+  private final List<StatsBucket.SchemaRegistryKey> schemaRegistryUsages = []
+
   private final Set<String> serviceNameOverrides = []
 
   @Override
@@ -33,6 +36,11 @@ class RecordingDatastreamsPayloadWriter implements DatastreamsPayloadWriter {
           this.@backlogs.add(backlog.getKey())
         }
       }
+      if (bucket.schemaRegistryUsages != null) {
+        for (Map.Entry<StatsBucket.SchemaRegistryKey, StatsBucket.SchemaRegistryCount> usage : bucket.schemaRegistryUsages) {
+          this.@schemaRegistryUsages.add(usage.getKey())
+        }
+      }
     }
   }
 
@@ -52,10 +60,15 @@ class RecordingDatastreamsPayloadWriter implements DatastreamsPayloadWriter {
     Collections.unmodifiableList(new ArrayList<>(this.@backlogs))
   }
 
+  synchronized List<StatsBucket.SchemaRegistryKey> getSchemaRegistryUsages() {
+    Collections.unmodifiableList(new ArrayList<>(this.@schemaRegistryUsages))
+  }
+
   synchronized void clear() {
     this.@payloads.clear()
     this.@groups.clear()
     this.@backlogs.clear()
+    this.@schemaRegistryUsages.clear()
   }
 
   void waitForPayloads(int count, long timeout = TimeUnit.SECONDS.toMillis(3)) {
@@ -70,6 +83,10 @@ class RecordingDatastreamsPayloadWriter implements DatastreamsPayloadWriter {
     waitFor(count, timeout, this.@backlogs)
   }
 
+  void waitForSchemaRegistryUsages(int count, long timeout = TimeUnit.SECONDS.toMillis(3)) {
+    waitFor(count, timeout, this.@schemaRegistryUsages)
+  }
+
   private static void waitFor(int count, long timeout, Collection collection) {
     long deadline = System.currentTimeMillis() + timeout
     while (System.currentTimeMillis() < deadline) {

dd-java-agent/instrumentation/confluent-schema-registry/confluent-schema-registry-7.0/README.md

@@ -0,0 +1,141 @@
+# Confluent Schema Registry Instrumentation
+
+This instrumentation module provides detailed observability for Confluent Schema Registry operations in Kafka applications.
+
+## Features
+
+This instrumentation captures:
+
+### Producer Operations
+- **Schema Registration**: Tracks when schemas are registered with the Schema Registry
+  - Subject name
+  - Schema ID assigned
+  - Success/failure status
+  - Compatibility check results
+- **Serialization**: Logs every message serialization with:
+  - Topic name
+  - Key schema ID (if applicable)
+  - Value schema ID
+  - Success/failure status
+
+### Consumer Operations  
+- **Deserialization**: Tracks every message deserialization with:
+  - Topic name
+  - Key schema ID (if present in message)
+  - Value schema ID (extracted from Confluent wire format)
+  - Success/failure status
+
+### Schema Registry Client Operations
+- **Schema Registration** (`register()` method)
+  - Successful registrations with schema ID
+  - Compatibility failures with error details
+- **Compatibility Checks** (`testCompatibility()` method)
+  - Pass/fail status
+  - Error messages for incompatible schemas
+- **Schema Retrieval** (`getSchemaById()` method)
+  - Schema ID lookups during deserialization
+
+## Metrics Collected
+
+The `SchemaRegistryMetrics` class tracks:
+
+- `schemaRegistrationSuccess` - Count of successful schema registrations
+- `schemaRegistrationFailure` - Count of failed schema registrations (compatibility issues)
+- `schemaCompatibilitySuccess` - Count of successful compatibility checks
+- `schemaCompatibilityFailure` - Count of failed compatibility checks
+- `serializationSuccess` - Count of successful message serializations
+- `serializationFailure` - Count of failed serializations
+- `deserializationSuccess` - Count of successful message deserializations
+- `deserializationFailure` - Count of failed deserializations
+
+## Log Output Examples
+
+### Successful Producer Operation
+```
+[Schema Registry] Schema registered successfully - Subject: myTopic-value, Schema ID: 123, Is Key: false, Topic: myTopic
+[Schema Registry] Produce to topic 'myTopic', schema for key: none, schema for value: 123, serializing: VALUE
+```
+
+### Failed Schema Registration (Incompatibility)
+```
+[Schema Registry] Schema registration FAILED - Subject: myTopic-value, Is Key: false, Topic: myTopic, Error: Schema being registered is incompatible with an earlier schema
+[Schema Registry] Schema compatibility check FAILED - Subject: myTopic-value, Error: Schema being registered is incompatible with an earlier schema
+[Schema Registry] Serialization FAILED for topic 'myTopic', VALUE - Error: Schema being registered is incompatible with an earlier schema
+```
+
+### Consumer Operation
+```
+[Schema Registry] Retrieved schema from registry - Schema ID: 123, Type: Schema
+[Schema Registry] Consume from topic 'myTopic', schema for key: none, schema for value: 123, deserializing: VALUE
+```
+
+## Supported Serialization Formats
+
+- **Avro** (via `KafkaAvroSerializer`/`KafkaAvroDeserializer`)
+- **Protobuf** (via `KafkaProtobufSerializer`/`KafkaProtobufDeserializer`)
+
+## Implementation Details
+
+### Instrumented Classes
+
+1. **CachedSchemaRegistryClient** - The main Schema Registry client
+   - `register(String subject, Schema schema)` - Schema registration
+   - `testCompatibility(String subject, Schema schema)` - Compatibility testing
+   - `getSchemaById(int id)` - Schema retrieval
+
+2. **AbstractKafkaSchemaSerDe and subclasses** - Serializers
+   - `serialize(String topic, Object data)` - Message serialization
+   - `serialize(String topic, Headers headers, Object data)` - With headers (Kafka 2.1+)
+
+3. **AbstractKafkaSchemaSerDe and subclasses** - Deserializers
+   - `deserialize(String topic, byte[] data)` - Message deserialization  
+   - `deserialize(String topic, Headers headers, byte[] data)` - With headers (Kafka 2.1+)
+
+### Context Management
+
+The `SchemaRegistryContext` class uses ThreadLocal storage to pass context between:
+- Serializer → Schema Registry Client (for logging topic information)
+- Deserializer → Schema Registry Client (for logging topic information)
+
+This allows the instrumentation to correlate schema operations with the topics they're associated with.
+
+## Usage
+
+This instrumentation is automatically activated when:
+1. Confluent Schema Registry client (version 7.0.0+) is present on the classpath
+2. The Datadog Java agent is attached to the JVM
+
+No configuration or code changes are required.
+
+## Metrics Access
+
+To access metrics programmatically:
+
+```java
+import datadog.trace.instrumentation.confluentschemaregistry.SchemaRegistryMetrics;
+
+// Get current counts
+long registrationFailures = SchemaRegistryMetrics.getSchemaRegistrationFailureCount();
+long compatibilityFailures = SchemaRegistryMetrics.getSchemaCompatibilityFailureCount();
+long serializationFailures = SchemaRegistryMetrics.getSerializationFailureCount();
+
+// Print summary
+SchemaRegistryMetrics.printSummary();
+```
+
+## Monitoring Schema Compatibility Issues
+
+The primary use case for this instrumentation is to detect and monitor schema compatibility issues that cause production failures. By tracking `schemaRegistrationFailure` and `schemaCompatibilityFailure` metrics, you can:
+
+1. **Alert on schema compatibility failures** before they impact production
+2. **Track the rate of schema-related errors** per topic
+3. **Identify problematic schema changes** that break compatibility
+4. **Monitor serialization/deserialization failure rates** as a proxy for schema issues
+
+## Future Enhancements
+
+Potential additions:
+- JSON Schema serializer support (currently excluded due to dependency issues)
+- Schema evolution tracking
+- Schema version diff logging
+- Integration with Datadog APM for schema-related span tags

dd-java-agent/instrumentation/confluent-schema-registry/confluent-schema-registry-7.0/build.gradle

@@ -0,0 +1,24 @@
+apply from: "$rootDir/gradle/java.gradle"
+
+muzzle {
+  pass {
+    group = "io.confluent"
+    module = "kafka-schema-registry-client"
+    versions = "[7.0.0,)"
+    assertInverse = true
+  }
+}
+
+dependencies {
+  compileOnly group: 'io.confluent', name: 'kafka-schema-registry-client', version: '7.0.0'
+  compileOnly group: 'io.confluent', name: 'kafka-avro-serializer', version: '7.0.0'
+  compileOnly group: 'io.confluent', name: 'kafka-protobuf-serializer', version: '7.0.0'
+  compileOnly group: 'org.apache.kafka', name: 'kafka-clients', version: '3.0.0'
+
+  testImplementation group: 'io.confluent', name: 'kafka-schema-registry-client', version: '7.5.2'
+  testImplementation group: 'io.confluent', name: 'kafka-avro-serializer', version: '7.5.2'
+  testImplementation group: 'io.confluent', name: 'kafka-protobuf-serializer', version: '7.5.1'
+  testImplementation group: 'org.apache.kafka', name: 'kafka-clients', version: '3.5.0'
+  testImplementation group: 'org.apache.avro', name: 'avro', version: '1.11.0'
+}
+

dd-java-agent/instrumentation/confluent-schema-registry/confluent-schema-registry-7.0/src/main/java/datadog/trace/instrumentation/confluentschemaregistry/ConfluentSchemaRegistryModule.java

@@ -0,0 +1,45 @@
+package datadog.trace.instrumentation.confluentschemaregistry;
+
+import static datadog.trace.agent.tooling.bytebuddy.matcher.ClassLoaderMatchers.hasClassNamed;
+import static java.util.Arrays.asList;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.agent.tooling.InstrumenterModule;
+import java.util.List;
+import net.bytebuddy.matcher.ElementMatcher;
+
+/**
+ * Instrumentation module for Confluent Schema Registry to capture schema operations including
+ * registration, compatibility checks, serialization, and deserialization.
+ */
+@AutoService(InstrumenterModule.class)
+public class ConfluentSchemaRegistryModule extends InstrumenterModule.Tracing {
+
+  public ConfluentSchemaRegistryModule() {
+    super("confluent-schema-registry");
+  }
+
+  @Override
+  public String[] helperClassNames() {
+    return new String[] {
+      packageName + ".SchemaRegistryMetrics", packageName + ".SchemaRegistryContext",
+    };
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public List<Instrumenter> typeInstrumentations() {
+    return (List<Instrumenter>)
+        (List<?>)
+            asList(
+                new SchemaRegistryClientInstrumentation(),
+                new KafkaAvroSerializerInstrumentation(),
+                new KafkaAvroDeserializerInstrumentation());
+  }
+
+  @Override
+  public ElementMatcher.Junction<ClassLoader> classLoaderMatcher() {
+    return hasClassNamed("io.confluent.kafka.schemaregistry.client.SchemaRegistryClient");
+  }
+}