anchor.diff:  
   %%%_* Internal functions -------------------------------------------------------
   
  --record(produce_error, { partition  :: partition()
  -                       , offset     :: offset()
  -                       , error_code :: error_code()
  -                       }).
  --type produce_error() :: #produce_error{}.
  -
   -spec handle_produce_response(pid(), corr_id(), [produce_error()], #state{}) ->
           {ok, #state{}} | {error, any()}.
   handle_produce_response(Leader, CorrId, [], State0) ->
  
positive.code:                , senders_buffer = dict:new() :: senders_buffer()
                 , pending = []     :: [{leader(), {corr_id(), [sender()]}}]
                 }).
  
  %%%_* API ----------------------------------------------------------------------
  -spec start_link([brod:host()], integer(), integer()) ->
  
negative.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:revert(
      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  -> [{Partition, Offset, 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
  