低版本(modifyThrottleTaskByZKClient)执行后导致限流丢失follower.replication.throttled.replicas

[mgjiang]

###问题代码段

// 当前ZK上的迁移信息
scala.collection.Map<TopicPartition, Seq<Object>> currentParts = kafkaZkClient.getReplicaAssignmentForTopics(
proposedParts.keySet().map(elem -> elem.topic()).toSet()
);

// 转为moveMap格式
scala.collection.mutable.Map<String, scala.collection.mutable.Map<Object, ReassignPartitionsCommand.PartitionMove>> moveMap =
ReassignPartitionsCommand.calculateProposedMoveMap(new scala.collection.mutable.HashMap<>(), proposedParts, currentParts);

// 对Topic进行限流
scala.collection.Map<String, String> leaderThrottles = ReassignPartitionsCommand.calculateLeaderThrottles(moveMap);
scala.collection.Map<String, String> followerThrottles = ReassignPartitionsCommand.calculateFollowerThrottles(moveMap);
ReassignPartitionsCommand.modifyTopicThrottles(kafkaZkClient, leaderThrottles, followerThrottles);

问题描述

1. 当前ZK上的迁移信息这里获取的是topic的所有分区的副本信息 对于已经开始迁移任务，返回结果中开始迁移的分区副本会包含未完成的broker 即不再是迁移前的副本列表了
2. 这里转为moveMap格式时模仿启动迁移任务时的逻辑，但是方法calculateProposedMoveMap第一个参数依然传递的为空的Map,
3. 传递空Map 这样导致calculateProposedMoveMap方法内部diff后的结果为空

预期结果

处理掉这个bug

实际结果

修改限流信息 不再影响follower.replication.throttled.replicas

[mgjiang]

4.diff后结果为空最终导致calculateFollowerThrottles为空