anchor.diff:    PrefetchCount =
       proplists:get_value(prefetch_count, Config, ?DEFAULT_PREFETCH_COUNT),
     Offset = proplists:get_value(offset, Config, ?DEFAULT_BEGIN_OFFSET),
  -
  -  %% 3. Retrieve valid starting offset
  -  State = State0#state{ socket_pid     = SocketPid
  -                      , begin_offset   = Offset
  -                      , max_wait_time  = MaxWaitTime
  -                      , min_bytes      = MinBytes
  -                      , max_bytes      = MaxBytes
  -                      , sleep_timeout  = SleepTimeout
  -                      , prefetch_count = PrefetchCount
  -                      },
  +  {ok, #state{ client_pid     = ClientPid
  +             , topic          = Topic
  +             , partition      = Partition
  +             , begin_offset   = Offset
  +             , max_wait_time  = MaxWaitTime
  +             , min_bytes      = MinBytes
  +             , max_bytes      = MaxBytes
  +             , sleep_timeout  = SleepTimeout
  +             , prefetch_count = PrefetchCount
  +             , socket_pid     = undefined
  +             }}.
  +
  +handle_info(init_socket, #state{ client_pid = ClientPid
  +                               , topic      = Topic
  +                               , partition  = Partition
  +                               , socket_pid = undefined
  +                               , subscriber = Subscriber
  +                               } = State0) ->
  +  %% 1. Lookup, or maybe (re-)establish a connection to partition leader
  +  {ok, SocketPid} =
  +    brod_client:get_leader_connection(ClientPid, Topic, Partition),
  +  _ = erlang:monitor(process, SocketPid),
  +
  +  %% 2. Retrieve valid starting offset
  +  State = State0#state{socket_pid = SocketPid},
     case resolve_begin_offset(State) of
       {ok, NewState} ->
         ok = brod_client:register_consumer(ClientPid, Topic, Partition),
  +      %% 3. Start fetching messages if subscriber is present
  +      case is_pid(Subscriber) of
  +        true  -> self() ! ?SEND_FETCH_REQUEST;
  +        false -> ok
  +      end,
         {noreply, NewState};
       {error, Reason} ->
         {stop, {error, Reason}, State}
  
positive.code: maybe_send_fetch_request(#state{ subscriber     = Subscriber
                                 , pending_acks   = PendingAcks
                                 , prefetch_count = PrefetchCount
                                 } = State) ->
    case is_pid(Subscriber) andalso length(PendingAcks) < PrefetchCount of
      true ->
        {ok, CorrId} = send_fetch_request(State),
        State#state{last_corr_id = CorrId};
      false ->
        State
    end.
  
negative.code: -define(DEFAULT_SLEEP_TIMEOUT, 1000). % 1 sec
  -define(DEFAULT_PREFETCH_COUNT, 1).
  -define(ERROR_COOLDOWN, 1000).
  -define(SOCKET_RETRY_DELAY_MS, 1000).
  
  -define(SEND_FETCH_REQUEST, send_fetch_request).
  
  