Context
Follow-up to the producer metrics API from #1166 / #1171.
AIOKafkaConsumer has no equivalent of metrics_collector. Teams that need fetch latency, lag, throughput, or commit timing still have to wrap application code and cannot see fetcher / coordinator internals (#38, #562).
Proposal
Add ConsumerMetricsCollector using the same pattern as ProducerMetricsCollector:
- sync callbacks on the hot path
- no aggregation / sampling / backend inside aiokafka
- keyword-only callback arguments
- no-op defaults so subclasses override only what they need
metrics_collector= on AIOKafkaConsumer- experimental until the shape stabilizes
Initial callback set
| Callback |
Intent |
on_fetch_completed |
fetch RTT + records/bytes returned (per topic-partition) |
on_fetch_failed |
fetch error after retries / terminal failure |
on_records_consumed |
records/bytes handed to the application via getone / getmany |
on_records_lag |
highwater − position after a fetch update |
on_commit_completed |
offset-commit latency |
on_commit_failed |
offset-commit failure |
topic and partition always passed where applicable; collectors decide on label cardinality. Durations in seconds.
Out of scope for the first PR
- rebalance / heartbeat lifecycle metrics
- connection / broker-level metrics
- OpenTelemetry helper package
Implementation notes
- Mirror producer conventions (
Null* / default no-ops, typo warning for unknown on_*, exceptions logged and ignored).
- Likely hook points:
Fetcher (fetch + lag), consumer poll path (records consumed), group coordinator (commits).
- Docs + example extension of the existing Prometheus example;
CHANGES.rst entry.
Questions
- Is this initial callback set the right first cut?
- Should lag be reported from the fetcher on every successful fetch, or only when the application consumes?
Context
Follow-up to the producer metrics API from #1166 / #1171.
AIOKafkaConsumerhas no equivalent ofmetrics_collector. Teams that need fetch latency, lag, throughput, or commit timing still have to wrap application code and cannot see fetcher / coordinator internals (#38, #562).Proposal
Add
ConsumerMetricsCollectorusing the same pattern asProducerMetricsCollector:metrics_collector=onAIOKafkaConsumerInitial callback set
on_fetch_completedon_fetch_failedon_records_consumedgetone/getmanyon_records_lagon_commit_completedon_commit_failedtopicandpartitionalways passed where applicable; collectors decide on label cardinality. Durations in seconds.Out of scope for the first PR
Implementation notes
Null*/ default no-ops, typo warning for unknownon_*, exceptions logged and ignored).Fetcher(fetch + lag), consumer poll path (records consumed), group coordinator (commits).CHANGES.rstentry.Questions