anchor.diff:                               | reset_to_earliest
                                | reset_to_latest.
   
  +-type bytes() :: non_neg_integer().
  +
   -record(state, { client_pid          :: pid()
                  , socket_pid          :: pid()
                  , topic               :: binary()
                  , partition           :: integer()
                  , begin_offset        :: offset_time()
                  , max_wait_time       :: integer()
  -               , min_bytes           :: integer()
  -               , max_bytes           :: integer()
  +               , min_bytes           :: bytes()
  +               , max_bytes_orig      :: bytes()
                  , sleep_timeout       :: integer()
                  , prefetch_count      :: integer()
                  , last_corr_id        :: corr_id()
  
positive.code:   MaxBytes = Cfg(max_bytes, ?DEFAULT_MAX_BYTES),
    MaxWaitTime = Cfg(max_wait_time, ?DEFAULT_MAX_WAIT_TIME),
    SleepTimeout = Cfg(sleep_timeout, ?DEFAULT_SLEEP_TIMEOUT),
    PrefetchCount = Cfg(prefetch_count, ?DEFAULT_PREFETCH_COUNT),
    BeginOffset = Cfg(begin_offset, ?DEFAULT_BEGIN_OFFSET),
    OffsetResetPolicy = Cfg(offset_reset_policy, ?DEFAULT_OFFSET_RESET_POLICY),
    ok = brod_client:register_consumer(ClientPid, Topic, Partition),
  
negative.code: %%     Maximum bytes to fetch in a batch of messages
  %%     NOTE: this value might be doubled in each retry when it is not enough
  %%           to fetch even one single message.
  %%  max_wait_time (optional, default = 10000 ms):
  %%     Max number of seconds allowd for the broker to collect min_bytes of
  %%     messages in fetch response
  