anchor.diff:    {noreply, NewState};
   handle_info({'DOWN', _MonitorRef, process, Pid, _Reason},
               #state{socket_pid = Pid} = State) ->
  -  Timeout = ?SOCKET_RETRY_DELAY_MS,
  -  erlang:send_after(Timeout, self(), ?INIT_SOCKET),
  +  ok = maybe_send_init_socket(State),
     State1 = State#state{socket_pid = ?undef},
     NewState = reset_buffer(State1),
     {noreply, NewState};
  
positive.code:     {error, consumer_down}
    end.
  
  %%%_* Tests ====================================================================
  
  -ifdef(TEST).
  
negative.code:                        State#state.min_bytes,
                         State#state.max_bytes),
    brod_sock:request_async(SocketPid, Request).
  
  handle_subscribe_call(Pid, Options,
                        #state{subscriber_mref = OldMref} = State0) ->
    case update_options(Options, State0) of
      {ok, State1} ->
        %% demonitor in case the same process tries to subscribe again
        is_reference(OldMref) andalso erlang:demonitor(OldMref, [flush]),
        Mref = erlang:monitor(process, Pid),
        State2 = State1#state{ subscriber      = Pid
                             , subscriber_mref = Mref
                             },
        %% always reset buffer to fetch again
        State3 = reset_buffer(State2),
        State4 = State3#state{is_suspended = false},
        State  = maybe_send_fetch_request(State4),
        {reply, ok, State};
      {error, Reason} ->
        {reply, {error, Reason}, State0}
    end.
  
  -spec update_options(options(), #state{}) -> {ok, #state{}} | {error, any()}.
  update_options(Options, #state{begin_offset = OldBeginOffset} = State) ->
  