Fix context store type mismatch in producer context propagation advices

The ContextPropagationAdvice classes were still using
InstrumentationContext.get(Metadata.class, String.class) after the
context store value type was changed from String to MetadataState,
causing an instrumentation error at runtime.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: piochelepiotr

dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaProducerInstrumentation.java

@@ -200,7 +200,9 @@ public static void onEnter(
         @Advice.Argument(value = 0, readOnly = false) ProducerRecord record) {
       AgentSpan span = activeSpan();
       if (span == null) return;
-      String clusterId = InstrumentationContext.get(Metadata.class, String.class).get(metadata);
+      MetadataState metadataState =
+          InstrumentationContext.get(Metadata.class, MetadataState.class).get(metadata);
+      String clusterId = metadataState != null ? metadataState.clusterId : null;
       TextMapInjectAdapterInterface setter = NoopTextMapInjectAdapter.NOOP_SETTER;
       // Do not inject headers for batch versions below 2
       // This is how similar check is being done in Kafka client itself:

dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/main/java17/datadog/trace/instrumentation/kafka_clients38/ProducerContextPropagationAdvice.java

@@ -18,6 +18,7 @@
 import datadog.trace.api.datastreams.DataStreamsTags;
 import datadog.trace.bootstrap.InstrumentationContext;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.instrumentation.kafka_common.MetadataState;
 import net.bytebuddy.asm.Advice;
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -31,7 +32,9 @@ public static void onEnter(
       @Advice.Argument(value = 0, readOnly = false) ProducerRecord record) {
     AgentSpan span = activeSpan();
     if (span == null) return;
-    String clusterId = InstrumentationContext.get(Metadata.class, String.class).get(metadata);
+    MetadataState metadataState =
+        InstrumentationContext.get(Metadata.class, MetadataState.class).get(metadata);
+    String clusterId = metadataState != null ? metadataState.clusterId : null;
     TextMapInjectAdapterInterface setter = NoopTextMapInjectAdapter.NOOP_SETTER;
     // Please note that the minimum magic for kafka 3.8+ is 2 so there is no need to check this
     if (Config.get().isKafkaClientPropagationEnabled()