2022-02-15 带宽瓶颈导致kafka消费无法rebala

美西机器消费宁夏kafka集群 跨洋网络消费 公司带宽有限制为20m

应用消费topic多 concurrency高 两台机器加起来有200+线程同时消费

问题：重启机器后 kafka rebalance期间 两台机器cpu同时飙高99% 跨洋专线带宽占满 导致服务端收不到心跳 又重新开始rebalance 进入恶性循环

原因：每次rebalance完成后，所有消费者线程获知各自被分配的partition，同时去kafka服务器拉取消息，导致瞬间带宽被占满，
稳定消费时带宽占用没有这么高是由于各消费线程拉取消息的时机不是完全同步

解决：每次服务端kafka消费线程收到rebalance完成自己被分配的消费分区时，不马上开始拉取消息，而是阻塞一段时间，错开rebalance完成后第一次消费的时机，减少瞬间带宽占用

public class MyListener extends AbstractConsumerSeekAware {

    @KafkaListener(id = "xxx", topics = "yyy")
    public void listen(String in) {
        ...
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        super.onPartitionsAssigned(assignments, callback);
        // 线上环境 每次consumer线程被分配了消费分区之后 第一次消费前阻塞随机时间 减少带宽瞬间被占满的情况
        try {
            String[] arr = Optional.ofNullable(onPartitionsAssignedBlockRange).orElse("0,0").split(",");
            Thread.sleep(RandomUtils.nextInt(Integer.parseInt(arr[0]), Integer.parseInt(arr[1])));
        } catch (InterruptedException e) {
            log.error("sleep interrupted.", e);
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("onPartitionsAssigned exception", e);
        }
    }

}