Implementation of ReplicationThrottler interface

[garyparrot]

這個 PR 實作對 replication 進行限流的 API

Demo

public class Ignore extends RequireBrokerCluster {

  @Test
  void aaa() {
    var bootstrap = bootstrapServers();
    var topicName = Utils.randomString();
    try (Admin admin = Admin.of(bootstrap)) {
      // create topic
      System.out.println("Create topic");
      admin.creator().topic(topicName).numberOfPartitions(2).create();
      admin.migrator().partition(topicName, 0).moveTo(List.of(0));
      admin.migrator().partition(topicName, 1).moveTo(List.of(0));

      // send data
      System.out.println("Send data");
      Sender<byte[], byte[]> sender = Producer.of(bootstrap).sender();
      IntStream.range(0, 1_000_000)
          .mapToObj(x -> sender.topic(topicName).value(new byte[1024]).run())
          .map(CompletionStage::toCompletableFuture)
          .collect(Collectors.toUnmodifiableList())
          .forEach(i -> Utils.packException(() -> i.get()));

      // limit
      System.out.println("Limit bandwidth");
      var rate = DataRate.of(10, DataUnit.MiB, ChronoUnit.SECONDS);
      admin
          .replicationThrottler()
          .limitBrokerBandwidth(
              Map.of(
                  0, BrokerThrottleRate.of(rate, rate),
                  1, BrokerThrottleRate.of(rate, rate),
                  2, BrokerThrottleRate.of(rate, rate)));
      admin
          .replicationThrottler()
          .applyLogThrottle(
              ReplicaType.Leader,
              TopicThrottleSetting.allLogThrottled(topicName));
      admin
          .replicationThrottler()
          .applyLogThrottle(
              ReplicaType.Follower,
              TopicThrottleSetting.allLogThrottled(topicName));

      // Do migration
      System.out.println("Migration");
      admin.migrator().partition(topicName, 0).moveTo(List.of(0, 1));
      admin.migrator().partition(topicName, 1).moveTo(List.of(0, 1));
      Utils.sleep(Duration.ofMillis(1000));

      // sync monitor
      System.out.println("Monitor");
      ReplicaSyncingMonitor.main(new String[] {"--bootstrap.servers", bootstrap, "--track"});
    }
  }
}

[chia7712]

@garyparrot 感謝這個方便的功能，可否麻煩你也開一隻議題來討論如何實作WEB APIs?

@brandboat FYI，如果你有興趣接著把WEB APIs完成的話

[garyparrot]

@chia7712 這是目前預計的設計界面，你看看這樣有沒有問題，如果沒有問題的話我再完成具體實作。

[chia7712]

@garyparrot 這次能先討論介面是個很棒的流程，希望繼續保持XDD

另外請問一下replica.alter.log.dirs.io.max.bytes.per.second這個參數有要一併支援嗎？

[garyparrot]

@chia7712 我對 Throttle API 做了一些修正，你能再看一次嗎?

另外請問一下replica.alter.log.dirs.io.max.bytes.per.second這個參數有要一併支援嗎？

我從來沒有聽過這個 config 所以做了一些研究，看起來他是源自 https://cwiki.apache.org/confluence/display/KAFKA/KIP-113%3A+Support+replicas+movement+between+log+directories

1. Add config intra.broker.throttled.rate. This config specified the maximum rate in bytes-per-second that can be used to move replica between log directories. This config defaults to MAX_LONG. The intra.broker.throttled.rate is per-broker and the specified capacity is shared by all replica-movement-threads.

如果要加入也可以，不過要放在哪裡

1. 放在 ReplicationThrottler 裡面，這樣有人要用的時候可能比較方便，畢竟這個和 throttle 也有點關係，雖然這個參數和 replication 無關
2. 或是做成額外的 Admin API，不過我怕如果給每個潛在的參數都做類似的事情會使 Admin API 的界面膨脹? 或是說其實這些參數的數量沒有很多，能直接這樣做。
3. 新則一個 ReplicaMigrator#moveto(Map<Integer, Pair<String, DataRate>>)，能夠順便在指出要搬移時順便限流

我覺得就邏輯上來說這個東西可能要放在 replication throttler 外面，不過如果要避免 API 變成遙控器的話，可能要放在既有的東西上，看起來 3 可能比較好一些

[chia7712]

新則一個 ReplicaMigrator#moveto(Map<Integer, Pair<String, DataRate>>)，能夠順便在指出要搬移時順便限流
我覺得就邏輯上來說這個東西可能要放在 replication throttler 外面，不過如果要避免 API 變成遙控器的話，可能要放在既有的東西上，看起來 3 可能比較好一些

這個概念聽起來不錯，可否開個議題追蹤?

[chia7712]

@garyparrot 介面設計的很棒，我覺得可以按圖施工了。只有幾個小建議

[chia7712]

LGTM

@garyparrot 你可以先合併當作介面，或是要在這邊一次完成實作都可以

[garyparrot]

@chia7712

Related comment: #478 (comment)

我剛剛看到 replication throttle 設定的程式碼文件，他的流量是不是限制到每個 topic 或是整個 broker?

https://github.com/apache/kafka/blob/99b9b3e84f4e98c3f07714e1de6a139a004cbc5b/core/src/main/scala/kafka/server/DynamicConfig.scala#L47-L49

"A long representing the upper bound (bytes/sec) on replication traffic for leaders enumerated in the " +
s"property ${LogConfig.LeaderReplicationThrottledReplicasProp} (for each topic). This property can be only set dynamically. It is suggested that the " +
s"limit be kept above 1MB/s for accurate behaviour."

[chia7712]

我剛剛看到 replication throttle 設定的程式碼文件，他的流量是不是限制到每個 topic 或是整個 broker?

Topic端需要設定有哪些replica需要被限流，然後broker端設定流量上限，之後在該節點上那些被圈到的replicas的流量就會被限制住

[garyparrot]

如果我的理解沒錯，然後 Kip 73 的內容是正確的，這個限流是套用到整個節點而非 log 層級。
也就是每個節點，針對被限制的 replication target，他們的複製流量不能超過 x byte/s
而非三個 log 有 3x bytes/s

主要是限流是針對 broker 或是 topic
我覺得他的文件內看起來是, 現在有 3 個 log 會乘上變成 3x bytes/s，和我之前的理解不太一樣

[chia7712]

主要是限流是針對 broker 或是 topic
我覺得他的文件內看起來是, 現在有 3 個 log 會乘上變成 3x bytes/s，和我之前的理解不太一樣

你是說從原始碼來看嗎？我印象中應該不是 3x bytes，而是 x bytes

[garyparrot]

你是說從原始碼來看嗎？我印象中應該不是 3x bytes，而是 x bytes

thanks