This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state:  [One of "Under Discussion", " Accepted", "Rejected"]

Discussion thread: here 

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

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Describe the problems you are trying to solve.

Public Interfaces

Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.

A public interface is any change to the following:

• Binary log format

• The network protocol and api behavior

• Any class in the public packages under clientsConfiguration, especially client configuration

  • org/apache/kafka/common/serialization

  • org/apache/kafka/common

  • org/apache/kafka/common/errors

  • org/apache/kafka/clients/producer

  • org/apache/kafka/clients/consumer (eventually, once stable)

• Monitoring

• Command line tools and arguments

• Anything else that will likely break existing users in some way when they upgrade

Proposed Changes

...

KAFKA-19773 

Motivation

KIP-714 introduced the ClientTelemetryReceiver interface to enable MetricsReporter instances running on brokers to collect client telemetry metrics. Typically a reporter stores metrics it receives in a "metrics registry" and exposes them in a format suitable for the external monitoring system it is built for. This is exactly what JmxReporter does. Kafka clients and brokers explicitly delete their metrics when they are not used anymore. However for client telemetry metrics there isn't an explicit deletion mechanism.

Clients send their telemetry metrics at a regular interval, configured via telemetry subscriptions. Reporters are not aware of this push interval. Without the interval and an explicit deletion mechanism, reporters don't know when to clear client telemetry metrics and stop exposing them. This makes it hard to not expose stale metrics. Having uncertainty in metrics is a major problem as operators rely on them to manage their clusters and investigate issues.

This KIP proposes updating the context associated with each client telemetry push to include the push interval. This will enable reporters to properly manage the life cycle of client telemetry metrics.

Public Interfaces

• A new interface ClientTelemetryContext to hold the request and telemetry (just the push interval) context:

  Code Block
  language java title ClientTelemetryContext
  package org.apache.kafka.server.telemetry; import org.apache.kafka.server.authorizer.AuthorizableRequestContext; /** * Context provided to {@link ClientTelemetryExporter} implementations when receiving client metrics. */ public interface ClientTelemetryContext { /** * The interval defined via <code>metrics.interval</code> in the client metrics subscription * @return The interval in milliseconds */ int pushIntervalMs(); /** * The context associated with this request * @return The AuthorizableRequestContext associated with this request */ AuthorizableRequestContext authorizableRequestContext(); }

  
  

• A new interface, ClientTelemetryExporter, that works like ClientTelemetryReceiver but uses ClientTelemetryContext.

  Code Block
  language java
  package org.apache.kafka.server.telemetry; /** * {@code ClientTelemetryExporter} defines the behaviour for telemetry exporter on the broker side * which receives client telemetry metrics. */ public interface ClientTelemetryExporter { /** * Called by the broker when a client reports telemetry metrics. The associated telemetry context * can be used by the metrics plugin to retrieve additional client information such as client ids, * endpoints or the push interval. * <p> * This method may be called from the request handling thread, and as such should avoid blocking. * * @param context the client telemetry context for the corresponding {@code PushTelemetryRequest} * api call. * @param payload the encoded telemetry payload as sent by the client. */ void exportMetrics(ClientTelemetryContext context, ClientTelemetryPayload payload); }

  
  

• A new interface, ClientTelemetryExporterProvider, that works like ClientTelemetry but provides a ClientTelemetryExporter instance instead of ClientTelemetryReceiver.

  Code Block
  language java
  import org.apache.kafka.common.metrics.MetricsReporter; /** * A {@link MetricsReporter} may implement this interface to indicate support for collecting client * telemetry on the server side. */ public interface ClientTelemetryExporterProvider { /** * Called by the broker to fetch instance of {@link ClientTelemetryExporter}. * <p> * This instance may be cached by the broker. * * @return broker side instance of {@link ClientTelemetryExporter}. */ ClientTelemetryExporter clientTelemetryExporter(); }

  
  

Proposed Changes

• Deprecate the existing ClientTelemetryReceiver and ClientTelemetry interfaces and recommend users to instead use ClientTelemetryExporter and ClientTelemetryExporterProvider respectively.

• Brokers will support both ClientTelemetry and ClientTelemetryExporterProvider implementations, and call the appropriate methods depending on the type of class provided.
• If a class implements both ClientTelemetry and ClientTelemetryExporterProvider, only the methods from ClientTelemetryExporterProvider, which is the new class, will be used.

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

...

The existing interfaces, ClientTelemetryReceiver and ClientTelemetry, are kept and deprecated. They will be removed in the next major version, Kafka 5.0.0. All existing implementations will keep working until then.

To use the new features, implementations will need to implement the new interfaces, ClientTelemetryExporterProvider and ClientTelemetryExporter.

Test Plan

Adequate unit and integration tests will be added to ensure ClientTelemetry/ClientTelemetryReceiver implementations keep working and implementations of the new class work as well.

Rejected Alternatives

• Add an override of ClientTelemetryReceiver.exportMetrics() that accepts ClientTelemetryContext: This would require implementations wanting to use ClientTelemetryContext to still implement the old ClientTelemetryReceiver.exportMetrics() method. If we want to remove of the old method in the future, implementations would need to be recompiled at that time too.