Versions Compared

Old Version 6

changes.mady.by.user Ken Huang

Saved on

compared with

New Version Current

changes.mady.by.user Ken Huang

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

Table of Contents

Status

Current state: Under Discussion Accepted

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

Vote thread: here

JIRA:

Jira
serverASF JIRA
columnIdsissuekey,summary,issuetype,created,updated,duedate,assignee,reporter,customfield_12311032,customfield_12311037,customfield_12311022,customfield_12311027,priority,status,resolution
columnskey,summary,type,created,updated,due,assignee,reporter,Priority,Priority,Priority,Priority,priority,status,resolution
serverId5aa69414-a9e9-3523-82ec-879b028fb15b
keyKAFKA-15186

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 As 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.

Public Interfaces

This approach would required modifying org.apache.kafka.common.utils.AppInfoParser.AppInfoMBean  interface to add a new method getClientId() .

Proposed Changes

The following code is a demonstration of how we will add to the getClientId method:

Code Block
public interface AppInfoMBean {
    String getVersion();
    String getCommitId();
    Long getStartTimeMs();
    String getClientId();
}

Also updated AppInfoParser to include a new field CLIENT_ID, so that when reading /kafka/kafka-version.properties, the clientId can be retrieved.

This KIP 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>}]

Proposed Changes

1) The new MBean will include a new tag, client-id

Code Block
private static void registerMetrics(Metrics metrics, AppInfo appInfo, String clientId) {
    if (metrics == null) return;
    // Most Kafka clients (producer/consumer/admin) set the client-id tag in the metrics config.
    // Although we don’t explicitly parse client-id here, these metrics are automatically tagged with client-id.
    metrics.addMetric(metricName(metrics, "version", Map.of()), (Gauge<String>) (config, now) -> appInfo.getVersion());
    metrics.addMetric(metricName(metrics, "commit-id", Map.of()), (Gauge<String>) (config, now) -> appInfo.getCommitId());
    metrics.addMetric(metricName(metrics, "start-time-ms", Map.of()), (Gauge<Long>) (config, now) -> appInfo.getStartTimeMs());
    // MirrorMaker/Worker doesn't set client-id tag into the metrics config, so we need to set it here.
    if (!metrics.config().tags().containsKey("client-id") && clientId != null) {
        metrics.addMetric(metricName(metrics, "version", Map.of("client-id", clientId)), (Gauge<String>) (config, now) -> appInfo.getVersion());
Code Block
languagejava
public class AppInfoParser {
    private static final Logger log = LoggerFactory.getLogger(AppInfoParser.class);
    private static final String VERSION;
    private static final String COMMIT_ID;
    private static final String CLIENT_ID;

    protected static final String DEFAULT_VALUE = "unknown";

    static {
        Properties props = new Properties();
        try (InputStream resourceStream = AppInfoParser.class.getResourceAsStream("/kafka/kafka-version.properties")) {
            props.load(resourceStream);
        } catch (Exception e) {
            log.warn("Error while loading kafka-version.properties: {}", e.getMessage());
        }
        VERSION = props.getProperty("version", DEFAULT_VALUE).trim();
        COMMIT_ID = props.getProperty("commitId", DEFAULT_VALUE).trim();
        CLIENT_ID = props.getProperty("clientId", DEFAULT_VALUE).trim();
    }

    public static String getVersion() {
        return VERSION;
    }

    public static String getCommitId() {
        return COMMIT_ID;
    }

    public static String getClientId() {
        return CLIENT_ID;
    }

    public static synchronized void registerAppInfo(String prefix, String id, Metrics metrics, long nowMs) {
        try {
            ObjectName name = new ObjectName(prefix + ":type=app-info,id=" + Sanitizer.jmxSanitize(id));
            MBeanServer server = ManagementFactory.getPlatformMBeanServer();
            if (server.isRegistered(name)) {
                log.info("The mbean of App info: [{}], id: [{}] already exists, so skipping a new mbean creation.", prefix, id);
                return;
            }
            DeprecatedAppInfo deprecatedMBean = new DeprecatedAppInfo(nowMs);
            AppInfo mBean = new AppInfo(nowMs);
            server.registerMBean(deprecatedMBean, name);
            server.registerMBean(mBean, name);

            registerMetrics(metrics, mBean); // prefix will be added later by JmxReporter
        } catch (JMException e) {
            log.warn("Error registering AppInfo mbean", e);
        }
    }

    public static synchronized void unregisterAppInfo(String prefix, String id, Metrics metrics) {
        MBeanServer server = ManagementFactory.getPlatformMBeanServer();
        try {
            ObjectName name = new ObjectName(prefix + ":type=app-info,id=" + Sanitizer.jmxSanitize(id));
            if (server.isRegistered(name))
                server.unregisterMBean(name);

            unregisterMetrics(metrics);
        } catch (JMException e) {
            log.warn("Error unregistering AppInfo mbean", e);
        } finally {
            log.info("App info {} for {} unregistered", prefix, id);
        }
    }

    private static MetricName metricName(Metrics metrics, String name) {
        return metrics.metricName(name, "app-info", "Metric indicating " + name);
    }

    private static void registerMetrics(Metrics metrics, AppInfoMBean appInfo) {
        if (metrics != null) {
            metrics.addMetric(metricName(metrics, "version"), new ImmutableValue<>(appInfo.getVersion()));
            metrics.addMetric(metricName(metrics, "commit-id", Map.of("client-id", clientId)), new(Gauge<String>) ImmutableValue<>((config, now) -> appInfo.getCommitId()));
            metrics.addMetric(metricName(metrics, "start-time-ms"), new ImmutableValue<>(appInfo.getStartTimeMs()));
            if (appInfo instanceof AppInfo) {
                metrics.addMetric(metricName(metrics, "client-id"), new ImmutableValue<>(appInfo.getClientId()));
            }
        }
    }

    Map.of("client-id", clientId)), (Gauge<Long>) (config, now) -> appInfo.getStartTimeMs());
    }
}

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

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

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

    public interface AppInfoMBean {
        String getVersion();
        String getCommitId();
        String getClientId();
        Long getStartTimeMs();
    }

    @Deprecated(since = "4.2")
    public static class DeprecatedAppInfo implements AppInfoMBean {

        private final Long startTimeMs;

        public DeprecatedAppInfo(long startTimeMsif (!metrics.config().tags().containsKey("client-id") && clientId != null) {
            this.startTimeMs = startTimeMs;
            log.info("Kafka version: {}", AppInfoParser.getVersion());
            log.info("Kafka commitId: {}", AppInfoParser.getCommitId());
            log.info("Kafka startTimeMs: {}", startTimeMs);
        }

        @Override
        public String getVersion() {
            return AppInfoParser.getVersion(metrics.removeMetric(metricName(metrics, "version", Map.of("client-id", clientId)));
        }

        @Override
        public String getCommitId() {
            return AppInfoParser.getCommitId();
        }

        @Override
        public String getClientId() {
            throw new UnsupportedOperationException("client-id is not implemented in DeprecatedAppInfo");
        }

        @Override
        public Long getStartTimeMs() {
            return startTimeMs;
        }

    }

    public static class AppInfo implements AppInfoMBean {

        private final Long startTimeMs;

        public AppInfo(long startTimeMs) {
            this.startTimeMs = startTimeMs;
            log.info("Kafka version: {}", AppInfoParser.getVersion(metrics.removeMetric(metricName(metrics, "commit-id", Map.of("client-id", clientId)));
            log.info("Kafka commitId: {}", AppInfoParser.getCommitId());
            log.info("Kafka startTimeMs: {}", startTimeMs);
            log.info("Kafka client id: {}", AppInfoParser.getClientId());
        }

        @Override
        public String getVersion() {
            return AppInfoParser.getVersion();
        }

        @Override
        public String getCommitId() {
            return AppInfoParser.getCommitId();
        }

        @Override
        public String getClientId() {
            return AppInfoParser.getClientId();
        }

        @Override
        public Long getStartTimeMs() {
            return startTimeMs;
        }

    }

    static class ImmutableValue<T> implements Gauge<T> {
        private final T value;

        public ImmutableValue(T value) {
            this.value = value;
        }

        @Override
        public T value(MetricConfig config, long now) {
            return value;
        }metrics.removeMetric(metricName(metrics, "start-time-ms", Map.of("client-id", clientId)));
    }
}

Compatibility, Deprecation, and Migration Plan

...

For users currently relying on the metric name without client-id tag, it will remain available until Kafka 5.0. New users are encouraged to use the new metric name, which includes the client-id tag. This deprecation will be documented in upgrade.html.

In Kafka 5.0, the following metric name will be 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={}]

Test Plan

change unit test or Integration test to verify the new method.

Rejected Alternatives

A new deprecated configuration could be introduced to control the registration of this MBean.

Nevertheless, this approach is considered unnecessary, as it would introduce complexity without significant benefit.n/a