anchor.diff:              #state{socket_pid = Pid} = State) ->
     {stop, {socket_down, Reason}, State};
   %% callback completed
  -handle_info({'DOWN', MRef, process, Pid, Reason}, State0) ->
  -  CbPending0 = State0#state.cb_pending,
  -  CbPendingCnt0 = State0#state.cb_pending_cnt,
  -  IsBlocked = CbPendingCnt0 > State0#state.prefetch_count,
  -  case lists:member(MRef, CbPending0) of
  -    true ->
  -      CbPending = lists:delete(MRef, CbPending0),
  -      CbPendingCnt = CbPendingCnt0 - 1,
  -      case Reason of
  -        {shutdown, {ok, CbState}} ->
  -          State1 = State0#state{ cb_pending = CbPending
  -                               , cb_pending_cnt = CbPendingCnt
  -                               , cb_state = CbState},
  -          {ok, State} = maybe_unblock(IsBlocked, State1),
  -          {noreply, State};
  -        {shutdown, {ok, CbState, NewCbOptions}} ->
  -          {ok, State1} = update_cb_options(NewCbOptions, State0),
  -          OldOffset = State0#state.offset,
  -          NewOffset0 = State1#state.offset,
  -          SocketPid = State1#state.socket_pid,
  -          Topic = State1#state.topic,
  -          Partition = State1#state.partition,
  -          {ok, NewOffset} = maybe_fetch_valid_offset(SocketPid, OldOffset,
  -                                                     NewOffset0, Topic,
  -                                                     Partition),
  -          State1 = State1#state{ cb_pending = CbPending
  -                               , cb_pending_cnt = CbPendingCnt
  -                               , cb_state = CbState
  -                               , offset = NewOffset},
  -          {ok, State} = maybe_unblock(IsBlocked, State1),
  -          {noreply, State};
  -        Other ->
  -          {stop, {callback_error, Other}, State0}
  -      end;
  -    false ->
  -      error_logger:warning_msg("Unexpected 'DOWN' in ~p ~p from ~p, reason: ~p",
  -                               [?MODULE, self(), Pid, Reason]),
  -      {noreply, State0}
  +handle_info({handle_messages_result, Ref, Result}, State) ->
  +  [Ref | Rest] = State#state.cb_pending, %% assert
  +  CbPendingCnt = State#state.cb_pending_cnt - 1,
  +  NewState = State#state{ cb_pending     = Rest
  +                        , cb_pending_cnt = CbPendingCnt
  +                        },
  +  case Result of
  +    {ok, HandleMessagesResult} ->
  +      do_handle_messages_result(HandleMessagesResult, NewState);
  +    {error, Reason} ->
  +      {stop, {callback_crash, Reason}, NewState}
     end;
   handle_info(Info, State) ->
     error_logger:warning_msg("~p [~p] ~p got unexpected info: ~p",
  
positive.code:     []       -> {error, no_available_offsets}
    end.
  
  maybe_unblock(false, State) ->
    {ok, State};
  maybe_unblock(true, State) ->
    maybe_send_fetch_request(State).
  
  maybe_send_fetch_request(State) ->
    case State#state.cb_pending_cnt > State#state.prefetch_count of
      true ->
        {ok, State};
      false ->
        {ok, CorrId} = send_fetch_request(State),
        {ok, State#state{corr_id = CorrId}}
    end.
  
  send_fetch_request(#state{socket_pid = SocketPid} = State) ->
  
negative.code:                                       State#state.prefetch_count),
    Offset = proplists:get_value(offset, NewCbOptions,
                                 State#state.offset),
    State1 = State#state{ offset = Offset
                        , min_bytes = MinBytes
                        , max_bytes = MaxBytes
                        , max_wait_time = MaxWaitTime
                        , sleep_timeout = SleepTimeout
                        , prefetch_count = PrefetchCount
                        },
    %% in case callback wants to fetch from a new offset, rest fetch buffer
    NewState =
      case OldOffset =/= Offset of
        true  ->
          State1;
        false ->
          State1#state{ cb_pending = []
                      , cb_pending_cnt = 0
                      , corr_id = undefined
                      }
      end,
    {ok, NewState}.
  
  %%%_* Tests ====================================================================
  
  