anchor.diff:    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)
  +            #state{last_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]),
  +  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}};
  +  {noreply, State#state{last_corr_id = CorrId}};
   handle_info({'DOWN', _MonitorRef, process, Pid, Reason},
               #state{socket_pid = Pid} = State) ->
     {stop, {socket_down, Reason}, State};
  
positive.code:                , max_bytes      :: integer()
                 , sleep_timeout  :: integer()
                 , prefetch_count :: integer()
                 , corr_id        :: corr_id()
                 , last_error     :: last_error()
                 }).
  
  
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;
  