Control the input size by setting kafka.fetch.max.{topic} configuration