Enhance the calculation of avg_record_size in kafka consumer

[WenyXu]

          I am concerned about significant fluctuations in `avg_record_size`. Here are a few suggestions for improvement:

• Exponential Moving Average (EMA):

let alpha = 0.2; // adjustable smoothing factor
*this.avg_record_size = ((1.0 - alpha) * *this.avg_record_size as f64 + 
                        alpha * (fetch_bytes as f64 / records_and_offsets.len() as f64)) as usize;

• Incremental Update with Limits:
  Set a maximum change magnitude to avoid large single-batch fluctuations.

let new_avg = fetch_bytes as usize / records_and_offsets.len();
let max_change = *this.avg_record_size / 10; // allow max 10% change
*this.avg_record_size = (*this.avg_record_size + new_avg.clamp(
    this.avg_record_size.saturating_sub(max_change),
    this.avg_record_size.saturating_add(max_change)
)) / 2;

• Sliding Window Average:
  Keep a record of the sizes of the last N batches and calculate their average.

this.record_sizes.push(fetch_bytes as usize / records_and_offsets.len());
if this.record_sizes.len() > 10 { // keep last 10 records
    this.record_sizes.remove(0);
}
*this.avg_record_size = this.record_sizes.iter().sum::<usize>() / this.record_sizes.len();

Originally posted by @fengjiachun in #4424 (comment)