anchor.diff:      , prefetch_count      = F(prefetch_count, State#state.prefetch_count)
       , offset_reset_policy = OffsetResetPolicy
       , max_bytes           = F(max_bytes, State#state.max_bytes)
  +    , size_stat_window    = F(size_stat_window, State#state.size_stat_window)
       },
     NewState =
       case NewBeginOffset =/= OldBeginOffset of
  
positive.code:                                             }, Offsets).
  
  %% @private
  maybe_shrink_max_bytes(#state{ prefetch_count = PrefetchCount
                               , max_bytes_orig = MaxBytesOrig
                               , max_bytes      = MaxBytes
  
negative.code: %%                          'begin_offset' option.
  %%     reset_to_earliest: consume from the earliest offset.
  %%     reset_to_latest: consume from the last available offset.
  %%  size_stat_window: (optional, default = 5)
  %%     The moving-average window size when shrinking 'max_bytes' after it has
  %%     been expanded to fetch a large message. Use 0 to immediately shrink
  %%     back to original 'max_bytes' from config.
  %% @end
  -spec start_link(pid(), topic(), partition(), config(), [any()]) ->
                      {ok, pid()} | {error, any()}.
  