Kafka metrics not split for consumer and producer

[ctoestreich]

The way the metrics are registered is a bit global now where producer and consumer metrics are not differentiated.

Here is the current state of metrics where I ran the app with a consumer only, a producer only and both. Where they overlap there might be odd/mixed metrics reported. When I looked at the data in an overlapping metrics the issue is that it doesn't have a tag that differentiates from consumer and producer either.

Also, I am not sure if converting the - to a . has any value specifically as we use DataDog and that accepts -'s just fine.

Overlapping metric

http://localhost:8080/metrics/kafka.connection-count

Metric Name and Usage

Name	Consumer	Producer
kafka.assigned-partitions	X
kafka.batch-size-avg		X
kafka.batch-size-max		X
kafka.batch-split-rate		X
kafka.batch-split-total		X
kafka.buffer-available-bytes		X
kafka.buffer-exhausted-rate		X
kafka.buffer-exhausted-total		X
kafka.buffer-total-bytes		X
kafka.bufferpool-wait-ratio		X
kafka.bufferpool-wait-time-total		X
kafka.byte-rate		X
kafka.byte-total		X
kafka.bytes-consumed-rate	X
kafka.bytes-consumed-total	X
kafka.commit-latency-avg	X
kafka.commit-latency-max	X
kafka.commit-rate	X
kafka.commit-total	X
kafka.compression-rate		X
kafka.compression-rate-avg		X
kafka.connection-close-rate	X	X
kafka.connection-close-total	X	X
kafka.connection-count	X	X
kafka.connection-creation-rate	X	X
kafka.connection-creation-total	X	X
kafka.count	X	X
kafka.failed-authentication-rate	X	X
kafka.failed-authentication-total	X	X
kafka.fetch-latency-avg	X
kafka.fetch-latency-max	X
kafka.fetch-rate	X
kafka.fetch-size-avg	X
kafka.fetch-size-max	X
kafka.fetch-throttle-time-avg	X
kafka.fetch-throttle-time-max	X
kafka.fetch-total	X
kafka.heartbeat-rate	X
kafka.heartbeat-response-time-max	X
kafka.heartbeat-total	X
kafka.incoming-byte-rate	X	X
kafka.incoming-byte-total	X	X
kafka.io-ratio	X	X
kafka.io-time-ns-avg	X	X
kafka.io-wait-ratio	X	X
kafka.io-wait-time-ns-avg	X	X
kafka.io-waittime-total	X	X
kafka.iotime-total	X	X
kafka.join-rate	X
kafka.join-time-avg	X
kafka.join-time-max	X
kafka.join-total	X
kafka.last-heartbeat-seconds-ago	X
kafka.metadata-age		X
kafka.network-io-rate	X	X
kafka.network-io-total	X	X
kafka.outgoing-byte-rate	X	X
kafka.outgoing-byte-total	X	X
kafka.produce-throttle-time-avg		X
kafka.produce-throttle-time-max		X
kafka.record-error-rate		X
kafka.record-error-total		X
kafka.record-queue-time-avg		X
kafka.record-queue-time-max		X
kafka.record-retry-rate		X
kafka.record-retry-total		X
kafka.record-send-rate		X
kafka.record-send-total		X
kafka.record-size-avg		X
kafka.record-size-max		X
kafka.records-consumed-rate	X
kafka.records-consumed-total	X
kafka.records-lag	X
kafka.records-lag-avg	X
kafka.records-lag-max	X
kafka.records-lead	X
kafka.records-lead-avg	X
kafka.records-lead-min	X
kafka.records-per-request-avg	X	X
kafka.request-latency-avg	X	X
kafka.request-latency-max	X	X
kafka.request-rate	X	X
kafka.request-size-avg	X	X
kafka.request-size-max	X	X
kafka.request-total	X	X
kafka.requests-in-flight		X
kafka.response-rate	X	X
kafka.response-total	X	X
kafka.select-rate	X	X
kafka.select-total	X	X
kafka.successful-authentication-rate	X	X
kafka.successful-authentication-total	X	X
kafka.sync-rate	X
kafka.sync-time-avg	X
kafka.sync-time-max	X
kafka.sync-total	X
kafka.waiting-threads		X

Spring Boot

In boot they add the prefix to the metrics, but with the Micronaut way of listening for all metrics, then adding them in both producer and consumer classes, I am still checking if there is a way to split them up and prefix.

Script used to generate the above data

//Consumer Only Metrics

def consumerMetrics = ["kafka.assigned-partitions",
                "kafka.bytes-consumed-rate",
                "kafka.bytes-consumed-total",
                "kafka.commit-latency-avg",
                "kafka.commit-latency-max",
                "kafka.commit-rate",
                "kafka.commit-total",
                "kafka.connection-close-rate",
                "kafka.connection-close-total",
                "kafka.connection-count",
                "kafka.connection-creation-rate",
                "kafka.connection-creation-total",
                "kafka.count",
                "kafka.failed-authentication-rate",
                "kafka.failed-authentication-total",
                "kafka.fetch-latency-avg",
                "kafka.fetch-latency-max",
                "kafka.fetch-rate",
                "kafka.fetch-size-avg",
                "kafka.fetch-size-max",
                "kafka.fetch-throttle-time-avg",
                "kafka.fetch-throttle-time-max",
                "kafka.fetch-total",
                "kafka.heartbeat-rate",
                "kafka.heartbeat-response-time-max",
                "kafka.heartbeat-total",
                "kafka.incoming-byte-rate",
                "kafka.incoming-byte-total",
                "kafka.io-ratio",
                "kafka.io-time-ns-avg",
                "kafka.io-wait-ratio",
                "kafka.io-wait-time-ns-avg",
                "kafka.io-waittime-total",
                "kafka.iotime-total",
                "kafka.join-rate",
                "kafka.join-time-avg",
                "kafka.join-time-max",
                "kafka.join-total",
                "kafka.last-heartbeat-seconds-ago",
                "kafka.network-io-rate",
                "kafka.network-io-total",
                "kafka.outgoing-byte-rate",
                "kafka.outgoing-byte-total",
                "kafka.records-consumed-rate",
                "kafka.records-consumed-total",
                "kafka.records-lag",
                "kafka.records-lag-avg",
                "kafka.records-lag-max",
                "kafka.records-lead",
                "kafka.records-lead-avg",
                "kafka.records-lead-min",
                "kafka.records-per-request-avg",
                "kafka.request-latency-avg",
                "kafka.request-latency-max",
                "kafka.request-rate",
                "kafka.request-size-avg",
                "kafka.request-size-max",
                "kafka.request-total",
                "kafka.response-rate",
                "kafka.response-total",
                "kafka.select-rate",
                "kafka.select-total",
                "kafka.successful-authentication-rate",
                "kafka.successful-authentication-total",
                "kafka.sync-rate",
                "kafka.sync-time-avg",
                "kafka.sync-time-max",
                "kafka.sync-total"]

//Producer Only Metrics

def producerMetrics = ["kafka.batch-size-avg",
                "kafka.batch-size-max",
                "kafka.batch-split-rate",
                "kafka.batch-split-total",
                "kafka.buffer-available-bytes",
                "kafka.buffer-exhausted-rate",
                "kafka.buffer-exhausted-total",
                "kafka.buffer-total-bytes",
                "kafka.bufferpool-wait-ratio",
                "kafka.bufferpool-wait-time-total",
                "kafka.byte-rate",
                "kafka.byte-total",
                "kafka.compression-rate",
                "kafka.compression-rate-avg",
                "kafka.connection-close-rate",
                "kafka.connection-close-total",
                "kafka.connection-count",
                "kafka.connection-creation-rate",
                "kafka.connection-creation-total",
                "kafka.count",
                "kafka.failed-authentication-rate",
                "kafka.failed-authentication-total",
                "kafka.incoming-byte-rate",
                "kafka.incoming-byte-total",
                "kafka.io-ratio",
                "kafka.io-time-ns-avg",
                "kafka.io-wait-ratio",
                "kafka.io-wait-time-ns-avg",
                "kafka.io-waittime-total",
                "kafka.iotime-total",
                "kafka.metadata-age",
                "kafka.network-io-rate",
                "kafka.network-io-total",
                "kafka.outgoing-byte-rate",
                "kafka.outgoing-byte-total",
                "kafka.produce-throttle-time-avg",
                "kafka.produce-throttle-time-max",
                "kafka.record-error-rate",
                "kafka.record-error-total",
                "kafka.record-queue-time-avg",
                "kafka.record-queue-time-max",
                "kafka.record-retry-rate",
                "kafka.record-retry-total",
                "kafka.record-send-rate",
                "kafka.record-send-total",
                "kafka.record-size-avg",
                "kafka.record-size-max",
                "kafka.records-per-request-avg",
                "kafka.request-latency-avg",
                "kafka.request-latency-max",
                "kafka.request-rate",
                "kafka.request-size-avg",
                "kafka.request-size-max",
                "kafka.request-total",
                "kafka.requests-in-flight",
                "kafka.response-rate",
                "kafka.response-total",
                "kafka.select-rate",
                "kafka.select-total",
                "kafka.successful-authentication-rate",
                "kafka.successful-authentication-total",
                "kafka.waiting-threads"]


//Producer & Consumer Metrics

def consumerAndProducerMetrics = ["kafka.assigned-partitions",
            "kafka.batch-size-avg",
            "kafka.batch-size-max",
            "kafka.batch-split-rate",
            "kafka.batch-split-total",
            "kafka.buffer-available-bytes",
            "kafka.buffer-exhausted-rate",
            "kafka.buffer-exhausted-total",
            "kafka.buffer-total-bytes",
            "kafka.bufferpool-wait-ratio",
            "kafka.bufferpool-wait-time-total",
            "kafka.byte-rate",
            "kafka.byte-total",
            "kafka.bytes-consumed-rate",
            "kafka.bytes-consumed-total",
            "kafka.commit-latency-avg",
            "kafka.commit-latency-max",
            "kafka.commit-rate",
            "kafka.commit-total",
            "kafka.compression-rate",
            "kafka.compression-rate-avg",
            "kafka.connection-close-rate",
            "kafka.connection-close-total",
            "kafka.connection-count",
            "kafka.connection-creation-rate",
            "kafka.connection-creation-total",
            "kafka.count",
            "kafka.failed-authentication-rate",
            "kafka.failed-authentication-total",
            "kafka.fetch-latency-avg",
            "kafka.fetch-latency-max",
            "kafka.fetch-rate",
            "kafka.fetch-size-avg",
            "kafka.fetch-size-max",
            "kafka.fetch-throttle-time-avg",
            "kafka.fetch-throttle-time-max",
            "kafka.fetch-total",
            "kafka.heartbeat-rate",
            "kafka.heartbeat-response-time-max",
            "kafka.heartbeat-total",
            "kafka.incoming-byte-rate",
            "kafka.incoming-byte-total",
            "kafka.io-ratio",
            "kafka.io-time-ns-avg",
            "kafka.io-wait-ratio",
            "kafka.io-wait-time-ns-avg",
            "kafka.io-waittime-total",
            "kafka.iotime-total",
            "kafka.join-rate",
            "kafka.join-time-avg",
            "kafka.join-time-max",
            "kafka.join-total",
            "kafka.last-heartbeat-seconds-ago",
            "kafka.metadata-age",
            "kafka.network-io-rate",
            "kafka.network-io-total",
            "kafka.outgoing-byte-rate",
            "kafka.outgoing-byte-total",
            "kafka.produce-throttle-time-avg",
            "kafka.produce-throttle-time-max",
            "kafka.record-error-rate",
            "kafka.record-error-total",
            "kafka.record-queue-time-avg",
            "kafka.record-queue-time-max",
            "kafka.record-retry-rate",
            "kafka.record-retry-total",
            "kafka.record-send-rate",
            "kafka.record-send-total",
            "kafka.record-size-avg",
            "kafka.record-size-max",
            "kafka.records-consumed-rate",
            "kafka.records-consumed-total",
            "kafka.records-lag",
            "kafka.records-lag-avg",
            "kafka.records-lag-max",
            "kafka.records-lead",
            "kafka.records-lead-avg",
            "kafka.records-lead-min",
            "kafka.records-per-request-avg",
            "kafka.request-latency-avg",
            "kafka.request-latency-max",
            "kafka.request-rate",
            "kafka.request-size-avg",
            "kafka.request-size-max",
            "kafka.request-total",
            "kafka.requests-in-flight",
            "kafka.response-rate",
            "kafka.response-total",
            "kafka.select-rate",
            "kafka.select-total",
            "kafka.successful-authentication-rate",
            "kafka.successful-authentication-total",
            "kafka.sync-rate",
            "kafka.sync-time-avg",
            "kafka.sync-time-max",
            "kafka.sync-total",
            "kafka.waiting-threads"]

//println "Overlapping " + "[\"${consumerMetrics.intersect(producerMetrics).join('",\n"')}\"]"
//
//def consumerClone = new ArrayList(consumerMetrics)
//consumerClone.removeAll(producerMetrics)
//
//println "Consumer Only " + "[\"${consumerClone.join('",\n"')}\"]"
//
//def producerClone = new ArrayList(producerMetrics)
//producerClone.removeAll(consumerMetrics)
//
//println "Producer Only " + "[\"${producerClone.join('",\n"')}\"]"
//
//println "\n\n\n"
println "| Name | Consumer | Producer |"
println "|------|----------|----------|"
consumerAndProducerMetrics.each {
    println "| $it | ${consumerMetrics.contains(it) ? "**X**":""} | ${producerMetrics.contains(it) ? "**X**":""} |"
}

[ctoestreich]

Holy !@#$ this is a lot of code to determine consumer or not https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/blob/master/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

[ctoestreich]

It would seem this is an issue with the larger KafkaMetrics and micrometer community as well.

micrometer-metrics/micrometer#1095
micrometer-metrics/micrometer#512