Page History

...

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 (`StreamTaskStreamTask.addRecords()`). It resumes the partition once the buffer drains below the threshold (`StreamTaskStreamTask.resumePollingForPartitionsWithAvailableSpace()`).
• Rebalance lifecycle: After a rebalance, Kafka Streams pauses all partitions that are not yet owned by fully initialized tasks (`TaskManager`TaskManager). Partitions are only resumed after state store restoration completes.
• Changelog restore management: The `StoreChangelogReader` StoreChangelogReader  pauses and resumes changelog partitions on the restore consumer depending on whether the corresponding tasks still exist.

...

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. Operators should be aware that registering metrics per partition increases cardinality linearly with the number of assigned partitions. In environments with large partition counts, this may increase memory usage in the metrics registry and in downstream monitoring systems. The paused-partitions-count aggregate metric is available at INFO level for general monitoring, while the per-partition metrics are intended for detailed debugging when partition-level visibility is needed.

...

Currently, each partition's internal state tracks whether it is paused with a simple boolean flag. To support the `pausedpaused-partitions-duration-seconds` seconds  metric, this KIP adds a timestamp that records when the partition was paused. The timestamp is set when pause()  is called, cleared (reset to -1) when resume() is called, and also reset to -1 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 `pausedpaused-partitions-count` count  gauge is registered in the consumer's top-level metrics. When queried, this gauge computes the current count of paused partitions 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 metric.

consumer-fetch-manager-metrics

The `pausedpaused-partitions` partitions  and `pausedpaused-partitions-duration-seconds` seconds  gauges are registered per partition in the fetch manager metrics, alongside the existing per-partition lag and lead metrics. They follow the same lifecycle:

...

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

...