A brod_transaction is a process that orchestates a set of producers to store messages within a transaction, it also supports committing offsets in the same transaction.

Simple produce sample:

   {ok, Tx} = brod_transaction:new(Client, TxId, []),
   lists:foreach(fun(Partition) ->
                     Key = rand(), Value = rand(),
                     {ok, _Offset} =
                     brod_transaction:produce(Tx,
                                              Topic,
                                              Partition,
                                              Key,
                                              Value),
                 end, Partitions),
   brod_transaction:commit(Tx),

handle callback of a group subscriber using offset commit within a transaction:

   handle_message(Topic,
                  Partition,
                  #kafka_message{ offset = Offset
                                , key = Key
                                , value = Value},
                  #{ client := Client
                   , group_id := GroupId} =  State) ->
     {ok, Tx} = brod_transaction:new(Client),
     {ok, _ProducedOffset} = brod_transaction:produce(Tx, ?TOPIC_OUTPUT, Partition, Key, Value),
     ok = brod_transaction:txn_add_offsets(Tx, GroupId, #{{Topic, Partition} => Offset}),
     ok = brod_transaction:commit(Tx)
 
     {ok, ack_no_commit, State}.