anchor.diff:                 , pending_acks = []   :: [offset()]
                  , is_suspended        :: boolean()
                  , offset_reset_policy :: offset_reset_policy()
  +               , avg_bytes           :: number()
  +               , max_bytes           :: bytes()
                  }).
   
   -define(DEFAULT_BEGIN_OFFSET, ?OFFSET_LATEST).
  
positive.code:                              | reset_to_earliest
                               | reset_to_latest.
  
  -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()
                 , sleep_timeout       :: integer()
                 , prefetch_count      :: integer()
                 , last_corr_id        :: corr_id()
  
negative.code:   State = maybe_delay_fetch_request(State0),
    {noreply, State};
  handle_message_set(#kafka_message_set{messages = [?incomplete_message]},
                     #state{ max_bytes = MaxBytes} = State0) ->
    %% max_bytes is too small to fetch ONE complete message, double it.
    NewMaxBytes = MaxBytes * 2,
    NewMaxBytes >= (1 bsl 31) andalso
        erlang:error({max_bytes_overflow,
                     <<"perhaps message corruption in broker?">>}),
    State1 = State0#state{max_bytes = NewMaxBytes},
    State = maybe_send_fetch_request(State1),
    {noreply, State};
  