Repository: apache/kafka
File: metadata/src/main/java/org/apache/kafka/image/loader/metrics/MetadataLoaderMetrics.java (Java)

Code before:
            @Override
            public Integer value() {
                return currentControllerId();
            }
        }));
        registry.ifPresent(r -> r.newGauge(HANDLE_LOAD_SNAPSHOT_COUNT, new Gauge<Long>() {
            @Override
            public Long value() {
                return handleLoadSnapshotCount();
            }
        }));
        registry.ifPresent(r -> r.newGauge(AVERAGE_IDLE_RATIO, new Gauge<Double>() {
            @Override
            public Double value() {
                synchronized (avgIdleTimeRatio) {
                    return avgIdleTimeRatio.measure();
                }
            }
        }));
    }

    public void updateIdleTime(long idleDurationMs) {
        synchronized (avgIdleTimeRatio) {
            avgIdleTimeRatio.record((double) idleDurationMs, time.milliseconds());
        }
    }

    private void addFinalizedFeatureLevelMetric(String featureName) {
        registry.ifPresent(r -> r.newGauge(
            getFeatureNameTagMetricName(
                "MetadataLoader",
                FINALIZED_LEVEL_METRIC_NAME,
                featureName
            ),
            new Gauge<Short>() {
                @Override
                public Short value() {
                    return finalizedFeatureLevel(featureName);
                }
            }
        ));
    }

    private void removeFinalizedFeatureLevelMetric(String featureName) {
        registry.ifPresent(r -> r.removeMetric(
            getFeatureNameTagMetricName(
                "MetadataLoader",
                FINALIZED_LEVEL_METRIC_NAME,
                featureName
            )
        ));

Review comment (suggestion):
I didn't notice this in the previous PR but the signature should be `void updateIdleTime(long idleDurationMs, long currentTimeMs)`. Otherwise the current time use to compute the idle time is different from the current time used by `TimeRatio`.

Can you fix this in this PR? This mean that KafkaEventQueue should take a `BiConsumer<Long, Long>`

Code after:
        registry.ifPresent(r -> r.newGauge(HANDLE_LOAD_SNAPSHOT_COUNT, new Gauge<Long>() {
            @Override
            public Long value() {
                return handleLoadSnapshotCount();
            }
        }));
        registry.ifPresent(r -> r.newGauge(AVERAGE_IDLE_RATIO, new Gauge<Double>() {
            @Override
            public Double value() {
                synchronized (avgIdleTimeRatio) {
                    return avgIdleTimeRatio.measure();
                }
            }
        }));
    }

    public void updateIdleTime(long idleDurationMs, long currentTimeMs) {
        synchronized (avgIdleTimeRatio) {
            avgIdleTimeRatio.record((double) idleDurationMs, currentTimeMs);
        }
    }

    private void addFinalizedFeatureLevelMetric(String featureName) {
        registry.ifPresent(r -> r.newGauge(
            getFeatureNameTagMetricName(
                "MetadataLoader",
                FINALIZED_LEVEL_METRIC_NAME,
                featureName
            ),
            new Gauge<Short>() {
                @Override
                public Short value() {
                    return finalizedFeatureLevel(featureName);
                }
            }
        ));
    }

    private void removeFinalizedFeatureLevelMetric(String featureName) {
        registry.ifPresent(r -> r.removeMetric(
            getFeatureNameTagMetricName(
                "MetadataLoader",
                FINALIZED_LEVEL_METRIC_NAME,
                featureName
            )
        ));
    }

    /**
     * Update the batch processing time histogram.
     */