anchor.diff:      []       -> {error, no_available_offsets}
     end.
   
  -maybe_send_fetch_request(State) ->
  +%% @private Send new fetch request if no pending error.
  +maybe_send_fetch_request(#state{last_error = undefined} = State) ->
     case State#state.cb_pending_cnt < State#state.prefetch_count of
       true ->
         {ok, CorrId} = send_fetch_request(State),
         {ok, State#state{corr_id = CorrId}};
       false ->
         {ok, State}
  -  end.
  +  end;
  +maybe_send_fetch_request(#state{} = State) ->
  +  {ok, State}.
   
   send_fetch_request(#state{socket_pid = SocketPid} = State) ->
     Request = #fetch_request{ topic = State#state.topic
  
positive.code: 
  %%%_* Internal Functions =======================================================
  
  do_handle_messages_result({ok, CbState}, State0) ->
    State1 = State0#state{cb_state = CbState},
    State2 = maybe_spawn_handle_messages(State1),
    {ok, State} = maybe_send_fetch_request(State2),
    {noreply, State};
  do_handle_messages_result({ok, CbState, CbOptions}, State0) ->
    {ok, State1} = update_cb_options(CbOptions, State0),
    OldOffset = State0#state.offset,
  
negative.code:   Parent = self(),
    spawn_link(
      fun() ->
        Result = F(CbState),
        Parent ! {handle_messages_result, Ref, Result}
      end),
    State#state{cb_pending = [Ref | Rest]};
  