Kafka添加副本时出现阻塞

问题可能是由于Kafka的副本管理器无法处理指定broker的Heartbeat请求和响应所致。可以通过增加broker的管理者数量解决问题，或者将没有响应的broker从集群中排除。

解法一：增加broker的管理者数量 在Kafka配置文件中设置broker的管理者数量（即“num.replica.fetchers”）为较高的数值（例如10），以增加副本管理器的处理能力。这可以通过以下代码实现：

# 修改Kafka配置文件（server.properties）
num.replica.fetchers=10

解法二：将没有响应的broker从集群中排除 可以通过协调器（coordinator）机制实现，例如使用Kafka的“ GroupCoordinator ”实例。如果一个broker在规定时间内没有响应，则可以使用GroupCoordinator将其从集群中排除。这可以通过以下代码实现：

# 修改Kafka配置文件（server.properties）
group.initial.rebalance.delay.ms=1000

# 新建一个GroupCoordinator实例
val groupCoordinator = new GroupCoordinator(…)
 
# 设置一个定时器，用于检查与broker的心跳状态
val timer = new Timer()
val task = new TimerTask() {
  def run() = {
    groupCoordinator.checkHeartbeat()
  }
}
timer.scheduleAtFixedRate(task, 0, 1000)