anchor.diff:    NewMaxBytes = erlang:max(EstimatedSetSize, MaxBytesOrig),
     %% maybe shrink the max_bytes to send in fetch request to NewMaxBytes
     State#state{max_bytes = erlang:min(NewMaxBytes, MaxBytes)};
  -maybe_shrink_max_bytes(#state{ prefetch_count = PrefetchCount
  -                             , avg_bytes      = AvgBytes
  +maybe_shrink_max_bytes(#state{ prefetch_count   = PrefetchCount
  +                             , avg_bytes        = AvgBytes
  +                             , size_stat_window = Window
                                } = State,
                          [#kafka_message{key = Key, value = Value} | Rest]) ->
     %% kafka adds 34 bytes of overhead (metadata) for each message
     %% use 40 to give some room for future kafka protocol versions
     MsgBytes = bytes(Key) + bytes(Value) + 40,
     %% See https://en.wikipedia.org/wiki/Moving_average
  -  WindowSize = erlang:max(PrefetchCount, ?DEFAULT_AVG_WINDOW),
  +  WindowSize = erlang:max(PrefetchCount, Window),
     NewAvgBytes = ((WindowSize - 1) * AvgBytes + MsgBytes) / WindowSize,
     maybe_shrink_max_bytes(State#state{avg_bytes = NewAvgBytes}, Rest).
   
  
positive.code:     , prefetch_count      = F(prefetch_count, State#state.prefetch_count)
      , offset_reset_policy = OffsetResetPolicy
      , max_bytes           = F(max_bytes, State#state.max_bytes)
      },
    NewState =
      case NewBeginOffset =/= OldBeginOffset of
  
negative.code:                , offset_reset_policy :: offset_reset_policy()
                 , avg_bytes           :: number()
                 , max_bytes           :: bytes()
                 , size_stat_window    :: non_neg_integer()
                 }).
  
  -type state() :: #state{}.
  