anchor.diff:                               | reset_to_latest.
   
   -type bytes() :: non_neg_integer().
  -
  +-type offset_range() :: {offset(),offset()}.
  +-type offsets_queue() :: queue:queue(offset_range()).
  +-record(pending_acks, { count         = 0            :: integer()
  +                      , offsets_queue = queue:new()  :: offsets_queue()
  +                      }).
   -record(state, { client_pid          :: pid()
                  , socket_pid          :: pid()
                  , topic               :: binary()
  
positive.code: maybe_send_fetch_request(#state{last_corr_id = I} = State) when is_integer(I) ->
    %% Waiting for the last request
    State;
  maybe_send_fetch_request(#state{ pending_acks   = PendingAcks
                                 , prefetch_count = PrefetchCount
                                 } = State) ->
    case length(PendingAcks) =< PrefetchCount of
      true ->
        case send_fetch_request(State) of
          {ok, CorrId} ->
  
negative.code: -define(SEND_FETCH_REQUEST, send_fetch_request).
  -define(INIT_SOCKET, init_socket).
  -define(DEFAULT_AVG_WINDOW, 5).
  -define(EMPTY_QUEUE, {[],[]}).
  
  %%%_* APIs =====================================================================
  %% @equiv start_link(ClientPid, Topic, Partition, Config, [])
  