anchor.diff:                   "Correlation ID discarded:~p, expecting: ~p",
                    [CorrIdReceived, CorrIdExpected]).
   
  -handle_produce(CallRef, Key, Value, #state{buffer = Buffer} = State) ->
  +%% @private
  +handle_produce(CallRef, Key, Value,
  +               #state{retry_tref = Ref} = State) when is_reference(Ref) ->
  +  %% pending on retry, add to buffer regardless of socket state
  +  do_handle_produce(CallRef, Key, Value, State);
  +handle_produce(CallRef, Key, Value,
  +               #state{sock_pid = Pid} = State) when is_pid(Pid) ->
  +  %% Socket is alive, add to buffer, and try send produce request
  +  do_handle_produce(CallRef, Key, Value, State);
  +handle_produce(CallRef, Key, Value, #state{} = State) ->
  +  %% this is the first request after fresh start/restart or socket death
  +  {ok, NewState} = maybe_reinit_socket(State),
  +  do_handle_produce(CallRef, Key, Value, NewState).
  +
  +%% @private
  +do_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) ->
  +%% @private
  +-spec maybe_reinit_socket(#state{}) -> #state{}.
  +maybe_reinit_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} ->
  
positive.code: %% @private
  -spec sock_send(?undef | pid(), kpro_ProduceRequest()) ->
          ok | {ok, corr_id()} | {error, any()}.
  sock_send(?undef, _KafkaReq) -> {error, sock_down};
  sock_send(SockPid, KafkaReq) -> brod_sock:request_async(SockPid, KafkaReq).
  
  %%%_* Emacs ====================================================================
  
negative.code:   {noreply, State};
  handle_info(?RETRY_MSG, #state{} = State0) ->
    State1 = State0#state{retry_tref = ?undef},
    {ok, State2} = maybe_reinit_socket(State1),
    %% For retry-interval deterministic, produce regardless of socket state.
    %% In case it has failed to find a new socket in maybe_reinit_socket/1
    %% the produce call should fail immediately with {error, no_leader}
    %% and a new retry should be scheduled (if not reached max_retries yet)
    {ok, State} = maybe_produce(State2),
    {noreply, State};
  handle_info({'DOWN', _MonitorRef, process, Pid, Reason},
              #state{sock_pid = Pid, buffer = Buffer0} = State) ->
    case brod_producer_buffer:is_empty(Buffer0) of
      true ->
        %% no socket restart in case of empty request buffer
        {noreply, State#state{sock_pid = ?undef}};
      false ->
        %% put sent requests back to buffer immediately after socket down
        %% to fail fast if retry is not allowed (reaching max_retries).
        {ok, Buffer} = brod_producer_buffer:nack_all(Buffer0, Reason),
        {ok, NewState} = schedule_retry(State#state{buffer = Buffer}),
        {noreply, NewState#state{sock_pid = ?undef}}
    end;
  handle_info({produce, CallRef, Key, Value}, #state{} = State) ->
    handle_produce(CallRef, Key, Value, State);
  handle_info({msg, Pid, CorrId, #kpro_ProduceResponse{} = R},
              #state{ sock_pid = Pid
                    , buffer   = Buffer
  