Telemetry events emitted by KafkaEx.

KafkaEx uses the :telemetry library to emit events that can be used for observability. This module documents all events and provides helper functions.

Attaching Handlers

To receive telemetry events, attach a handler:

:telemetry.attach("my-handler", [:kafka_ex, :request, :stop], &MyModule.handle_event/4, nil)

Or attach to multiple events:

:telemetry.attach_many("my-handler", KafkaEx.Telemetry.events(), &MyModule.handle_event/4, nil)

Events

Request Events

Low-level Kafka protocol request/response events.

• [:kafka_ex, :request, :start] - Emitted when a Kafka protocol request begins

  • Measurements: %{system_time: integer()}
  • Metadata: %{operation: atom(), api_version: integer(), correlation_id: integer(), client_id: binary(), broker: map()}
  • Note: broker is empty %{} at start (actual broker determined after send)

• [:kafka_ex, :request, :stop] - Emitted when a Kafka protocol request completes

  • Measurements: %{duration: integer()}
  • Metadata: Same as start event plus %{bytes_sent: integer(), bytes_received: integer(), broker: map()}
    • bytes_sent - Number of bytes sent in the request
    • bytes_received - Number of bytes received in the response (0 on error)
    • broker - Map with node_id, host, port of the broker that handled the request

• [:kafka_ex, :request, :exception] - Emitted when a Kafka protocol request fails

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

Connection Events

• [:kafka_ex, :connection, :start] - Emitted when connecting to a broker

  • Measurements: %{system_time: integer()}
  • Metadata: %{host: binary(), port: integer(), ssl: boolean()}

• [:kafka_ex, :connection, :stop] - Emitted when connection is established

  • Measurements: %{duration: integer()}
  • Metadata: Same as start event plus %{success: boolean()}

• [:kafka_ex, :connection, :exception] - Emitted when connection fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

• [:kafka_ex, :connection, :close] - Emitted when a connection is closed

  • Measurements: %{count: 1}
  • Metadata: %{host: binary(), port: integer(), reason: atom()}
    • reason - Why the connection closed:
      • :remote_closed - Remote broker closed the connection
      • :timeout - Request timed out
      • :send_error - Error sending data
      • :recv_error - Error receiving data
      • :shutdown - Client shutdown
      • :init_error - Error during client initialization
      • :metadata_update - Broker removed due to cluster topology change
      • :reconnecting - Socket closed before reconnection attempt

Authentication Events

• [:kafka_ex, :auth, :start] - Emitted when SASL authentication begins

  • Measurements: %{system_time: integer()}
  • Metadata: %{host: binary(), port: integer(), mechanism: binary()}
    • mechanism - The SASL mechanism name ("PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512", "OAUTHBEARER")

• [:kafka_ex, :auth, :stop] - Emitted when SASL authentication completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, error: term()}

    • result - :ok on success, :error on failure
    • error - Error reason (only on failure)

• [:kafka_ex, :auth, :exception] - Emitted when SASL authentication fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

Produce Events

• [:kafka_ex, :produce, :start] - Emitted when a produce operation begins

  • Measurements: %{system_time: integer(), message_count: integer()}
  • Metadata: %{topic: binary(), partition: integer(), client_id: binary(), required_acks: integer()}

• [:kafka_ex, :produce, :stop] - Emitted when a produce operation completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, offset: integer(), error: term()}

    • result - :ok on success, :error on failure
    • offset - Base offset (only on success when acks > 0)
    • error - Error reason (only on failure)

• [:kafka_ex, :produce, :exception] - Emitted when a produce operation fails

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

Fetch Events

• [:kafka_ex, :fetch, :start] - Emitted when a fetch operation begins

  • Measurements: %{system_time: integer()}
  • Metadata: %{topic: binary(), partition: integer(), offset: integer(), client_id: binary()}

• [:kafka_ex, :fetch, :stop] - Emitted when a fetch operation completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, message_count: integer(), error: term()}

    • result - :ok on success, :error on failure
    • message_count - Number of messages fetched (only on success)
    • error - Error reason (only on failure)

• [:kafka_ex, :fetch, :exception] - Emitted when a fetch operation fails

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

Consumer Events

• [:kafka_ex, :consumer, :commit, :start] - Emitted when an offset commit begins

  • Measurements: %{system_time: integer()}
  • Metadata: %{group_id: binary(), client_id: binary(), topic: binary(), partition_count: integer()}

• [:kafka_ex, :consumer, :commit, :stop] - Emitted when an offset commit completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, error: term()}

    • result - :ok on success, :error on failure
    • error - Error reason (only on failure)

• [:kafka_ex, :consumer, :commit, :exception] - Emitted when an offset commit fails

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

Consumer Group Events

• [:kafka_ex, :consumer, :join, :start] - Emitted when joining a consumer group

  • Measurements: %{system_time: integer()}
  • Metadata: %{group_id: binary(), member_id: binary(), topics: [binary()]}

• [:kafka_ex, :consumer, :join, :stop] - Emitted when join group completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, generation_id: integer(), is_leader: boolean(), error: term()}

• [:kafka_ex, :consumer, :join, :exception] - Emitted when join group fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

• [:kafka_ex, :consumer, :sync, :start] - Emitted when syncing consumer group state

  • Measurements: %{system_time: integer()}
  • Metadata: %{group_id: binary(), member_id: binary(), generation_id: integer(), is_leader: boolean()}

• [:kafka_ex, :consumer, :sync, :stop] - Emitted when sync group completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, assigned_partitions: integer(), error: term()}

• [:kafka_ex, :consumer, :sync, :exception] - Emitted when sync group fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

• [:kafka_ex, :consumer, :heartbeat, :start] - Emitted when sending a heartbeat

  • Measurements: %{system_time: integer()}
  • Metadata: %{group_id: binary(), member_id: binary(), generation_id: integer()}

• [:kafka_ex, :consumer, :heartbeat, :stop] - Emitted when heartbeat completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, error: term()}

• [:kafka_ex, :consumer, :heartbeat, :exception] - Emitted when heartbeat fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

• [:kafka_ex, :consumer, :leave, :start] - Emitted when leaving a consumer group

  • Measurements: %{system_time: integer()}
  • Metadata: %{group_id: binary(), member_id: binary()}

• [:kafka_ex, :consumer, :leave, :stop] - Emitted when leave group completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, error: term()}

• [:kafka_ex, :consumer, :leave, :exception] - Emitted when leave group fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

• [:kafka_ex, :consumer, :rebalance] - Emitted when a consumer group rebalance is triggered

  • Measurements: %{count: 1}
  • Metadata: %{group_id: binary(), member_id: binary(), generation_id: integer(), reason: atom()}

• [:kafka_ex, :consumer, :offset_reset] - Emitted when a consumer resets to a start offset because it has no valid committed offset — a new consumer group (:no_committed_offset) or an out-of-range committed offset (:offset_out_of_range). A reset to :latest skips backlog and a reset to :earliest replays it, so attach to this event to alert on unexpected resets.

  • Measurements: %{offset: integer()}

  • Metadata: %{group_id: binary(), topic: binary(), partition: integer(), reset: :earliest | :latest, reason: :no_committed_offset | :offset_out_of_range}

• [:kafka_ex, :consumer, :member_terminated] - Emitted when a consumer-group member dies permanently — it stops (or crashes) and is NOT restarted (under the one_for_all/max_restarts: 0 supervisor). Distinct from a graceful [:kafka_ex, :consumer, :leave, :stop]; attach to alert on members that won't come back. reason distinguishes the cause: :fenced_instance_id / :group_authorization_failed (terminal), {:crashed, module} (an uncaught exception or give-up raise), {:error, atom} (an unrecoverable heartbeat error), {:client_died, term}, or {:crash_loop, term} (the heartbeat crashed repeatedly within the configured window and the member gave up — see crash_rejoin_max_restarts / crash_rejoin_window_ms). terminal_class is a flat, low-cardinality atom (:fenced | :auth | :crash_loop | :crashed | :error | :client_died | :other) derived from reason — tag alerts on it rather than on reason, whose tuple shape carries an unbounded inner term.

  • Measurements: %{count: 1}

  • Metadata: %{group_id: binary(), member_id: binary(), reason: atom() | {:crashed, module()} | {:error, atom()} | {:client_died, term()} | {:crash_loop, term()}, terminal_class: atom()}

• [:kafka_ex, :consumer, :heartbeat_crash] - Emitted each time a member's heartbeat process dies abnormally and the member rejoins in place. NOT emitted on the crash that exceeds the budget — that one trips a terminal stop and surfaces as member_terminated with {:crash_loop, _} instead. Attach to a rate metric (on count) to spot a flapping-but-recovering member before it trips. Do not tag by reason — it carries an unbounded EXIT term; use it for logs/debugging. crashes_in_window is the current sliding-window count (decays as old crashes age out of the window; drops to zero only when the member restarts fresh), not a monotonic gauge.

  • Measurements: %{count: 1, crashes_in_window: non_neg_integer()}
  • Metadata: %{group_id: binary(), member_id: binary(), reason: term()}

Metadata Events

• [:kafka_ex, :metadata, :update, :start] - Emitted when a metadata request begins

  • Measurements: %{system_time: integer()}
  • Metadata: %{client_id: binary(), topics: [binary()]}

• [:kafka_ex, :metadata, :update, :stop] - Emitted when a metadata request completes

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{result: :ok | :error, broker_count: integer(), topic_count: integer(), error: term()}

    • result - :ok on success, :error on failure
    • broker_count - Number of brokers in the cluster (only on success)
    • topic_count - Number of topics in the response (only on success)
    • error - Error reason (only on failure)

• [:kafka_ex, :metadata, :update, :exception] - Emitted when a metadata request fails with exception

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}

Consumer Processing Events

• [:kafka_ex, :consumer, :process, :start] - Emitted when GenConsumer begins processing a message batch

  • Measurements: %{system_time: integer(), message_count: integer()}
  • Metadata: %{group_id: binary(), topic: binary(), partition: integer(), consumer_module: binary()}

• [:kafka_ex, :consumer, :process, :stop] - Emitted when GenConsumer finishes processing a message batch

  • Measurements: %{duration: integer()}

  • Metadata: Same as start event plus %{commit_mode: :async_commit | :sync_commit}

    • commit_mode - The commit mode returned by the consumer callback

• [:kafka_ex, :consumer, :process, :exception] - Emitted when GenConsumer message processing fails

  • Measurements: %{duration: integer()}
  • Metadata: Start metadata plus %{kind: atom(), reason: term(), stacktrace: list()}