Status

Current state: Accepted

Under Discussion thread: here
Discussion

Vote thread: here

JIRA:

Motivation

All Kafka component components register AppInfo metrics to track the application details such as start time or and commit-id etc. These metrics are useful valuable for monitoring and debugging. However, the AppInfo doesn't provide AppInfo does not include the client-id for Worker and MM2 clients, which is an important piece of information for custom metrics reporter. reporters. While most Kafka clients (consumer, producer, admin) already register their client-id in the metrics config, there are cases in MM2 and Worker where AppInfo is registered without the client-id.

The AppInfoParser class registers a JMX MBean with the provided client-id, but when it adds adding metrics to the Metrics registry, the client-id is not included. This KIP aims to add proposes adding the client-id as a metric tag.

Public Interfaces

Currently MetricName for Worker and MM2 clients:

• [name=start-time-ms, group=app-info, description=Metric indicating start-time-ms, tags={}]
• [name=commit-id, group=app-info, description=Metric indicating commit-id, tags={}]
• [name=version, group=app-info, description=Metric indicating version, tags={}]

New MetricName for Worker and MM2 clients:

• [name=start-time-ms, group=app-info, description=Metric indicating start-time-ms, tags={client-id=<component-id>}]
• [name=commit-id, group=app-info, description=Metric indicating commit-id, tags={client-id=<component-id>}]
• [name=version, group=app-info, description=Metric indicating version, tags={client-id=<component-id>}]

...

private static void registerMetrics(Metrics metrics, AppInfo appInfo, String clientId) {
    if (metrics != null && clientId != null) {return;
    // Most Kafka clients metrics.addMetric(metricName(metrics, "version", Map.of("client-id", clientId)), new ImmutableValue<>(appInfo.getVersion()));(producer/consumer/admin) set the client-id tag in the metrics config.
    // Although we  metrics.addMetric(metricName(metrics, "commit-id", Map.of("don’t explicitly parse client-id" here, clientId)), new ImmutableValue<>(appInfo.getCommitId()));
    these metrics are automatically tagged with client-id.
    metrics.addMetric(metricName(metrics, "start-time-msversion", Map.of("client-id", clientId)), new ImmutableValue<>(appInfo.getStartTimeMs()));
    } else if (metrics != null) {
        metrics.addMetric(metricName(metrics, "version", Map.of()), new ImmutableValue<>( (Gauge<String>) (config, now) -> appInfo.getVersion()));
        metrics.addMetric(metricName(metrics, "commit-id", Map.of()), new(Gauge<String>) ImmutableValue<>((config, now) -> appInfo.getCommitId()));
        metrics.addMetric(metricName(metrics, "start-time-ms", Map.of()), new(Gauge<Long>) ImmutableValue<>((config, now) -> appInfo.getStartTimeMs()));
    }
}

2) When AppInfoParser registers an MBean, it will register both the new and the deprecated MBean.

public static synchronized void registerAppInfo(String prefix, String id, Metrics metrics, long nowMs) {
    try {// MirrorMaker/Worker doesn't set client-id tag into the metrics config, so we need to set it here.
        // skip...

        registerMetrics(metrics, mBean, null); // prefix will be added later by JmxReporterif (!metrics.config().tags().containsKey("client-id") && clientId != null) {
        registerMetricsmetrics.addMetric(metricName(metrics, mBean, id);
    } catch (JMException e) {
        log.warn("Error registering AppInfo mbean", e);
    }
}

3) When unregisters an MBean, also remove the new and deprecated MBean.

public static synchronized void unregisterAppInfo(String prefix, String id, Metrics metrics) {
    MBeanServer server = ManagementFactory.getPlatformMBeanServer()"version", Map.of("client-id", clientId)), (Gauge<String>) (config, now) -> appInfo.getVersion());
    try {
        ObjectName name = new ObjectName(prefix + ":type=app-info,id=" + Sanitizer.jmxSanitize(id));
        if (server.isRegistered(name))metrics.addMetric(metricName(metrics, "commit-id", Map.of("client-id", clientId)), (Gauge<String>) (config, now) -> appInfo.getCommitId());
            server.unregisterMBean(name);

        unregisterMetricsmetrics.addMetric(metricName(metrics, null);
        unregisterMetrics(metrics, id);
    } catch (JMException e) {
        log.warn("Error unregistering AppInfo mbean", e);
    } finally {
        log.info("App info {} for {} unregistered", prefix, id"start-time-ms", Map.of("client-id", clientId)), (Gauge<Long>) (config, now) -> appInfo.getStartTimeMs());
    }
}

2) When unregisters an MBean, also remove the new and deprecated MBean.

private static void unregisterMetrics(Metrics metrics, String clientId) {
    if (metrics != null && clientId != null) {return;

        metrics.removeMetric(metricName(metrics, "version", Map.of("client-id", clientId)));
        metrics.removeMetric(metricName(metrics, "commit-id", Map.of("client-id", clientId)));
        metrics.removeMetric(metricName(metrics, "start-time-ms", Map.of("client-id", clientId)));

    } else if (!metrics.config().tags().containsKey("client-id") && clientId != null) {
        metrics.removeMetric(metricName(metrics, "version", Map.of("client-id", clientId)));
        metrics.removeMetric(metricName(metrics, "commit-id", Map.of("client-id", clientId)));
        metrics.removeMetric(metricName(metrics, "start-time-ms", Map.of("client-id", clientId)));
    }
}

Compatibility, Deprecation, and Migration Plan

...

In Kafka 5.0, the following metric name will be deprecated removed. 

• [name=start-time-ms, group=app-info, description=Metric indicating start-time-ms, tags={}]
• [name=commit-id, group=app-info, description=Metric indicating commit-id, tags={}]
• [name=version, group=app-info, description=Metric indicating version, tags={}]

...