Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java (Java)

Code before:

        this.numPartitionsActive = kafkaMetricName(
            NUM_PARTITIONS_METRIC_NAME,
            "The number of partitions in Active state.",
            "state", "active"
        );

        this.numPartitionsFailed = kafkaMetricName(
            NUM_PARTITIONS_METRIC_NAME,
            "The number of partitions in Failed state.",
            "state", "failed"
        );

        this.eventQueueSize = kafkaMetricName("event-queue-size", "The event accumulator queue size.");

        this.bufferCacheSize = kafkaMetricName(
            BATCH_BUFFER_CACHE_SIZE_METRIC_NAME,
            "The current total size in bytes of the append buffers being held in the coordinator's cache."
        );

        this.bufferCacheDiscardCount = kafkaMetricName(
            BATCH_BUFFER_CACHE_DISCARD_COUNT_METRIC_NAME,
            "The count of over-sized append buffers that were discarded instead of being cached upon release."
        );
        
        metrics.addMetric(bufferCacheDiscardCount, (Gauge<Long>) (config, now) -> bufferCacheDiscardCounter.get());
        metrics.addMetric(numPartitionsLoading, (Gauge<Long>) (config, now) -> numPartitionsLoadingCounter.get());
        metrics.addMetric(numPartitionsActive, (Gauge<Long>) (config, now) -> numPartitionsActiveCounter.get());
        metrics.addMetric(numPartitionsFailed, (Gauge<Long>) (config, now) -> numPartitionsFailedCounter.get());

        this.partitionLoadSensor = metrics.sensor(this.metricsGroup + "-PartitionLoadTime");
        this.partitionLoadSensor.add(
            metrics.metricName(
                "partition-load-time-max",
                this.metricsGroup,
                "The max time it took to load the partitions in the last 30 sec."
            ), new Max());
        this.partitionLoadSensor.add(
            metrics.metricName(
                "partition-load-time-avg",
                this.metricsGroup,
                "The average time it took to load the partitions in the last 30 sec."
            ), new Avg());

        this.threadIdleSensor = metrics.sensor(this.metricsGroup + "-ThreadIdleRatio");
        this.threadIdleSensor.add(
            metrics.metricName(
                "thread-idle-ratio-avg",
                this.metricsGroup,
                "The fraction of time the threads spent waiting for an event. This is an average across " +
                    "all coordinator event processor threads."),

Review comment (bug):
nit: Could we move this after `metrics.addMetric(numPartitionsFailed` to match the order of the field definitions?

Code after:

        this.numPartitionsActive = kafkaMetricName(
            NUM_PARTITIONS_METRIC_NAME,
            "The number of partitions in Active state.",
            "state", "active"
        );

        this.numPartitionsFailed = kafkaMetricName(
            NUM_PARTITIONS_METRIC_NAME,
            "The number of partitions in Failed state.",
            "state", "failed"
        );

        this.eventQueueSize = kafkaMetricName("event-queue-size", "The event accumulator queue size.");

        this.bufferCacheSize = kafkaMetricName(
            BATCH_BUFFER_CACHE_SIZE_METRIC_NAME,
            "The current total size in bytes of the append buffers being held in the coordinator's cache."
        );

        this.bufferCacheDiscardCount = kafkaMetricName(
            BATCH_BUFFER_CACHE_DISCARD_COUNT_METRIC_NAME,
            "The count of over-sized append buffers that were discarded instead of being cached upon release."
        );
        
        metrics.addMetric(numPartitionsLoading, (Gauge<Long>) (config, now) -> numPartitionsLoadingCounter.get());
        metrics.addMetric(numPartitionsActive, (Gauge<Long>) (config, now) -> numPartitionsActiveCounter.get());
        metrics.addMetric(numPartitionsFailed, (Gauge<Long>) (config, now) -> numPartitionsFailedCounter.get());
        metrics.addMetric(bufferCacheDiscardCount, (Gauge<Long>) (config, now) -> bufferCacheDiscardCounter.get());

        this.partitionLoadSensor = metrics.sensor(this.metricsGroup + "-PartitionLoadTime");
        this.partitionLoadSensor.add(
            metrics.metricName(
                "partition-load-time-max",
                this.metricsGroup,
                "The max time it took to load the partitions in the last 30 sec."
            ), new Max());
        this.partitionLoadSensor.add(
            metrics.metricName(
                "partition-load-time-avg",
                this.metricsGroup,
                "The average time it took to load the partitions in the last 30 sec."
            ), new Avg());

        this.threadIdleSensor = metrics.sensor(this.metricsGroup + "-ThreadIdleRatio");
        this.threadIdleSensor.add(
            metrics.metricName(
                "thread-idle-ratio-avg",
                this.metricsGroup,
                "The fraction of time the threads spent waiting for an event. This is an average across " +
                    "all coordinator event processor threads."),