void clearThrottles(List<ExecutionTask> completedTasks, List<ExecutionTask> inProgressTasks)
  throws ExecutionException, InterruptedException, TimeoutException {
    if (throttlingEnabled()) {
      List<ExecutionProposal> completedProposals =
        completedTasks
          .stream()
          // Filter for completed tasks related to inter-broker replica movement
          .filter(this::shouldRemoveThrottleForTask)
          .map(ExecutionTask::proposal)
          .collect(Collectors.toList());

      // These are the brokers which have completed a task with
      // inter-broker replica movement
      Set<Integer> participatingBrokers = getParticipatingBrokers(completedProposals);

      List<ExecutionProposal> inProgressProposals =
        inProgressTasks
          .stream()
          .filter(this::taskIsInProgress)
          .map(ExecutionTask::proposal)
          .collect(Collectors.toList());

      // These are the brokers which currently have in-progress
      // inter-broker replica movement
      Set<Integer> brokersWithInProgressTasks = getParticipatingBrokers(inProgressProposals);

      // Remove the brokers with in-progress replica moves from the brokers that have
      // completed inter-broker replica moves
      Set<Integer> brokersToRemoveThrottlesFrom = new TreeSet<>(participatingBrokers);
      brokersToRemoveThrottlesFrom.removeAll(brokersWithInProgressTasks);

      LOG.info("Removing replica movement throttles from brokers in the cluster: {}", brokersToRemoveThrottlesFrom);
      for (int broker : brokersToRemoveThrottlesFrom) {
        removeThrottledRateFromBroker(broker);
      }

      Map<String, Set<String>> throttledReplicas = getThrottledReplicasByTopic(completedProposals);
      for (Map.Entry<String, Set<String>> entry : throttledReplicas.entrySet()) {
        removeThrottledReplicasFromTopic(entry.getKey(), entry.getValue());
      }
    }
  }
@Test
  public void testClearThrottleOnNonExistentTopic() throws Exception {
    final long throttleRate = 100L;
    final int brokerId0 = 0;
    final int brokerId1 = 1;
    final int brokerId2 = 2;
    final List<Integer> brokers = Arrays.asList(brokerId0, brokerId1, brokerId2);
    final int partitionId = 0;
    // A proposal to move a partition with 2 replicas from broker 0 and 1 to broker 0 and 2
    ExecutionProposal proposal = new ExecutionProposal(new TopicPartition(TOPIC0, partitionId),
                                                       100,
                                                       new ReplicaPlacementInfo(brokerId0),
                                                       Arrays.asList(new ReplicaPlacementInfo(brokerId0), new ReplicaPlacementInfo(brokerId1)),
                                                       Arrays.asList(new ReplicaPlacementInfo(brokerId0), new ReplicaPlacementInfo(brokerId2)));

    AdminClient mockAdminClient = EasyMock.mock(AdminClient.class);
    ReplicationThrottleHelper throttleHelper = new ReplicationThrottleHelper(mockAdminClient, throttleRate);

    // Case 1: a situation where Topic0 does not exist. Hence no property is returned upon read.

    // Before the dynamic broker config is removed for the follower throttle rate
    Config brokerConfig = new Config(Arrays.asList(
            mockConfigEntry(ReplicationThrottleHelper.LEADER_THROTTLED_RATE, "200", ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG),
            mockConfigEntry(ReplicationThrottleHelper.FOLLOWER_THROTTLED_RATE, "200", ConfigEntry.ConfigSource.DYNAMIC_BROKER_CONFIG)));
    // After the dynamic broker config is removed for the follower throttle rate
    Config brokerConfig2 = new Config(Arrays.asList(
            mockConfigEntry(ReplicationThrottleHelper.LEADER_THROTTLED_RATE, "200", ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG),
            mockConfigEntry(ReplicationThrottleHelper.FOLLOWER_THROTTLED_RATE, "300", ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG)));
    // Expect that only the dynamic throttle rate configs are removed when clearing throttles
    expectDescribeBrokerConfigs(mockAdminClient, brokers, brokerConfig);
    expectIncrementalBrokerConfigs(mockAdminClient, brokers);
    expectDescribeBrokerConfigs(mockAdminClient, brokers, brokerConfig2);
    expectDescribeTopicConfigs(mockAdminClient, TOPIC0, EMPTY_CONFIG, false);
    expectListTopics(mockAdminClient, Collections.emptySet());
    ExecutionTask mockCompleteTask = prepareMockCompleteTask(proposal);
    EasyMock.replay(mockAdminClient);

    throttleHelper.clearThrottles(Collections.singletonList(mockCompleteTask), Collections.emptyList());
    EasyMock.verify(mockAdminClient, mockCompleteTask);

    // Case 2: a situation where Topic0 gets deleted after its configs were read.
    EasyMock.reset(mockAdminClient);
    expectDescribeBrokerConfigs(mockAdminClient, brokers);
    expectIncrementalBrokerConfigs(mockAdminClient, brokers);
    expectDescribeBrokerConfigs(mockAdminClient, brokers, EMPTY_CONFIG);
    String throttledReplicas = brokerId0 + "," + brokerId1;
    Config topicConfigProps = new Config(Arrays.asList(
            new ConfigEntry(ReplicationThrottleHelper.LEADER_THROTTLED_REPLICAS, throttledReplicas),
            new ConfigEntry(ReplicationThrottleHelper.FOLLOWER_THROTTLED_REPLICAS, throttledReplicas)));
    expectDescribeTopicConfigs(mockAdminClient, TOPIC0, topicConfigProps, true);
    expectIncrementalTopicConfigs(mockAdminClient, TOPIC0, false);
    expectListTopics(mockAdminClient, Collections.emptySet());
    mockCompleteTask = prepareMockCompleteTask(proposal);
    EasyMock.replay(mockAdminClient);

    // Expect no exception
    throttleHelper.clearThrottles(Collections.singletonList(mockCompleteTask), Collections.emptyList());
    EasyMock.verify(mockAdminClient, mockCompleteTask);
  }