Add a lightweight metrics collector protocol for the producer #1166

Motivation

When end-to-end produce latency is high, an application-level timer
wrapped around send_and_wait() cannot tell whether the cause is network
RTT, replication wait on the broker side, producer-side accumulator
back-pressure, retries, or linger_ms batching. Each of these has a
different remediation path (cluster topology, broker tuning, producer
config, application code) and a different operational owner. Without
visibility into the producer's internal stages, every investigation
starts from scratch with guesswork or ad-hoc instrumentation.

The Java client exposes ~25 producer metrics that segment this latency
directly — record-queue-time-avg/max, request-latency-avg/max,
bufferpool-wait-time-total, record-retry-rate, and so on — and they
are the first thing operators reach for. aiokafka exposes none, which
forces every team running it in production to write the same custom
timing wrappers and still end up unable to distinguish the same causes.

This proposal aims to close that gap with a minimal, pluggable API — not
by re-introducing the heavyweight Sensor/Stat/Histogram framework
that was removed in #1117.

Prior art in this repo

• Add metrics #38 (open since 2016) — "Add metrics". Original intent was to reuse the
  python-kafka Sensor framework, but it was never completed.
• Recommended way to expose metrics to prometheus?  #562 (open since 2019) — "Recommended way to expose metrics to Prometheus".
• Add compability with OpenTelemetry #862 (open since 2022) — "Add compatibility with OpenTelemetry". Blocked
  by the lack of hook points.
• Remove dead code vendored from python-kafka (metrics) #1117 (merged 2025-07) — removed 1969 lines of vendored kafka-python
  metrics code. This proposal deliberately does not bring any of it back.

Proposal: callback-based ProducerMetricsCollector protocol

A single new module aiokafka/metrics.py defines a Protocol with sync
callback methods. The producer invokes these at known transition points in
the message lifecycle. No aggregation, no sampling, no sliding windows
implemented inside aiokafka — the user plugs in their preferred backend
(prometheus_client, OpenTelemetry, statsd, in-memory test double, etc.).

from typing import Protocol, runtime_checkable

@runtime_checkable
class ProducerMetricsCollector(Protocol):
    """Receive lifecycle events from AIOKafkaProducer.

    All methods are synchronous and called from the hot path; implementations
    should be fast and non-blocking. If asynchronous work is required, defer
    it via an internal queue.

    All time-valued arguments are in **seconds** (consistent with
    ``time.monotonic()`` and the Prometheus naming convention).
    """

    def on_batch_drained(
        self,
        topic: str,
        queue_time_seconds: float,
        batch_size_bytes: int,
        record_count: int,
    ) -> None:
        """Called when a batch leaves the accumulator and is handed to the
        sender. ``queue_time_seconds`` measures time from batch creation
        to drain — equivalent to Java client's ``record-queue-time``."""

    def on_batch_done(
        self,
        topic: str,
        request_latency_seconds: float,
        record_count: int,
    ) -> None:
        """Called when the broker acknowledges the batch. ``request_latency``
        is measured from drain (handoff to sender) to ack — equivalent to
        Java client's ``request-latency``."""

    def on_batch_failure(
        self,
        topic: str,
        exception: BaseException,
        record_count: int,
    ) -> None:
        """Called when a batch ultimately fails (after retries)."""

    def on_buffer_wait(
        self,
        topic: str,
        wait_seconds: float,
    ) -> None:
        """Called when ``add_message`` had to wait for accumulator space —
        equivalent to Java client's ``bufferpool-wait-time``. Direct
        back-pressure signal."""


class NullMetricsCollector:
    """Default no-op implementation. Zero overhead when nothing is plugged in."""
    def on_batch_drained(self, *args, **kwargs): pass
    def on_batch_done(self, *args, **kwargs): pass
    def on_batch_failure(self, *args, **kwargs): pass
    def on_buffer_wait(self, *args, **kwargs): pass

Wired into the producer via constructor:

producer = AIOKafkaProducer(
    bootstrap_servers=...,
    metrics_collector=MyCollector(),   # default = NullMetricsCollector()
)

Hook locations

One new float field on MessageBatch (_drained_at). All other timings
come from data already tracked by the producer.

Mapping to Java client metrics

Users typically aggregate the raw events into Prometheus Histograms or
similar. The mapping to Java metric names is:

Why callbacks rather than an embedded Sensor framework

1. Doesn't reintroduce what Remove dead code vendored from python-kafka (metrics) #1117 removed. That PR deleted ~2000 lines
   of dead aggregation code. This proposal adds ~150 lines of callback
   plumbing and no aggregation code.
2. No new dependencies. Users who want Prometheus plug in
   prometheus_client.Histogram; users who want OTel plug in OTel.
   aiokafka stays dependency-clean.
3. Better fit for Python ecosystem. prometheus_client, OpenTelemetry
   and statsd all already provide aggregation/percentiles natively — far
   better than the Java Avg/Max sliding-window stats, which lose
   information.
4. Unblocks Add compability with OpenTelemetry #862 (OpenTelemetry). With these hooks, OTel
   auto-instrumentation can be written as a third-party package without any
   further changes to aiokafka.
5. Unblocks Recommended way to expose metrics to prometheus?  #562 (Prometheus exposure). Documented in the new
   docs/metrics.rst with a reference examples/prometheus_metrics.py.
6. Minimal hot-path cost. With NullMetricsCollector, each hook is a
   single attribute lookup + method call returning None. ~50 ns in CPython.
   Hooks fire per-batch, not per-record.

Out of scope for this issue

To keep the PR reviewable, the first PR will include only the
producer-side metrics listed above. Follow-ups can add:

• compression-rate / retry-rate / throttle-time (small additions)
• per-topic dimension via additional labels (topic is already passed)
• consumer-side equivalents (fetch-latency, lag, commit time)
• OTel-specific helper package (likely belongs in a separate repo)

Scope of the first PR

• New module aiokafka/metrics.py (~80 lines: Protocol + NullCollector + docstrings)
• Constructor parameter metrics_collector on AIOKafkaProducer
• Hook calls in MessageBatch.drain_ready, MessageBatch.done,
  MessageBatch.failure, MessageAccumulator.add_message
• New _drained_at field on MessageBatch
• tests/test_metrics.py with a fake collector verifying:
  • events fire on normal send
  • on_buffer_wait fires when accumulator is full
  • on_batch_failure fires when delivery fails
  • NullMetricsCollector does not introduce overhead (smoke test)
• docs/metrics.rst with API reference + Prometheus example
• examples/prometheus_metrics.py
• CHANGES.rst entry under 0.15.0

Total expected diff: ~300 lines added, ~10 lines modified in existing files.

Questions for maintainers before I start coding

1. Are you open to this direction in principle?
2. Naming: I went with Java-aligned terminology (queue_time, request_latency).
   Any preference for renaming?
3. Should topic be passed always, or only when an opt-in
   enable_per_topic_metrics=True flag is set? Java client emits per-topic
   metrics by default; passing topic per-callback adds zero cost but
   downstream aggregation cardinality is the user's choice.
4. Anything about how Remove dead code vendored from python-kafka (metrics) #1117 was decided that I should be aware of, beyond
   "the old framework was dead code"?