Status

Current state: Draft Voting in progress

Discussion thread: thread: https://lists.apache.org/thread/z4o5093zqxmc18nzb74vyhvb1hb399v4

Vote thread: https://lists.apache.org/thread/jxf789v0kdpvsb9wsosghh57odkksxt7

JIRA:

...

The Kafka consumer allows users to `pausepause()`   and `resumeresume()`   individual partitions. When a partition is paused, subsequent calls to `pollpoll()`   will not return any records from that partition until it is resumed. This mechanism is used for application-level flow control and backpressure management.

...

• Backpressure: When an internal record buffer for a partition exceeds its capacity, Kafka Streams pauses that partition to prevent further fetching (`StreamTask. addRecords()`). It resumes the partition once the buffer drains below the threshold (`StreamTask.resumePollingForPartitionsWithAvailableSpace()`).
• Rebalance lifecycle: After a rebalance, Kafka Streams pauses all partitions that are not yet owned by fully initialized tasks (`TaskManager`). Partitions are only resumed after state store restoration completes.
• Changelog restore management: The `StoreChangelogReader` pauses StoreChangelogReader pauses and resumes changelog partitions on the restore consumer depending on whether the corresponding tasks still exist.

...

1. How many partitions are currently paused? A high number of paused partitions may indicate backpressure issues, slow state restoration, or application-level processing bottlenecks.
2. Which specific partitions are paused? Knowing which partitions are paused helps pinpoint which topics or partitions are experiencing issues.
3. How long has a partition been paused? A partition that has been paused for an extended period may indicate a stuck consumer, an unrecoverable error in processing, or a state store restoration that is taking too long.
4. How frequently are partitions being paused? A high rate of pause calls may indicate instability in the application's flow control logic or repeated rebalances triggering pause/resume cycles.

This KIP proposes adding new consumer metrics that track paused partitions and their pause duration.

Public Interfaces

New metrics

consumer-coordinator-metrics

The metric will have the following tags:

• client-id

consumer-fetch-manager-metrics

Each metric will have the following tags:

• client-id
• topic
• partition

Note: These are per-partition metrics, consistent with existing per-partition consumer metrics such as records-lag  and records-lead . Cardinality increases linearly with the number of assigned partitions, which may increase memory usage in the metrics registry and in downstream monitoring systems.

Proposed Changes

Track Pause Timestamp

Currently, each partition's internal state tracks whether it is paused with a simple boolean flag. To support the `partitionpaused-pausedpartitions-timeduration-ms` seconds  metric, this KIP adds a timestamp that records when the partition was paused. The timestamp is set when `pausepause()`   is called and cleared (reset to `-1`) when `resume, cleared when resume() ` is called. This allows the metric layer to compute the elapsed pause duration on demand, without requiring a background timer or periodic sampling.

consumer-metrics

, and also cleared on partition reassignment regardless of prior pause status. This ensures that a partition re-assigned to the same consumer after a rebalance starts with a clean pause state.

consumer-coordinator-metrics

A paused-partitions / paused-partitions-duration-seconds-max  gauges and paused-partitions-rate / paused-partitions-total sensors are registered at coordinator level. When queried, these gauges compute A `paused-partitions-count` gauge is registered in the consumer's top-level metrics. When queried, this gauge computes the current count of paused partitions and max paused duration by reading from the consumer's subscription state. Since the consumer already maintains the set of paused partitions internally, no additional data structures are needed for this metricSensor metrics are updated on each pause() call, incrementing the cumulative count and contributing to the rate calculation.

consumer-fetch-manager-metrics

The `partition-paused` and `partition-paused-time-ms` gauges paused-partition / paused-partition-duration-seconds gauges and paused-partition-rate / paused-partition-total sensors are registered per partition in the fetch manager metrics, alongside the existing per-partition lag and lead metrics. They follow the same lifecycle:

• On partition assignment: The metrics are registered for each newly assigned partition.
• On partition revocation: The metrics are removed for each revoked partition, preventing stale metrics from accumulating. Both Gauge metrics are implemented as on-demand gauges that read directly from the partition's pause state when queried, rather than being updated on every pause/resume call.

...

• Verify that the pause timestamp is set when `pausepause()`   is called, cleared on `resumeresume()` , and reset on partition reassignment.
• Verify that the `paused-partitions-count` metric is coordinator-level metrics are registered and returns the correct count as partitions are paused and resumed.
• Verify that `partition-paused` and `partition-paused-time-ms` per-partition metrics are registered per partition, return correct values, and are removed when partitions are revoked.

...