anchor.diff:                 , max_bytes      :: integer()
                  , sleep_timeout  :: integer()
                  , prefetch_count :: integer()
  -               , corr_id        :: corr_id()
  +               , last_corr_id   :: corr_id()
                  , last_error     :: last_error()
                  }).
   
  
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:                           },
        %% 4. Start consuming
        {ok, CorrId} = send_fetch_request(State),
        {noreply, State#state{last_corr_id = CorrId}}
    end;
  handle_info({msg, _Pid, CorrId, R}, #state{last_corr_id = CorrId} = State0) ->
    #fetch_response{topics = [TopicFetchData]} = R,
    #topic_fetch_data{ topic = Topic
                     , partitions = [PM]} = TopicFetchData,
  