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:                                             }, Offsets).
  
  %% @private
  maybe_shrink_max_bytes(#state{ prefetch_count = PrefetchCount
                               , max_bytes_orig = MaxBytesOrig
                               , max_bytes      = MaxBytes
  
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) ->
  