anchor.diff:        false ->
           State1#state{ cb_pending = []
                       , cb_pending_cnt = 0
  -                    , corr_id = undefined
  +                    , last_corr_id = undefined
                       }
       end,
     {ok, NewState}.
  
positive.code:   end;
  %% handle obsolete fetch responses in case we got new offset from callback
  handle_info({msg, Pid, CorrId1, _R},
              #state{corr_id = CorrId2, socket_pid = Pid} = State)
    when CorrId1 < CorrId2 ->
    error_logger:info_msg("~p ~p Dropping obsolete fetch response with corr_id = ~p",
                          [?MODULE, self(), CorrId1]),
    {noreply, State};
  handle_info(?SEND_FETCH_REQUEST, State) ->
    {ok, CorrId} = send_fetch_request(State),
    {noreply, State#state{corr_id = CorrId}};
  handle_info({'DOWN', _MonitorRef, process, Pid, Reason},
              #state{socket_pid = Pid} = State) ->
    {stop, {socket_down, Reason}, State};
  
negative.code:   case State#state.cb_pending_cnt < State#state.prefetch_count of
      true ->
        {ok, CorrId} = send_fetch_request(State),
        {ok, State#state{last_corr_id = CorrId}};
      false ->
        {ok, State}
    end;
  