Status

Current state: Under Discussion Accepted

Discussion thread: here 

JIRA: here

...

Motivation

With the delivery of KIP-714, cluster operators can now pull metrics from a broker (sent by clients) via a plugin. KIP KIP-714 dramatically simplifies observing client behavior via metrics.  Still, it only accounts for metrics from clients (admin, consumer, and producer), leaving out applications that embed client instances, such as Kafka Streams. While the importance of capturing client metrics is evident, there should be a mechanism for capturing the enclosing application metrics, as reviewing both sets will be required to get a complete performance picture. This proposal aims to provide a mechanism for applications containing Kafka clients to supplement the client metrics with some application-specific metrics.

...

In this KIP, we propose to add a two new method methods to the Admin, Consumer, and Producer  interfaces. The KafkaAdminClient, KafkaConsumer, and KafkaProducer  will implement this new methodmethods.

Proposed Changes

The method registerMetricsForSubscriptionmethods registerMetricForSubscription and ungregisterMetricFromSubscription  will be added to the Admin, Consumer,  and Producer  interfaces. Because of the ad hoc connections of the AdminClient this KIP will also set the default value of the AdminClient.ENABLE_METRICS_PUSH_CONFIG to false.

package org.apache.kafka.clients.admin;

public interface Admin extends AutoCloseable {  

/**
 *  ApplicationAn application metricsmetric provided for subscription.
 *  TheseThis metricsmetric will be added to this client's metrics
 *  that are available for subscription and sent as 
 *  telemetry data to the broker.
 *
 * @param metricsmetric, the application metricsmetric to register
 */ 
void registerMetricForSubscription(KafkaMetric metric)

/**
 *  @returnAn trueapplication ifto metricsbe acceptedremoved forfrom subscription.
 *  This metric is removed from this client's metrics
 *  and will not be Aavailable returnfor value of false indicates that the ENABLE_METRICS_PUSH_CONFIG is set to false.subscription.
 *
 * @param metric, the application metric to remove
 */ 
booleanvoid registerMetricsForSubscriptionunregisterMetricFromSubscription(Collection<? extends Metric> metrics)KafkaMetric metric)

package org.apache.kafka.clients.consumer;

public interface Consumer<K, V> extends Closeable {
     

/**
 *  ApplicationAn application metricsmetric provided for subscription.
 *  TheseThis metricsmetric will be added to this client's metrics
 *  that are available for subscription and sent as 
 *  telemetry data to the broker.
 *
 * @param metricsmetric, the application metricsmetric to register
 */ 
void registerMetricForSubscription(KafkaMetric metric)

/**
 *  @returnAn trueapplication ifto metricsbe acceptedremoved forfrom subscription.
 *  This metric is removed from this client's metrics
 *  and will not be Aavailable returnfor value of false indicates that the ENABLE_METRICS_PUSH_CONFIG is set to false.subscription.
 *
 * @param metric, the application metric to remove
 */   
booleanvoid registerMetricsForSubscriptionunregisterMetricFromSubscription(Collection<? extends Metric> metricsKafkaMetric metric)

package org.apache.kafka.clients.producer;

public interface Producer<K, V> extends Closeable {     

/**
 *  ApplicationAn application metricsmetric provided for subscription.
 *  TheseThis metricsmetric will be added to this client's metrics
 *  that are available for subscription and sent as 
 *  telemetry data to the broker.
 *
 * @param metricsmetric, the application metricsmetric to register
 */ 
void registerMetricForSubscription(KafkaMetric metric)

/**
 *  @returnAn trueapplication ifto metricsbe acceptedremoved forfrom subscription.
 *  This metric is removed from this client's metrics
 *  and will not be Aavailable returnfor value of false indicates that the ENABLE_METRICS_PUSH_CONFIG is set to false.subscription.
 *
 * @param metric, the application metric to remove
 */  
booleanvoid registerMetricsForSubscriptionunregisterMetricFromSubscription(Collection<? extends Metric> metricsKafkaMetric metric)

Metrics naming

...

Since this KIP uses the entire telemetry metrics pipeline of KIP-714, the metrics naming and formatting will leverage the same process and rules found in KIP-714. 

...

Kafka Streams metric name	Telemetry metric name
"process-records-avg, group=stream-thread-metrics"	"org.apache.kafka.stream.thread.process.records.avg"
"commit-latency-max, group=stream-thread-metrics"	"org.apache.kafka.stream.thread.commit.latency.max"

Metric format

Since this proposed KIP is an extension of KIP-714, it will follow the same approach for supported metric types.  When providing metrics via registerMetricsForSubscription, only metrics of type sum (a monotonic counter, like process-total) and gauge (a non-monotonic value, like process-rate) are registered, other types are dropped generating a log WARN statement.   

Configuration

With KIP-714, clients use the ENABLE_METRICS_PUSH_CONFIG configuration to enable pushing metrics.  Kafka Streams also has the ENABLE_METRICS_PUSH_CONFIG  configuration, which is meant to extend to the internal client's ability to push metrics.

...

It's important to note that since Kafka Streams leverages the client metrics mechanism, disabling metrics for internal clients removes the possibility of pushing Kafka Streams metrics.  In practice, this is not an anticipated scenario, but in the interest of completeness, it's worth specifying it here in the KIP

To alert the user to this significant configuration mismatch, a Kafka Streams application configured with ENABLE_METRICS_PUSH_CONFIG=true and a main or admin consumer configured with ENABLE_METRICS_PUSH_CONFIG=false, will result in ConfigException  on start-up.

Compatibility, Deprecation, and Migration Plan

...