anchor.diff:      {ok, Buffer} ->
         {ok, State#state{buffer = Buffer}};
       {error, Reason} ->
  -      schedule_retry(State, {error, Reason})
  +      schedule_retry(State, Reason)
     end;
   maybe_retry(_State) ->
  +  false.
  +
  +is_retriable(ErrorCode) when ErrorCode =:= ?EC_CORRUPT_MESSAGE;
  +                             ErrorCode =:= ?EC_UNKNOWN_TOPIC_OR_PARTITION;
  +                             ErrorCode =:= ?EC_LEADER_NOT_AVAILABLE;
  +                             ErrorCode =:= ?EC_NOT_LEADER_FOR_PARTITION;
  +                             ErrorCode =:= ?EC_REQUEST_TIMED_OUT;
  +                             ErrorCode =:= ?EC_NOT_ENOUGH_REPLICAS;
  +                             ErrorCode =:= ?EC_NOT_ENOUGH_REPLICAS_AFTER_APPEND
  +                             ->
  +  true;
  +is_retriable(_) ->
     false.
   
   %%%_* Emacs ====================================================================
  
positive.code:                                     MaxBatchSize, SendFun),
  
    MaxRetries = ?config(max_retries, ?DEFAULT_MAX_RETRIES),
    RetryTimeout = ?config(retry_timeout, ?DEFAULT_RETRY_TIMEOUT),
    ReconnectTimeout = ?config(reconnect_timeout, ?DEFAULT_RECONNECT_TIMEOUT),
  
    self() ! ?INIT_SOCKET_MSG,
  
    {ok, #state{ client_pid        = ClientPid
               , topic             = Topic
               , partition         = Partition
               , buffer            = Buffer
               , max_retries       = MaxRetries
               , retry_timeout     = RetryTimeout
               , reconnect_timeout = ReconnectTimeout
               }}.
  
  handle_info(?INIT_SOCKET_MSG, #state{ client_pid = ClientPid
                                      , topic      = Topic
                                      , partition  = Partition
                                      } = State) ->
    %% 1. Lookup, or maybe (re-)establish a connection to partition leader
    case brod_client:get_leader_connection(ClientPid, Topic, Partition) of
      {ok, SockPid} ->
        _ = erlang:monitor(process, SockPid),
        %% 2. Register self() to client.
        ok = brod_client:register_producer(ClientPid, Topic, Partition),
  
        %% 3. Update state.
        {noreply, State#state{sock_pid = SockPid}};
      {error, _} ->
        ReconnectTimeout = State#state.reconnect_timeout,
        erlang:send_after(ReconnectTimeout, self(), ?INIT_SOCKET_MSG),
        {noreply, State}
    end;
  handle_info({'DOWN', _MonitorRef, process, Pid, _Reason},
              #state{sock_pid = Pid} = State) ->
    ReconnectTimeout = State#state.reconnect_timeout,
    erlang:send_after(ReconnectTimeout, self(), ?INIT_SOCKET_MSG),
    {noreply, State#state{sock_pid = undefined}};
  handle_info({msg, Pid, CorrId, #produce_response{} = R},
              #state{ sock_pid = Pid
                    , buffer   = Buffer
  
negative.code:   ok.
  
  %%%_* Internal Functions =======================================================
  init_socket(#state{ client_pid = ClientPid
                    , sock_mref  = OldSockMref
                    , topic      = Topic
                    , partition  = Partition
                    } = State) ->
    %% 1. Lookup, or maybe (re-)establish a connection to partition leader
    case brod_client:get_leader_connection(ClientPid, Topic, Partition) of
      {ok, SockPid} ->
        ok = maybe_demonitor(OldSockMref),
        SockMref = erlang:monitor(process, SockPid),
        %% 2. Register self() to client.
        ok = brod_client:register_producer(ClientPid, Topic, Partition),
  
        %% 3. Update state.
        {ok, State#state{sock_pid = SockPid, sock_mref = SockMref}};
      {error, Error} ->
        {error, Error}
    end.
  
  maybe_demonitor(undefined) ->
    ok;
  maybe_demonitor(Mref) ->
    true = erlang:demonitor(Mref, [flush]),
    ok.
  
  schedule_retry(#state{retry_tref = undefined} = State, Msg) ->
    {ok, TRef} = timer:send_after(State#state.retry_backoff_ms, {?RETRY_MSG, Msg}),
    {ok, State#state{retry_tref = TRef}};
  schedule_retry(State, _Msg) ->
    %% retry timer has been already activated
  