anchor.diff:          is_retriable(ErrorCode) orelse exit({not_retriable, Error}),
           case brod_producer_buffer:nack(Buffer, CorrId, Error) of
             {ok, NewBuffer}  ->
  -            do_schedule_retry(State#state{buffer = NewBuffer});
  +            schedule_retry(State#state{buffer = NewBuffer});
             {error, CorrIdExpected} ->
               _ = log_discarded_corr_id(CorrId, CorrIdExpected),
               maybe_produce(State)
  
positive.code:                  "Correlation ID discarded:~p, expecting: ~p",
                   [CorrIdReceived, CorrIdExpected]).
  
  handle_produce(CallRef, Key, Value, #state{buffer = Buffer} = State) ->
    {ok, NewBuffer} = brod_producer_buffer:add(Buffer, CallRef, Key, Value),
    State1 = State#state{buffer = NewBuffer},
    {ok, NewState} = maybe_produce(State1),
    {noreply, NewState}.
  
  init_socket(#state{ client_pid = ClientPid
                    , sock_pid   = OldSockPid
                    , sock_mref  = OldSockMref
                    , topic      = Topic
                    , partition  = Partition
                    , buffer     = Buffer0
                    } = State) ->
    %% Lookup, or maybe (re-)establish a connection to partition leader
    case brod_client:get_leader_connection(ClientPid, Topic, Partition) of
      {ok, OldSockPid} ->
  
negative.code:                       , sock_mref = SockMref
                        , buffer    = Buffer
                        }};
      {error, Reason} ->
        ok = maybe_demonitor(OldSockMref),
        %% Make sure the sent but not acked ones are put back to buffer
        {ok, Buffer} = brod_producer_buffer:nack_all(Buffer0, no_leader),
        brod_utils:log(warning, "Failed to (re)init socket, reason:\n~p",
                       [Reason]),
        {ok, State#state{ sock_pid  = ?undef
                        , sock_mref = ?undef
                        , buffer    = Buffer
                        }}
    end.
  
  %% @private
  