KAFKA-19684: Move Gauge#value to MetricValueProvider

[majialoong]

KIP-188 introduced MetricValueProvider, adding Measurable and Gauge as its sub interfaces. However, this left a legacy issue: move the value method from Gauge to the super interface, MetricValueProvider.

This PR moves the value method from Gauge to MetricValueProvider and provides a default implementation in Measurable. This unifies the methods used by Gauge and Measurable to obtain monitoring values.

[majialoong]

For binary compatibility issues, the following tests were performed：

1. Binary compatibility tested using japicmp

kafka-clients-4.2.0-SNAPSHOT.jar is the code before this PR, and kafka-clients-4.2.0-SNAPSHOT-new.jar contains the changed code of this PR.

2. Write a test code to verify

import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Gauge;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.stats.Min;

public class CustomMetrics {
    public static void main(String[] args) {
        Metrics metrics = new Metrics();

        MetricName gaugeMetricNameOne = new MetricName("gaugeMetricNameOne", "test-metrics", "description for gauge "
                + "metric "
                + "one",
                new HashMap<>());
        metrics.addMetric(gaugeMetricNameOne, (Gauge<String>) (config, now) -> "hello gauge.");

        MetricName gaugeMetricNameTwo = new MetricName("gaugeMetricNameTwo", "test-metrics", "description for gauge "
                + "metric two",
                new HashMap<>());
        metrics.addMetric(gaugeMetricNameTwo, (Gauge<Integer>) (config, now) -> 123);

        MetricName measurableMetricNameOne = new MetricName("measurableMetricNameOne", "test-metrics", "description for "
                + " measurable metric one",
                new HashMap<>());
        metrics.addMetric(measurableMetricNameOne, (config, now) -> 100L);

        MetricName measurableMetricNameTwo = new MetricName("measurableMetricNameTwo", "test-metrics", "description "
                + "for measurable metric two",
                new HashMap<>());
        Min min = new Min();
        long windowMs = 100;
        int samples = 2;
        MetricConfig config = new MetricConfig().timeWindow(windowMs, TimeUnit.MILLISECONDS).samples(samples);
        min.record(config, 50, System.currentTimeMillis());
        metrics.addMetric(measurableMetricNameTwo, min);

        System.out.println("gaugeMetricNameOne: " + metrics.metric(gaugeMetricNameOne).metricValue());
        System.out.println("gaugeMetricNameTwo: " + metrics.metric(gaugeMetricNameTwo).metricValue());
        System.out.println("measurableMetricNameOne: " + metrics.metric(measurableMetricNameOne).metricValue());
        System.out.println("measurableMetricNameTwo: " + metrics.metric(measurableMetricNameTwo).metricValue());
    }
}

Compile using the code in the current trunk branch and version 1.0.1. Then run using the code that includes the changes in this PR.

trunk beanch :

1.0.1 version :

[majialoong]

Hello, @ijuma @rajinisivaram @chia7712 , when have time, please review this PR , thanks !

[chia7712]

@majialoong thanks for this patch!

[chia7712]

Can we update the docs so they match these changes?

[majialoong]

Thanks for the reminder, I will update this doc.

[chia7712]

The Gauge type seems redundant at this stage. Would it be worth filing a KIP to deprecate it?

[majialoong]

Keeping the Gauge type can improves code readability. Readers immediately understand that the monitoring metric represents an instantaneous value, and it can also be equated with the concept of Gauge in mainstream monitoring systems. Directly using the MetricValueProvider may be a bit too abstract.

[chia7712]

Another follow-up (KIP) could be removing addMetric(MetricName metricName, Measurable measurable), which would let us drop the explicit type from the lambda
before

metrics.addMetric(metricName(metrics, "version", Map.of()), (Gauge<String>) (config, now) -> appInfo.getVersion());

after

metrics.addMetric(metricName(metrics, "version", Map.of()),  (config, now) -> appInfo.getVersion());

[majialoong]

This is a good suggestion, I've created a JIRA to track this issue：https://issues.apache.org/jira/browse/KAFKA-19729

[ijuma]

Thanks for the PR. Regarding compatibility, you'd want to test both binary and source compatibility.

[majialoong]

Thanks for the PR. Regarding compatibility, you'd want to test both binary and source compatibility.

@ijuma Thank you for your comments. In the above conversation, I listed two methods for testing binary compatibility. If there are any scenarios I've overlooked, please comment and add them. I'll test and verify them promptly. I will also conduct some further testing and verification and add to this in subsequent conversations.

[majialoong]

I performed the following tests and found no binary or source incompatibilities.

To test compatibility, I wrote the following test code, which includes code demonstrating the usage of Gauge and Measurable.

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Gauge;
import org.apache.kafka.common.metrics.Measurable;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.stats.Min;

import java.util.HashMap;
import java.util.concurrent.TimeUnit;

public class TestCompatibility {
    public static void main(String[] args) {
        Metrics metrics = new Metrics();

        // Kafka about MetricValueProvider have two sub-interface Gauge and Measurable

        // test Gauge
        MetricName gaugeMetricNameOne = new MetricName("gaugeMetricNameOne", "test-metrics", "description for gauge "
                + "metric "
                + "one",
                new HashMap<>());
        metrics.addMetric(gaugeMetricNameOne, (Gauge<String>) (config, now) -> "hello gauge.");

        MetricName gaugeMetricNameTwo = new MetricName("gaugeMetricNameTwo", "test-metrics", "description for gauge "
                + "metric two",
                new HashMap<>());
        metrics.addMetric(gaugeMetricNameTwo, (Gauge<Integer>) (config, now) -> 123);

        Gauge<String> stringGauge = (config, now) -> "string gauge";
        MetricName gaugeMetricNameThree = new MetricName("gaugeMetricNameThree", "test-metrics", "description for gauge "
                + "metric three",
                new HashMap<>());
        metrics.addMetric(gaugeMetricNameThree, stringGauge);

        System.out.println("gaugeMetricNameOne: " + metrics.metric(gaugeMetricNameOne).metricValue());
        System.out.println("gaugeMetricNameTwo: " + metrics.metric(gaugeMetricNameTwo).metricValue());
        System.out.println("gaugeMetricNameThree: " + metrics.metric(gaugeMetricNameThree).metricValue());

        // test Measurable
        MetricName measurableMetricNameOne = new MetricName("measurableMetricNameOne", "test-metrics", "description for "
                + " measurable metric one",
                new HashMap<>());
        metrics.addMetric(measurableMetricNameOne, (config, now) -> 100L);

        MetricName measurableMetricNameTwo = new MetricName("measurableMetricNameTwo", "test-metrics", "description "
                + "for measurable metric two",
                new HashMap<>());

        Measurable measurable = (Measurable) (config, now) -> 200L;
        metrics.addMetric(measurableMetricNameTwo, measurable);

        MetricName measurableMetricNameThree = new MetricName("measurableMetricNameThree", "test-metrics", "description "
                + "for measurable metric three",
                new HashMap<>());

        // test Measurable implementation class Min
        Min min = new Min();
        long windowMs = 100;
        int samples = 2;
        MetricConfig config = new MetricConfig().timeWindow(windowMs, TimeUnit.MILLISECONDS).samples(samples);
        min.record(config, 50, System.currentTimeMillis());
        metrics.addMetric(measurableMetricNameThree, min);

        System.out.println("measurableMetricNameOne: " + metrics.metric(measurableMetricNameOne).metricValue());
        System.out.println("measurableMetricNameTwo: " + metrics.metric(measurableMetricNameTwo).metricValue());
        System.out.println("measurableMetricNameThree: " + metrics.metric(measurableMetricNameThree).metricValue());
    }
}

I have prepared three versions of the Kafka-clients JAR files :

1. kafka-clients-1.0.1.jar（The Gauge and MetricValueProvider interfaces have been added（1.0.0 version）, and KafkaMetric first time as public type（1.0.1 version）.）
2. kafka-clients-4.2.0-SNAPSHOT.jar（trunk branch code）
3. kafka-clients-4.2.0-SNAPSHOT-new.jar （with this PR code）

Regarding source code compatibility, I tested the code using the three different versions of the JAR file, and it compiled successfully in all three cases.

Regarding binary compatibility, I compiled the test code using each of these three different versions of the JAR file, and all of them ran successfully in all three versions.

1. Compile using version kafka-clients-1.0.1.jar，and run directly using the three versions of the JAR files.

2. Compile using version kafka-clients-4.2.0-SNAPSHOT.jar，and run directly using the three versions of the JAR files.

3. Compile using version kafka-clients-4.2.0-SNAPSHOT-new.jar，and run directly using the three versions of the JAR files.

Regarding binary compatibility testing, we also used japicmp for verification.

1. Use the libs files generated by compiling the code included in this PR. The binary compatibility check using japicmp shows no changes.

3. And use libs files were compiled using code from the trunk branch. The binary compatibility check using japicmp shows no changes.