diff --git a/entity-types/infra-kafkatopic/definition.stg.yml b/entity-types/infra-kafkatopic/definition.stg.yml
index e521685db..356879d3c 100644
--- a/entity-types/infra-kafkatopic/definition.stg.yml
+++ b/entity-types/infra-kafkatopic/definition.stg.yml
@@ -32,6 +32,7 @@ synthesis:
     tags:
       instrumentation.provider:
       kafka.cluster.name:
+      kafka.cluster.id:
       topic:
 
   - ruleName: infra_kafkatopic_otel_jmxscraper_stg
@@ -59,6 +60,7 @@ synthesis:
     tags:
       instrumentation.provider:
       kafka.cluster.name:
+      kafka.cluster.id:
       topic:
 
   - ruleName: infra_kafkatopic_otel_kafkametrics_stg
@@ -114,6 +116,7 @@ synthesis:
     tags:
       instrumentation.provider:
       kafka.cluster.name:
+      kafka.cluster.id:
       topic:
 
   # Log-based rule: OTel kafkareceiver logs carry topic, kafka.cluster.name, and producer.service.name.
diff --git a/relationships/synthesis/EXT-SERVICE-to-KAFKATOPIC.stg.yml b/relationships/synthesis/EXT-SERVICE-to-KAFKATOPIC.stg.yml
index faf76f019..768a05a0e 100644
--- a/relationships/synthesis/EXT-SERVICE-to-KAFKATOPIC.stg.yml
+++ b/relationships/synthesis/EXT-SERVICE-to-KAFKATOPIC.stg.yml
@@ -425,7 +425,7 @@ relationships:
             - field: producerServiceName
               attribute: producer.service.name
 
-  # Self-hosted Kafka producer span → KAFKATOPIC via kafka.cluster.id + messaging.destination.name
+  # OTel producer span → KAFKATOPIC via messaging.kafka.cluster.id (OTel incubating semconv)
   - name: extServiceProducesSelfHostKafkaTopicClusterIdStg
     version: "1"
     origins:
@@ -442,7 +442,7 @@ relationships:
         anyOf: [ "producer", "client" ]
       - attribute: messaging.destination.name
         present: true
-      - attribute: kafka.cluster.id
+      - attribute: messaging.kafka.cluster.id
         present: true
     relationship:
       expires: PT75M
@@ -455,11 +455,11 @@ relationships:
           candidateCategory: KAFKATOPIC
           fields:
             - field: clusterId
-              attribute: kafka.cluster.id
+              attribute: messaging.kafka.cluster.id
             - field: topic
               attribute: messaging.destination.name
 
-  # Self-hosted Kafka consumer span → KAFKATOPIC via kafka.cluster.id + messaging.destination.name
+  # OTel consumer span → KAFKATOPIC via messaging.kafka.cluster.id (OTel incubating semconv)
   - name: extServiceConsumesSelfHostKafkaTopicClusterIdStg
     version: "1"
     origins:
@@ -476,7 +476,7 @@ relationships:
         anyOf: [ "consumer", "server" ]
       - attribute: messaging.destination.name
         present: true
-      - attribute: kafka.cluster.id
+      - attribute: messaging.kafka.cluster.id
         present: true
     relationship:
       expires: PT75M
@@ -489,7 +489,73 @@ relationships:
           candidateCategory: KAFKATOPIC
           fields:
             - field: clusterId
-              attribute: kafka.cluster.id
+              attribute: messaging.kafka.cluster.id
             - field: topic
               attribute: messaging.destination.name
 
+  # Java kafka-clients-cluster-metrics producer metrics → KAFKATOPIC via messaging.kafka.cluster.id + topic
+  - name: extServiceProducesKafkaClientMetricsTopicClusterIdStg
+    version: "1"
+    origins:
+      - OpenTelemetry
+    conditions:
+      - attribute: eventType
+        anyOf: [ "Metric" ]
+      - attribute: instrumentation.provider
+        anyOf: [ "opentelemetry" ]
+      - attribute: metricName
+        startsWith: "kafka.producer"
+      - attribute: entity.type
+        anyOf: [ "SERVICE" ]
+      - attribute: messaging.kafka.cluster.id
+        present: true
+      - attribute: topic
+        present: true
+    relationship:
+      expires: PT75M
+      relationshipType: PRODUCES
+      source:
+        extractGuid:
+          attribute: entity.guid
+      target:
+        lookupGuid:
+          candidateCategory: KAFKATOPIC
+          fields:
+            - field: clusterId
+              attribute: messaging.kafka.cluster.id
+            - field: topic
+              attribute: topic
+
+  # Java kafka-clients-cluster-metrics consumer metrics → KAFKATOPIC via messaging.kafka.cluster.id + topic
+  - name: extServiceConsumesKafkaClientMetricsTopicClusterIdStg
+    version: "1"
+    origins:
+      - OpenTelemetry
+    conditions:
+      - attribute: eventType
+        anyOf: [ "Metric" ]
+      - attribute: instrumentation.provider
+        anyOf: [ "opentelemetry" ]
+      - attribute: metricName
+        startsWith: "kafka.consumer"
+      - attribute: entity.type
+        anyOf: [ "SERVICE" ]
+      - attribute: messaging.kafka.cluster.id
+        present: true
+      - attribute: topic
+        present: true
+    relationship:
+      expires: PT75M
+      relationshipType: CONSUMES
+      source:
+        extractGuid:
+          attribute: entity.guid
+      target:
+        lookupGuid:
+          candidateCategory: KAFKATOPIC
+          fields:
+            - field: clusterId
+              attribute: messaging.kafka.cluster.id
+            - field: topic
+              attribute: topic
+