源码解析：Kafka副本同步限流机制

发表于 2年以前 | 总阅读数：406 次

这一篇我们主要来看看分区副本重分配限流是如何实现的，此源码分析基于kafka2.5版本。在开始之前我们先来回顾一下分区重分配的流程，如图一所示。图一

执行分区重分配脚本，这里相当于起了个kafka客户端，执行完脚本将重分配数据写入zk即返回了。
kafkaController监听到zk分区重分配节点数据变动，会调相应的handler来处理数据，这里相当于kafka服务端大脑的角色，控制了分区重分配主要处理的流程，并发送命令给其他节点执行，描述的简单一点就是会根据重分配脚本在对应的broker 创建新的副本并删除需要移除的副本。
最后，kafkaController收到各个broker执行完操作的命令，会执行一些重分配结束后操作，包括修改zk 节点数据，更新元数据信息等。

大家也可以思考一下，如果把限流这个功能交给你，你会如何设计与开发？今天我们想聊的限流，就是在创建新的副本中实现的。在需要创建副本时，kafkaController会向对应的broker发送LeaderAndISRRequest请求，下面我们就从这里开始。

broker对LeaderAndIsr请求的处理

入口在kafka.server.KafkaApis.handleLeaderAndIsrRequest，处理LeaderAndIsrRequest 请求的代码就不放了，处理也比较简单，因为我们现在是做分区重分配，是先增加副本然后再下线删除不要的副本.所以会走里面makeFollowers方法的逻辑，在makeFollowers 中主要就是建立对应的文件夹，然后启对应的ReplicaFetcherThread来跟leader通信拉取数据，下面我们主要来分析一下ReplicaFetcherThread的代码。

这里主要就是构建FetchRequest然后处理FetchResponse，如果fetchRequestOpt为空，则会让线程等待一秒，

 private def maybeFetch(): Unit = {
      //同步数据，获取锁
      val fetchRequestOpt = inLock(partitionMapLock) {
        //2.1 构建同步数据request
        val ResultWithPartitions(fetchRequestOpt, partitionsWithError) = buildFetch(partitionStates.partitionStateMap.asScala)

        handlePartitionsWithErrors(partitionsWithError, "maybeFetch")

        if (fetchRequestOpt.isEmpty) {
          trace(s"There are no active partitions. Back off for $fetchBackOffMs ms before sending a fetch request")
          //如果发现没有需要构建的数据，则等待1秒
          partitionMapCond.await(fetchBackOffMs, TimeUnit.MILLISECONDS)
        }

        fetchRequestOpt
      }
      //2.2 在限流的情况下fetchRequestOpt是空，所以不会发请求，直接样本重置，或者达到限流标准
      fetchRequestOpt.foreach { case ReplicaFetch(sessionPartitions, fetchRequest) =>
        processFetchRequest(sessionPartitions, fetchRequest)
      }
    }

构建同步数据request代码分析

在这里，我们看到了关键代码shouldFollowerThrottle，下面我们进入到这个方法，看是如何实现限流的。

  override def buildFetch(partitionMap: Map[TopicPartition, PartitionFetchState]): ResultWithPartitions[Option[ReplicaFetch]] = {
    val partitionsWithError = mutable.Set[TopicPartition]()

    val builder = fetchSessionHandler.newBuilder(partitionMap.size, false)
    partitionMap.foreach { case (topicPartition, fetchState) =>
      // 2.1.1判断是否需要限流
      if (fetchState.isReadyForFetch && !shouldFollowerThrottle(quota, fetchState, topicPartition)) {
        try {
          val logStartOffset = this.logStartOffset(topicPartition)
          builder.add(topicPartition, new FetchRequest.PartitionData(
            fetchState.fetchOffset, logStartOffset, fetchSize, Optional.of(fetchState.currentLeaderEpoch)))
        } catch {
          case _: KafkaStorageException =>
            // The replica has already been marked offline due to log directory failure and the original failure should have already been logged.
            // This partition should be removed from ReplicaFetcherThread soon by ReplicaManager.handleLogDirFailure()
            partitionsWithError += topicPartition
        }
      }
    }

    val fetchData = builder.build()
    //判断如果fetchData是空的话直接返回None
    val fetchRequestOpt = if (fetchData.sessionPartitions.isEmpty && fetchData.toForget.isEmpty) {
      None
    } else {
      val requestBuilder = FetchRequest.Builder
        .forReplica(fetchRequestVersion, replicaId, maxWait, minBytes, fetchData.toSend)
        .setMaxBytes(maxBytes)
        .toForget(fetchData.toForget)
        .metadata(fetchData.metadata)
      Some(ReplicaFetch(fetchData.sessionPartitions(), requestBuilder))
    }

    ResultWithPartitions(fetchRequestOpt, partitionsWithError)
  }

shouldFollowerThrottle源码分析

这里代码实际很简单，有三个判断，第一个是判断副本是否同步，如果没同步则!fetchState.isReplicaInSync为true。第二个会去判断topicPartition是否在限流的内存配置中，这里的内存配置就是在分区重分配第一步时，将配置数据写入zk后触发的。

第三个就是真正的限流了，判断是否达到阙值。如果达到zk中设置的阙值，就会返回false，fetchData就会为空，在外层代码中会使等待一秒再重新请求。

private def shouldFollowerThrottle(quota: ReplicaQuota, fetchState: PartitionFetchState, topicPartition: TopicPartition): Boolean = {
    !fetchState.isReplicaInSync && quota.isThrottled(topicPartition) && quota.isQuotaExceeded
  }

在leader端同理，leader所在的broker会接收Fetch请求，入口在kafka.server.KafkaApis#handleFetchRequest，里面也有一段类似的代码，如下

 val shouldLeaderThrottleResult = shouldLeaderThrottle(quota, tp, replicaId)
          val fetchDataInfo = if (shouldLeaderThrottleResult) {
            // If the partition is being throttled, simply return an empty set.
            FetchDataInfo(readInfo.fetchedData.fetchOffsetMetadata, MemoryRecords.EMPTY)
          } else if (!hardMaxBytesLimit && readInfo.fetchedData.firstEntryIncomplete) {
            // For FetchRequest version 3, we replace incomplete message sets with an empty one as consumers can make
            // progress in such cases and don't need to report a `RecordTooLargeException`
            FetchDataInfo(readInfo.fetchedData.fetchOffsetMetadata, MemoryRecords.EMPTY)
          } else {
            readInfo.fetchedData
          }
   //此处省略若干代码       
 def shouldLeaderThrottle(quota: ReplicaQuota, topicPartition: TopicPartition, replicaId: Int): Boolean = {
    val isReplicaInSync = nonOfflinePartition(topicPartition).exists(_.inSyncReplicaIds.contains(replicaId))
    !isReplicaInSync && quota.isThrottled(topicPartition) && quota.isQuotaExceeded
  }

限流的实现

从上面的代码分析我们知道了限流是在哪一步代码中控制的，却不知道是怎么实现的，这里我们单独来讲解一下，具体见图二

Metrics及KafkaMetric类说明

这里我只画出了简单的组合关系，限流有关的是从Metrics类开始的，Metrics类相当于全局的限流工厂类，里面有两个ConcurrentMap分别装载KafkaMetric及Sensor，我们先来看看KafkaMetric ，里面主要就是初始化一些参数，然后提供获取当前值的方法，KafkaMetric的作用就是对一种限流类型的封装。

    double measurableValue(long timeMs) {
        synchronized (this.lock) {
            if (this.metricValueProvider instanceof Measurable)
                return ((Measurable) metricValueProvider).measure(config, timeMs);
            else
                return 0;
        }
    }

在Kafka服务端启动时会初始化一个全局的Metrics及QuotaManagers限流信息管理类，可以看到QuotaFactory.instantiate中分别初始化了leader及follower的限流管理类，QuotaManagers或者是ReplicationQuotaManager对象就会一层一层作为入参传入到其他方法使用，是单例的。

def startup(): Unit = {
    try {
      //省略若干代码
        metrics = new Metrics(metricConfig, reporters, time, true)

        /* register broker metrics */
        _brokerTopicStats = new BrokerTopicStats

        quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse(""))
        //省略若干代码
    }
    catch {
     //省略若干代码
    }
  }

  def instantiate(cfg: KafkaConfig, metrics: Metrics, time: Time, threadNamePrefix: String): QuotaManagers = {

      val clientQuotaCallback = Option(cfg.getConfiguredInstance(KafkaConfig.ClientQuotaCallbackClassProp,
        classOf[ClientQuotaCallback]))
      QuotaManagers(
        new ClientQuotaManager(clientFetchConfig(cfg), metrics, Fetch, time, threadNamePrefix, clientQuotaCallback),
        new ClientQuotaManager(clientProduceConfig(cfg), metrics, Produce, time, threadNamePrefix, clientQuotaCallback),
        new ClientRequestQuotaManager(clientRequestConfig(cfg), metrics, time, threadNamePrefix, clientQuotaCallback),
        new ReplicationQuotaManager(replicationConfig(cfg), metrics, LeaderReplication, time),
        new ReplicationQuotaManager(replicationConfig(cfg), metrics, FollowerReplication, time),
        new ReplicationQuotaManager(alterLogDirsReplicationConfig(cfg), metrics, AlterLogDirsReplication, time),
        clientQuotaCallback
      )
    }

Sensor类说明

对于Sensor，这个类的作用就比较多了，代码如下，由于篇幅的关系这里并没有放完整的代码，这里提供了两个重要的方法，一个是record，会在关键代码处调用此方法记录流量值，另一个是checkQuotas ，判断是否达到阙值.

这里可能会有小伙伴疑惑，为什么在Metrics类中装载了KafkaMetric，而Sensor又有个Map<MetricName, KafkaMetric> metrics对象，可以这样理解，Metrics中装载的是所有的KafkaMetric对象，而在Sensor中是针对一种记录值特有的限制。

public final class Sensor {

    private final Metrics registry;
    private final String name;
    private final Sensor[] parents;
    private final List<Stat> stats;
    private final Map<MetricName, KafkaMetric> metrics;
    public final MetricConfig config;
    private final Time time;
    private volatile long lastRecordTime;
    private final long inactiveSensorExpirationTimeMs;
    private final Object metricLock;

    public void record(double value, long timeMs) {
        record(value, timeMs, true);
    }
    public void checkQuotas(long timeMs) {
        for (KafkaMetric metric : this.metrics.values()) {
            MetricConfig config = metric.config();
            if (config != null) {
                Quota quota = config.quota();
                if (quota != null) {
                     //获取当前计算的流量值
                    double value = metric.measurableValue(timeMs);
                    //与配置的流量值比对，如果不满足则抛异常
                    if (!quota.acceptable(value)) {
                        throw new QuotaViolationException(metric.metricName(), value,
                            quota.bound());
                    }
                }
            }
        }
    }</br>

Sensor#record的实现

代码在SampledStat的record方法中，代码的逻辑我已注解，主要就是根据传入的时间获取当前样本对象，如果超出了config 中配置的值，则增加样本，否则更新样本数据。

这里要特别说明一下，样本的数量及是否增加样本的时间都是在MetricConfig初始化的时候传入的，样本数量是samples默认为11，可通过配置replication.quota.window .num来修改，样本时间是timeWindow，默认为1秒，可以通过replication.quota.window.size.seconds配置来修改。

public void record(MetricConfig config, double value, long timeMs) {
        //记录流量，获取当前的样本
        Sample sample = current(timeMs);
        //如果过去了1秒，则增加样本
        if (sample.isComplete(timeMs, config))
            sample = advance(config, timeMs);
        //否则，更新当前样本值
        update(sample, config, value, timeMs);
        sample.eventCount += 1;
    }

Sensor#checkQuotas的实现

前面我们知道在checkQuotas中主要就是获取当前的流量值与配置的比对，而计算当前流量值就是在Rate#measure方法中实现的，在measure方法里面主要是会调用SampledStat#measure，在SampledStat#measure方法中会调用purgeObsoleteSamples 方法来重置过期的副本.

然后再调用具体的combine方法，我们这里combine调用的是在WindowedSum中实现的，也就是循环各个样本累加在一起，回到Rate#measure 中，发现逻辑无非就是统计各个样本中的流量相加，然后除以时间来计算平均流量。

    //Rate#measure
    public double measure(MetricConfig config, long now) {
        double value = stat.measure(config, now);
        return value / convert(windowSize(config, now));
    }
    //SampledStat#measure
    public double measure(MetricConfig config, long now) {
        purgeObsoleteSamples(config, now);
        return combine(this.samples, config, now);
    }
    protected void purgeObsoleteSamples(MetricConfig config, long now) {
            long expireAge = config.samples() * config.timeWindowMs();
            //如果样本数据过去了11秒，则重置
            for (Sample sample : samples) {
                if (now - sample.lastWindowMs >= expireAge) {
                    //即11秒reset一次
                    System.out.println("reset====" + now + "----" + sample.lastWindowMs + "---" + expireAge);
                    sample.reset(now);
                }
            }
        }
     //WindowedSum#combine
    public double combine(List<Sample> samples, MetricConfig config, long now) {
         double total = 0.0;
         for (Sample sample : samples) {
             total += sample.value;
         }
         return total;
     }

流量是在哪儿记录的？

相信读到这里很多人都有这个疑问，在followers端是在FetchResponse返回的时候记录的，代码见图三

在leader端，是在读完日志文件之后会记录流量值

可以看到在follower端需要判断是否在限流副本中才记录流量值，而在leader端则没有这个限制，不知道这是不是kafka的bug？针对这个问题我也做了测试，的确是在leader端被同步副本时，不在限流配置中的topic 也会被记录流量值，不太能猜到这样做的意图。在这里我们也可以梳理一下Kafka限流的实现原理：

1、在broker启动的时候初始化限流管理类及全局的限流配置。

2、如果在zk节点中写入了leader及follower的限流配置，则会在同步副本时调用方法isQuotaExceeded来判断是否达到限流值.在这里我还想说明一下为什么在zk中写入leader.replication .throttled.replicas数据时为什么要包含原来所有的副本，因为在同步新副本的时候leader有可能会掉线然后重新选举leader，所以不如一次性全写入。

3、如果没有达到限流值则会在对应的地方分别记录流量，leader与follower的流量值是分开记录的。

4、判断是否达到限流值的方式就是记录最近一定数量的样本并计算平均值。

对于限流的几点思考

学到这里可能大家对限流代码逻辑有了一个基本的认识，我们下面就通过几个问题来让大家加深印象。

设置限流值为1是否永远都无法完成分区重分配任务？

不完全是，可以设想一下，如果副本数据小于一次fetch的值，leader跟follower之前也完全没有流量记录，那一次同步之后就结束了，样本那也只能记录一下这次的流量，但达到完全没有流量记录是十分苛责的，只存在于demo 版的kafka.

在follower端还可以达到，而在leader端，根据我的测试，平常完全没有数据的一次fetch也会返回18k的流量，一秒大概是36k的数据，所以如果数据大于一次fetch的上限，那在之后fetch的时候都会被leader端限制。

我们在设置限流值的时候一定要考虑副本的日常流量，这点在官方文档上也有提及，如果小于日常流量的话那将无法完成迁移。

针对leader的测试

准备3个broker，topic_1与topic_2，leader都在broker1上，topic_1迁移到broker2上，topic_2迁移到broker3上，副本数据为200M，限流值设置为300*1024b/s即300k/s
结果如下,200M = 2001024k,2001024k/300k/60≈11.37min，因为有两个topic都需要迁移，且leader端限制了传输速率，所以最终迁移持续时间大概为19分钟

[controller-event-thread] INFO kafka.controller.KafkaController - [Controller id=1] reassignment :Map(topic_1-0 -> ReplicaAssignment(replicas=1,2, addingReplicas=2, removingReplicas=), topic_2-0 -> ReplicaAssignment(replicas=1,3, addingReplicas=3, removingReplicas=)) start at 2021-11-13 13:18:20

   ......

[controller-event-thread] INFO kafka.controller.KafkaController - [Controller id=1] No more partitions need to be reassigned. Deleting zk path /admin/reassign_partitions at 2021-11-13 13:37:31

针对follower的测试

准备3个broker，topic_1与topic_2，topic_1的leader在broker1上，topic_2的leader在broker2上，均迁移到broker3，副本数据为200M，限流值设置为300*1024b/s
结果如下，这里跟上一个例子一样，原本只需要11分钟左右的，因为有两个topic都需要迁移，follower端限制了传输速率，所以最终迁移持续时间有21分钟

[controller-event-thread] INFO kafka.controller.KafkaController - [Controller id=1] reassignment :Map(topic_1-0 -> ReplicaAssignment(replicas=1,3, addingReplicas=3, removingReplicas=), topic_2-0 -> ReplicaAssignment(replicas=2,3, addingReplicas=3, removingReplicas=)) start at 2021-11-13 14:22:42

  ......

[controller-event-thread] INFO kafka.controller.KafkaController - [Controller id=1] No more partitions need to be reassigned. Deleting zk path /admin/reassign_partitions at 2021-11-13 14:43:17

限制互不影响的测试

准备3个broker，topic_1与topic_2，topic_1的leader在broker1上，topic_2的leader在broker2上，topic_1迁移到broker2，topic_2迁移到broker3 副本数据为200M，限流值设置为300*1024b/s
结果如下，可以看到leader与follower的针对不同topic互不影响的情况下就是11分钟完成迁移。

[controller-event-thread] INFO kafka.controller.KafkaController - [Controller id=1] reassignment :Map(topic_1-0 -> ReplicaAssignment(replicas=1,2, addingReplicas=2, removingReplicas=), topic_2-0 -> ReplicaAssignment(replicas=2,3, addingReplicas=3, removingReplicas=)) start at 2021-11-13 15:04:37

  ......

[controller-event-thread] INFO kafka.controller.KafkaController - [Controller id=1] No more partitions need to be reassigned. Deleting zk path /admin/reassign_partitions at 2021-11-13 15:15:40