anchor.diff:                 , offset_reset_policy :: offset_reset_policy()
                  , avg_bytes           :: number()
                  , max_bytes           :: bytes()
  +               , size_stat_window    :: non_neg_integer()
                  }).
   
   -type state() :: #state{}.
  
positive.code: %%                          'begin_offset' option.
  %%     reset_to_earliest: consume from the earliest offset.
  %%     reset_to_latest: consume from the last available offset.
  %% @end
  -spec start_link(pid(), topic(), partition(), config(), [any()]) ->
                      {ok, pid()} | {error, any()}.
  
negative.code:              , offset_reset_policy = OffsetResetPolicy
               , avg_bytes           = 0
               , max_bytes           = MaxBytes
               , size_stat_window    = Cfg(size_stat_window, ?DEFAULT_AVG_WINDOW)
               }}.
  
  handle_info(?INIT_SOCKET, #state{subscriber = Subscriber} = State0) ->
  