public static void assignThrottledPartitionReplicas(Admin adminClient, Map<TopicPartition, List<Integer>> allReplicasByPartition) throws InterruptedException, ExecutionException {
        Map<ConfigResource, List<Entry<TopicPartition, List<Integer>>>> configResourceToPartitionReplicas =
            allReplicasByPartition.entrySet().stream()
            .collect(Collectors.groupingBy(
                topicPartitionListEntry -> new ConfigResource(ConfigResource.Type.TOPIC, topicPartitionListEntry.getKey().topic()))
            );

        Map<ConfigResource, List<AlterConfigOp>> throttles = configResourceToPartitionReplicas.entrySet().stream()
            .collect(
                Collectors.toMap(Entry::getKey, entry -> {
                    List<AlterConfigOp> alterConfigOps = new ArrayList<>();
                    Map<TopicPartition, List<Integer>> replicaThrottle =
                        entry.getValue().stream().collect(Collectors.toMap(Entry::getKey, Entry::getValue));
                    alterConfigOps.add(new AlterConfigOp(
                        new ConfigEntry(QuotaConfigs.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicaThrottle)),
                        AlterConfigOp.OpType.SET));
                    alterConfigOps.add(new AlterConfigOp(
                        new ConfigEntry(QuotaConfigs.FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG, formatReplicaThrottles(replicaThrottle)),
                        AlterConfigOp.OpType.SET));
                    return alterConfigOps;
                }
            ));
        adminClient.incrementalAlterConfigs(new HashMap<>(throttles)).all().get();
    }
Remove a set of throttled partitions and reset the overall replication quota.
tools/src/test/java/org/apache/kafka/tools/ToolsTestUtils.java