anchor.diff:                 , senders_buffer = dict:new() :: senders_buffer()
                  , pending = []     :: [{leader(), {corr_id(), [sender()]}}]
                  }).
  +
  +-type produce_error() :: {partition(), offset(), error_code()}.
   
   %%%_* API ----------------------------------------------------------------------
   -spec start_link([brod:host()], integer(), integer()) ->
  
positive.code: %% @private Flatten the per-topic-partition error codes in produce result.
  -spec get_error_codes([#produce_topic{}]) -> [produce_error()].
  get_error_codes(Topics) ->
    lists:foldl(
      fun(#produce_topic{offsets = Offsets}, TopicAcc) ->
        lists:foldl(
          fun(#produce_offset{partition  = Partition,
                              error_code = ErrorCode,
                              offset     = Offset }, OffsetAcc) ->
            case brod_kafka:is_error(ErrorCode) of
              true  -> [#produce_error{ partition  = Partition
                                      , offset     = Offset
                                      , error_code = ErrorCode
                                      } | OffsetAcc];
              false -> OffsetAcc
            end
          end, TopicAcc, Offsets)
      end, [], Topics).
  
  handle_produce(Sender, Ref, Topic, Partition, KVList, State0) ->
    case get_leader(Topic, Partition, State0) of
  
negative.code: 
  %%%_* Internal functions -------------------------------------------------------
  
  -spec handle_produce_response(pid(), corr_id(), [produce_error()], #state{}) ->
          {ok, #state{}} | {error, any()}.
  handle_produce_response(Leader, CorrId, [], State0) ->
  