anchor.diff:                                  , data    = LeaderBuffer},
         {ok, CorrId} = brod_sock:send(Leader, Produce),
         DataBuffer = lists:keydelete(Leader, 1, State#state.data_buffer),
  -      FromList = dict:fetch(Leader, State#state.from_buffer),
  -      FromBuffer = dict:erase(Leader, State#state.from_buffer),
  -      Pending = [{Leader, {CorrId, FromList}} | State#state.pending],
  -      {ok, State#state{ data_buffer = DataBuffer
  -                      , from_buffer = FromBuffer
  -                      , pending     = Pending}};
  +      SendersList = dict:fetch(Leader, State#state.senders_buffer),
  +      SendersBuffer = dict:erase(Leader, State#state.senders_buffer),
  +      Pending = [{Leader, {CorrId, SendersList}} | State#state.pending],
  +      {ok, State#state{ data_buffer    = DataBuffer
  +                      , senders_buffer = SendersBuffer
  +                      , pending        = Pending}};
       {_, _} -> % waiting for ack
         {ok, State}
     end.
  
positive.code:   {stop, normal, ok, State};
  handle_call(get_sockets, _From, State) ->
    {reply, {ok, State#state.sockets}, State};
  handle_call(Request, _From, State) ->
    {reply, {error, {unsupported_call, Request}}, State}.
  
  handle_cast(_Msg, State) ->
    {noreply, State}.
  
  handle_info({produce, From, Topic, Partition, KVList}, State0) ->
    case handle_produce(From, Topic, Partition, KVList, State0) of
      {ok, Leader, State1} ->
        {ok, State} = maybe_send(Leader, State1),
        {noreply, State};
      {error, Error} ->
        {Pid, Ref} = From,
        Pid ! {{Ref, self()}, {error, Error}},
        {noreply, State0}
    end;
  handle_info({msg, Leader, CorrId, #produce_response{}}, State0) ->
    {value, {_, {CorrId, FromList}}, Pending} =
      lists:keytake(Leader, 1, State0#state.pending),
    lists:foreach(fun({Pid, Ref}) ->
                      Pid ! {{Ref, self()}, ack}
                  end, FromList),
    {ok, State} = maybe_send(Leader, State0#state{pending = Pending}),
    {noreply, State};
  handle_info({'EXIT', Pid, Reason}, State) ->
  
negative.code: -type partition() :: non_neg_integer().
  -type leader()    :: pid(). % brod_sock pid
  -type corr_id()   :: non_neg_integer().
  -type sender()    :: {pid(), reference()}.
  
  -record(state, { hosts            :: [brod:host()]
                 , debug            :: [term()]
                 , acks             :: integer()
                 , ack_timeout      :: integer()
                 , sockets = []     :: [#socket{}]
                 , leaders = []     :: [{{topic(), partition()}, leader()}]
                 , data_buffer = [] :: [{leader(), [{topic(), dict()}]}]
                 , senders_buffer = dict:new() :: dict() % (leader(), [sender()])
                 , pending = []     :: [{leader(), {corr_id(), [sender()]}}]
                 }).
  
  %%%_* API ----------------------------------------------------------------------
  