Depend on telemetry and emit produce request event

Add a dependency on the telemetry library and a brod_metrics module
confining all telemetry calls, based on the pattern established in
[kafka4beam/wolff](https://github.com/kafka4beam/wolff/blob/4feffa6534a960377507fb2dc840db07548b2533/src/wolff_metrics.erl).

The first event, `[brod, product_request_sent]`, is emitted by
brod_producer for every produce request successfully sent on the wire.
This makes the actual wire-level batching observable, which is
otherwise invisible to callers

More events can be added to brod_metrics incrementally.

Related: kafka4beam#503, kafka4beam#512

Author: zachallaun

CHANGELOG.md

@@ -6,6 +6,10 @@
     because the request no longer includes a per-request retention time. Use
     the broker-side `offsets.retention.minutes` setting instead. Negotiated
     versions v2 through v4 continue to use `offset_retention_seconds`.
+  - Emit [`telemetry`](https://github.com/beam-telemetry/telemetry) events,
+    starting with `[brod, produce_request_sent]`.
+    Events are documented in the new `brod_metrics` module.
+    This adds a dependency on the `telemetry` library.
 
 - 4.6.0
   - Add KIP-345 **static group membership** for coordinators configured with an

README.md

@@ -434,6 +434,11 @@ start(ClientId) ->
                                    _CallbackInitArg = []).
 ```
 
+## Beam Telemetry Hooks
+
+Brod emits [Beam Telemetry](https://github.com/beam-telemetry/telemetry) events that users can attach handler functions to.
+The emitted events are documented in the `brod_metrics` module.
+
 ## Authentication support
 
 brod supports SASL `PLAIN`, `SCRAM-SHA-256` and `SCRAM-SHA-512` authentication mechanisms out of the box.

rebar.config

@@ -1,4 +1,4 @@
-{deps, [{kafka_protocol, "4.3.5"}]}.
+{deps, [{kafka_protocol, "4.3.5"}, {telemetry, "1.4.2"}]}.
 {project_plugins, [{rebar3_lint, "~> 3.2.5"}]}.
 {edoc_opts, [{preprocess, true}]}.
 {erl_opts, [warnings_as_errors, warn_unused_vars,warn_shadow_vars,warn_obsolete_guard,debug_info]}.

src/brod.app.src

@@ -3,7 +3,7 @@
  [{description,"Apache Kafka Erlang client library"},
   {vsn,"git"},
   {registered,[]},
-  {applications,[kernel,stdlib,kafka_protocol]},
+  {applications,[kernel,stdlib,kafka_protocol,telemetry]},
   {env,[]},
   {mod, {brod, []}},
   {modules,[]},

src/brod_metrics.erl

@@ -0,0 +1,60 @@
+%%%
+%%%   Copyright (c) 2026 kafka4beam contributors
+%%%
+%%%   Licensed under the Apache License, Version 2.0 (the "License");
+%%%   you may not use this file except in compliance with the License.
+%%%   You may obtain a copy of the License at
+%%%
+%%%       http://www.apache.org/licenses/LICENSE-2.0
+%%%
+%%%   Unless required by applicable law or agreed to in writing, software
+%%%   distributed under the License is distributed on an "AS IS" BASIS,
+%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%%%   See the License for the specific language governing permissions and
+%%%   limitations under the License.
+%%%
+
+%% @doc Telemetry events emitted by brod.
+%%
+%% Brod emits events using the
+%% <a href="https://github.com/beam-telemetry/telemetry">Beam Telemetry</a>
+%% library. Users can attach handler functions to these events, for
+%% example to report metrics. Each event is emitted by a function in
+%% this module; see the function documentation for the measurements
+%% and metadata attached to each event.
+-module(brod_metrics).
+
+-export([produce_request_sent/3]).
+
+-type batch_input() :: brod:batch_input().
+-type partition() :: brod:partition().
+-type topic() :: brod:topic().
+
+%% @doc Emit a `[brod, produce_request_sent]' event.
+%%
+%% Emitted by producers for each produce request successfully sent on
+%% wire, i.e. one event per Kafka produce request (message batch).
+%%
+%% Measurements:
+%% <ul>
+%% <li>`count': number of messages in the request.</li>
+%% <li>`bytes': total size of keys, values and headers of the messages
+%%     in the request (before encoding and compression).</li>
+%% </ul>
+%%
+%% Metadata: `topic' and `partition' the request was sent to.
+-spec produce_request_sent(topic(), partition(), batch_input()) -> ok.
+produce_request_sent(Topic, Partition, BatchInput) ->
+  telemetry:execute([brod, produce_request_sent],
+                    #{ count => length(BatchInput)
+                     , bytes => brod_utils:bytes(BatchInput)
+                     },
+                    #{ topic => Topic
+                     , partition => Partition
+                     }).
+
+%%%_* Emacs ====================================================================
+%%% Local Variables:
+%%% allout-layout: t
+%%% erlang-indent-level: 2
+%%% End:

src/brod_producer.erl

@@ -442,10 +442,12 @@ do_send_fun(ExtraArg, Conn, BatchInput, Vsn) ->
     brod_kafka_request:produce(Vsn, Topic, Partition, BatchInput,
                                RequiredAcks, AckTimeout, Compression),
   case send(Conn, ProduceRequest) of
-    ok when ProduceRequest#kpro_req.no_ack ->
-      ok;
     ok ->
-      {ok, ProduceRequest#kpro_req.ref};
+      brod_metrics:produce_request_sent(Topic, Partition, BatchInput),
+      case ProduceRequest#kpro_req.no_ack of
+        true -> ok;
+        false -> {ok, ProduceRequest#kpro_req.ref}
+      end;
     {error, Reason} ->
       {error, Reason}
   end.

test/brod_producer_stub_SUITE.erl

@@ -29,12 +29,16 @@
 %% Test cases
 -export([ t_normal_flow/1
         , t_no_required_acks/1
+        , t_produce_request_telemetry/1
         , t_retry_on_same_connection/1
         , t_retry_on_kafka_storage_error/1
         , t_connection_down_retry/1
         , t_leader_migration/1
         ]).
 
+%% Telemetry test handler
+-export([ handle_telemetry_event/4 ]).
+
 
 -include_lib("eunit/include/eunit.hrl").
 -include("brod_int.hrl").
@@ -168,6 +172,39 @@ t_no_required_acks(Config) when is_list(Config) ->
                             result = brod_produce_req_acked}, ok, 2000),
   ok = brod_producer:stop(Producer).
 
+t_produce_request_telemetry(Config) when is_list(Config) ->
+  Tester = self(),
+  HandlerId = <<"t-produce-request-telemetry">>,
+  ok = telemetry:attach(HandlerId, [brod, produce_request_sent],
+                        fun ?MODULE:handle_telemetry_event/4,
+                        #{pid => Tester}),
+  try
+    meck:expect(brod_client, get_leader_connection,
+                fun(_, <<"topic">>, 0) -> {ok, Tester} end),
+    meck:expect(kpro, request_async,
+                fun(Connection, KafkaReq) ->
+                    Connection ! {request_async, KafkaReq},
+                    ok
+                end),
+    ProducerConfig = [{max_linger_ms, 1000},
+                      {max_linger_count, 3}],
+    {ok, Producer} = brod_producer:start_link(client, <<"topic">>, 0,
+                                              ProducerConfig),
+    {ok, _} = brod_producer:produce(Producer, <<"k1">>, <<"v1">>),
+    {ok, _} = brod_producer:produce(Producer, <<"k2">>, <<"v2">>),
+    {ok, _} = brod_producer:produce(Producer, <<"k3">>, <<"v3">>),
+    ?WAIT({request_async, _}, ok, 2000),
+    ?WAIT({telemetry, [brod, produce_request_sent], Measurements, Metadata},
+          begin
+            ?assertMatch(#{count := 3, bytes := Bytes} when Bytes > 0,
+                         Measurements),
+            ?assertEqual(#{topic => <<"topic">>, partition => 0}, Metadata)
+          end, 2000),
+    ok = brod_producer:stop(Producer)
+  after
+    telemetry:detach(HandlerId)
+  end.
+
 t_retry_on_same_connection(Config) when is_list(Config) ->
   Tester = self(),
   %% A mocked connection process which expects 2 requests to be sent
@@ -413,6 +450,9 @@ t_leader_migration(Config) when is_list(Config) ->
 
 %%%_* Help functions ===========================================================
 
+handle_telemetry_event(Event, Measurements, Metadata, #{pid := Pid}) ->
+  Pid ! {telemetry, Event, Measurements, Metadata}.
+
 meck_module(Module) ->
   meck:new(Module, [passthrough, no_passthrough_cover, no_history]).