synchronized KafkaMetric registerMetric(KafkaMetric metric) {
        MetricName metricName = metric.metricName();
        KafkaMetric existingMetric = this.metrics.putIfAbsent(metricName, metric);
        if (existingMetric != null) {
            return existingMetric;
        }
        // newly added metric
        for (MetricsReporter reporter : reporters) {
            try {
                reporter.metricChange(metric);
            } catch (Exception e) {
                log.error("Error when registering metric on " + reporter.getClass().getName(), e);
            }
        }
        log.trace("Registered metric named {}", metricName);
        return null;
    }
Register a metric if not present or return the already existing metric with the same name.
When a metric is newly registered, this method returns null

@param metric The KafkaMetric to register
@return the existing metric with the same name or null
clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java