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:                       , 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
  