kafka: Add topic partition size metric (DataDog#23028)

* kafka: Add topic partition size metric

Add kafka.log.partition.size metric from the kafka.log:type=Log,name=Size
JMX bean, tagged with topic and partition. Reports the on-disk size in
bytes of each topic partition log per broker.

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

* kafka: Rename excluded version tag to request_version for RequestsPerSec beans

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

---------

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

Author: piochelepiotr

kafka/changelog.d/23028.added

@@ -0,0 +1 @@
+Add kafka.log.partition.size metric for topic partition disk usage.

kafka/datadog_checks/kafka/data/metrics.yaml

@@ -469,6 +469,10 @@ jmx_metrics:
     - include:
         domain: 'kafka.network'
         bean_regex: 'kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce(?:,version=.*)?'
+        exclude_tags:
+          - version
+        tags:
+          request_version: $version
         attribute:
           Count:
             metric_type: rate
@@ -494,6 +498,10 @@ jmx_metrics:
     - include:
         domain: 'kafka.network'
         bean_regex: 'kafka.network:type=RequestMetrics,name=RequestsPerSec,request=FetchConsumer(?:,version=.*)?'
+        exclude_tags:
+          - version
+        tags:
+          request_version: $version
         attribute:
           Count:
             metric_type: rate
@@ -505,6 +513,10 @@ jmx_metrics:
     - include:
         domain: 'kafka.network'
         bean_regex: 'kafka.network:type=RequestMetrics,name=RequestsPerSec,request=FetchFollower(?:,version=.*)?'
+        exclude_tags:
+          - version
+        tags:
+          request_version: $version
         attribute:
           Count:
             metric_type: rate
@@ -932,3 +944,21 @@ jmx_metrics:
         - tag_name: kafka_cluster_id
           bean_name: kafka.server:type=KafkaServer,name=ClusterId
           attribute: Value
+
+    #
+    # Topic partition size
+    #
+    - include:
+        domain: 'kafka.log'
+        bean_regex: 'kafka\.log:type=Log,name=Size,topic=(.*),partition=(.*)'
+        attribute:
+          Value:
+            metric_type: gauge
+            alias: kafka.log.partition.size
+        tags:
+          topic: $1
+          partition: $2
+      dynamic_tags:
+        - tag_name: kafka_cluster_id
+          bean_name: kafka.server:type=KafkaServer,name=ClusterId
+          attribute: Value

kafka/metadata.csv

@@ -17,6 +17,7 @@ kafka.expires_sec,gauge,10,eviction,second,Rate of delayed producer request expi
 kafka.follower.expires_per_second,gauge,10,eviction,second,Rate of request expiration on followers.,-1,kafka,follow expire rate,,
 kafka.log.directory.offline,gauge,10,,,Whether a Kafka log directory is offline. 0 means healthy.,0,kafka,log dir offline,,
 kafka.log.flush_rate.rate,gauge,10,flush,second,Log flush rate.,0,kafka,log flush rate,memory,
+kafka.log.partition.size,gauge,10,byte,,The size in bytes of a topic partition log on disk.,0,kafka,partition size,,
 kafka.messages_in.rate,gauge,10,message,,Incoming message rate.,0,kafka,messages in,cpu,
 kafka.net.bytes_in.rate,gauge,10,byte,second,Incoming byte rate.,0,kafka,bytes in,cpu,
 kafka.net.bytes_out,gauge,,byte,second,Outgoing byte total.,0,kafka,bytes out total,,

kafka/tests/common.py

@@ -13,15 +13,14 @@
 
 
 """
-Metrics that do not work in our e2e:
+Metrics that require a consumer in our e2e:
     "kafka.consumer.bytes_in",
     "kafka.consumer.kafka_commits",
     "kafka.consumer.messages_in",
     "kafka.consumer.zookeeper_commits",
-    "kafka.request.produce.rate",
-    "kafka.request.fetch_follower.rate",
-    "kafka.request.fetch_consumer.rate",
     "kafka.consumer.fetch_rate",
+    "kafka.request.fetch_consumer.rate",
+    "kafka.request.fetch_follower.rate",
 """
 
 KAFKA_E2E_METRICS = [
@@ -60,6 +59,7 @@
     "kafka.replication.unclean_leader_elections.rate",
     "kafka.request.fetch.failed.rate",
     "kafka.request.produce.failed.rate",
+    "kafka.request.produce.rate",
     "kafka.session.zookeeper.disconnect.rate",
     "kafka.session.zookeeper.expire.rate",
     "kafka.session.zookeeper.readonly.rate",
@@ -73,4 +73,9 @@
     "kafka.broker.start_time",
     # Log directory
     "kafka.log.directory.offline",
+    # Topic partition size
+    "kafka.log.partition.size",
+    # Topic metrics
+    "kafka.topic.messages_in.rate",
+    "kafka.topic.net.bytes_in.rate",
 ]

kafka/tests/compose/docker-compose.yml

@@ -23,10 +23,24 @@ services:
       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
-      KAFKA_CREATE_TOPICS: "marvel:2:1,dc:2:1"
       KAFKA_JMX_PORT: 9999
       KAFKA_JMX_HOSTNAME: localhost
     depends_on:
       zookeeper:
         condition: service_healthy
+  kafka-init:
+    image: confluentinc/cp-kafka:${CONFLUENT_VERSION}
+    network_mode: "service:kafka"
+    depends_on:
+      - kafka
+    entrypoint: ["/bin/bash", "-c"]
+    command:
+      - |
+        echo "Waiting for Kafka to be ready..."
+        while ! kafka-topics --bootstrap-server localhost:9092 --list > /dev/null 2>&1; do sleep 1; done
+        kafka-topics --create --if-not-exists --topic marvel --partitions 2 --replication-factor 1 --bootstrap-server localhost:9092
+        kafka-topics --create --if-not-exists --topic dc --partitions 2 --replication-factor 1 --bootstrap-server localhost:9092
+        echo "test" | kafka-console-producer --bootstrap-server localhost:9092 --topic marvel
+        echo "test" | kafka-console-producer --bootstrap-server localhost:9092 --topic dc
+        echo "Topics created and seeded."