anchor.diff:      {error, consumer_down}
     end.
   
  +%% @private Init payload socket regardless of subscriber state.
  +-spec maybe_init_socket(#state{}) ->
  +        {ok, #state{}} | {{error, any()}, #state{}}.
  +maybe_init_socket(#state{ client_pid = ClientPid
  +                        , topic      = Topic
  +                        , partition  = Partition
  +                        , socket_pid = ?undef
  +                        } = State0) ->
  +  %% Lookup, or maybe (re-)establish a connection to partition leader
  +  case brod_client:get_leader_connection(ClientPid, Topic, Partition) of
  +    {ok, SocketPid} ->
  +      _ = erlang:monitor(process, SocketPid),
  +      State = State0#state{socket_pid = SocketPid},
  +      {ok, State};
  +    {error, Reason} ->
  +      {{error, Reason}, State0}
  +  end;
  +maybe_init_socket(State) ->
  +  {ok, State}.
  +
  +
  +%% @private Send a ?INIT_SOCKET delayed loopback message to re-init socket.
  +-spec maybe_send_init_socket(#state{}) -> ok.
  +maybe_send_init_socket(#state{subscriber = Subscriber}) ->
  +  Timeout = ?SOCKET_RETRY_DELAY_MS,
  +  %% re-init payload socket only when subscriber is alive
  +  is_pid(Subscriber) andalso is_process_alive(Subscriber) andalso
  +    erlang:send_after(Timeout, self(), ?INIT_SOCKET),
  +  ok.
  +
   %%%_* Tests ====================================================================
   
   -ifdef(TEST).
  
positive.code:   maybe_send_fetch_request(State).
  
  %% @private Send new fetch request if no pending error.
  maybe_send_fetch_request(#state{ subscriber     = Subscriber
                                 , pending_acks   = PendingAcks
                                 , prefetch_count = PrefetchCount
                                 , socket_pid     = SocketPid
                                 , is_suspended   = IsSuspended
                                 } = State) ->
    case is_pid(SocketPid)  andalso
         is_pid(Subscriber) andalso
         not IsSuspended    andalso
         length(PendingAcks) =< PrefetchCount of
      true ->
        case send_fetch_request(State) of
          {ok, CorrId} ->
  
negative.code:   {noreply, NewState};
  handle_info({'DOWN', _MonitorRef, process, Pid, _Reason},
              #state{socket_pid = Pid} = State) ->
    ok = maybe_send_init_socket(State),
    State1 = State#state{socket_pid = ?undef},
    NewState = reset_buffer(State1),
    {noreply, NewState};
  