Status

Current state: Under Discussion

Discussion thread: here

JIRA: here

Motivation

When Kafka Streams uses the Streams rebalance protocol, each stream thread receives StreamsGroupHeartbeatResponse messages from the group coordinator. These responses may include statuses such as MISSING_SOURCE_TOPICS , MISSING_INTERNAL_TOPICS ,   INCORRECTLY_PARTITIONED_TOPICS , ASSIGNMENT_DELAYED , or SHUTDOWN_APPLICATION .

Today, these statuses are used internally by the Streams client to determine group readiness and handle errors, but they are not directly exposed as metrics. In some cases, operators must inspect client logs to understand why a specific stream thread is not processing, why the group is not ready, or why assignment is delayed.

This becomes more expensive in environments with many Kafka Streams clients and many stream threads. Even within the same application, different stream threads may temporarily observe different heartbeat responses. For example, one thread may observe ASSIGNMENT_DELAYED , while another may not. Application-level or client-level state alone does not identify which stream thread is affected by which coordinator-reported condition.

As a result, operators may not be able to narrow down the cause from dashboards or alerts alone. They must identify the affected client and thread, then inspect logs. This increases operational cost and can delay root-cause analysis and recovery.

This KIP proposes exposing the latest Streams group heartbeat statuses observed by each stream thread as Kafka Streams thread-level metrics. This allows operators to identify coordinator-reported conditions directly from metrics. 

Public Interfaces

This KIP adds the following Kafka Streams thread-level metric.

• A single StreamsGroupHeartbeatResponse  may contain multiple statuses. Therefore, multiple heartbeat-status metrics for the same stream thread may have value 1 at the same time, each with a different status tag value.
• The status tag value is derived from StreamsGroupHeartbeatResponse.Status by converting the enum name to lower-case hyphen-separated form.
  • stale-topology
  • missing-source-topics
  • incorrectly-partitioned-topics
  • missing-internal-topics

  • missing-client-tags

  • shutdown-application
  • assignment-delayed

For example, StreamsGroupHeartbeatResponse.Status.MISSING_SOURCE_TOPICS  is exposed as status=missing-source-topics.
This metric is registered only for stream threads using the Streams rebalance protocol. When the classic rebalance protocol is used, this metric is not registered.

Proposed Changes

Kafka Streams will expose the latest Streams group heartbeat statuses observed by each stream thread as thread-level metrics.
For each stream thread using the Streams rebalance protocol, Kafka Streams will register one metric per known Streams group heartbeat status.
Each metric reports whether the latest successful StreamsGroupHeartbeatResponse observed by that stream thread contains the corresponding status.

This metric is a Gauge. Its value is evaluated when the metric is read.

• 1 = the latest successful StreamsGroupHeartbeatResponse  contains the status
• 0 = the latest successful StreamsGroupHeartbeatResponse  does not contain the status

A single StreamsGroupHeartbeatResponse may contain multiple statuses. Therefore, multiple status metrics for the same stream thread may return 1 at the same time.

This metric is scoped to a stream thread because Streams group heartbeats are sent and received by individual stream threads. As a result, different stream threads may temporarily report different heartbeat statuses.

The status tag value is derived from StreamsGroupHeartbeatResponse.Status  by converting the enum name to lower-case hyphen-separated form. For example, INCORRECTLY_PARTITIONED_TOPICS  is exposed as incorrectly-partitioned-topics.

Kafka Streams will not expose statusDetail as part of this metric. The statusDetail  field may contain free-form or high-cardinality values, such as topic names or detailed error messages, and is therefore not suitable as a metric tag or metric value. Kafka Streams will register this metric only for stream threads using the Streams rebalance protocol (KIP-1071). When the classic rebalance protocol is used, this metric is not registered.

Thread safety

The consumer background thread processes Streams group heartbeat responses and publishes the latest heartbeat status collection to StreamsRebalanceData through an AtomicReference. The status gauge may be evaluated by a different thread, such as a JMX request-handling thread, a MetricsReporter thread, or an application thread reading Kafka Streams metrics. The gauge reads the status collection from the same AtomicReference.

AtomicReference provides atomic replacement and the required memory visibility between the consumer background thread and metric-reading threads. A metric read observes either the previously published collection or the newly published collection and does not observe a partially published reference.

Edge case and failure behavior

The heartbeat-status gauge values are defined as follows.

A failed heartbeat does not provide a new status snapshot and therefore does not change the gauge values.

Compatibility, Deprecation, and Migration Plan

This change is backward compatible.
It only adds new metrics. No existing metric names, tags, or values are changed. No public Java API changes are required.

Test Plan

• Verify that ThreadMetrics registers the new status Gauge metrics with the expected group, name, and tags.
• Verify that StreamThread exposes 1 for statuses present in StreamsRebalanceData and 0 for absent statuses.
• Verify that when a heartbeat response contains multiple statuses, multiple status metrics for the same stream thread may return 1 at the same time.
• Verify that the status tag value is exposed in lower-case hyphen-separated form.
• Verify that the metric is not registered when the Streams rebalance protocol is not used.
• Verify that the new metric is removed as part of existing thread-level metric cleanup.

Rejected Alternatives

This KIP does not expose statusDetail as a metric tag or value. statusDetail may contain high-cardinality values such as topic names or detailed error messages, which could create unstable metric identities and excessive time series.

Cumulative status counters

A cumulative counter for each heartbeat status was considered as a way to preserve statuses that appear and clear between metric reads. A response-based counter would count the number of successful heartbeat responses containing a status. However, this value would depend on the configured heartbeat interval and would not represent the number of distinct incidents or status transitions. A transition-based counter would require maintaining and defining additional transition state. this KIP is intended to expose the latest known heartbeat statuses for diagnosing the current state of a stream thread, rather than to provide a history of status occurrences. Therefore, this KIP exposes gauges only. A status that appears and clears entirely between metric reads may not be observed.