@ClusterTest
    public void testThrottledReassignment() throws Exception {
        createTopics();
        produceMessages("foo", 0, 50);
        produceMessages("baz", 2, 60);
        String assignment = "{\"version\":1,\"partitions\":" +
                "[{\"topic\":\"foo\",\"partition\":0,\"replicas\":[0,3,2],\"log_dirs\":[\"any\",\"any\",\"any\"]}," +
                "{\"topic\":\"baz\",\"partition\":2,\"replicas\":[3,2,1],\"log_dirs\":[\"any\",\"any\",\"any\"]}" +
                "]}";

        // Check that the assignment has not yet been started yet.
        Map<TopicPartition, PartitionReassignmentState> initialAssignment = new HashMap<>();
        initialAssignment.put(new TopicPartition("foo", 0),
                new PartitionReassignmentState(asList(0, 1, 2), asList(0, 3, 2), true));
        initialAssignment.put(new TopicPartition("baz", 2),
                new PartitionReassignmentState(asList(0, 2, 1), asList(3, 2, 1), true));
        try (Admin admin = Admin.create(Collections.singletonMap(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
            assertEquals(new VerifyAssignmentResult(initialAssignment), runVerifyAssignment(admin, assignment, false));
            assertEquals(unthrottledBrokerConfigs, describeBrokerLevelThrottles(admin, unthrottledBrokerConfigs.keySet()));

            // Execute the assignment
            long interBrokerThrottle = 300000L;
            runExecuteAssignment(false, assignment, interBrokerThrottle, -1L);
            waitForInterBrokerThrottle(admin, asList(0, 1, 2, 3), interBrokerThrottle);

            Map<TopicPartition, PartitionReassignmentState> finalAssignment = new HashMap<>();
            finalAssignment.put(new TopicPartition("foo", 0),
                    new PartitionReassignmentState(asList(0, 3, 2), asList(0, 3, 2), true));
            finalAssignment.put(new TopicPartition("baz", 2),
                    new PartitionReassignmentState(asList(3, 2, 1), asList(3, 2, 1), true));

            // Wait for the assignment to complete
            TestUtils.waitForCondition(() -> {
                // Check the reassignment status.
                VerifyAssignmentResult result = runVerifyAssignment(admin, assignment, true);

                if (!result.partsOngoing) {
                    return true;
                } else {
                    assertFalse(
                            result.partStates.values().stream().allMatch(state -> state.done),
                            "Expected at least one partition reassignment to be ongoing when result = " + result
                    );
                    assertEquals(asList(0, 3, 2), result.partStates.get(new TopicPartition("foo", 0)).targetReplicas);
                    assertEquals(asList(3, 2, 1), result.partStates.get(new TopicPartition("baz", 2)).targetReplicas);
                    waitForInterBrokerThrottle(admin, asList(0, 1, 2, 3), interBrokerThrottle);
                    return false;
                }
            }, "Expected reassignment to complete.");

            waitForVerifyAssignment(admin, assignment, true,
                    new VerifyAssignmentResult(finalAssignment));
            // The throttles should still have been preserved, since we ran with --preserve-throttles
            waitForInterBrokerThrottle(admin, asList(0, 1, 2, 3), interBrokerThrottle);
            // Now remove the throttles.
            waitForVerifyAssignment(admin, assignment, false,
                    new VerifyAssignmentResult(finalAssignment));
            waitForBrokerLevelThrottles(admin, unthrottledBrokerConfigs);
        }
    }
Test running a reassignment with the interBrokerThrottle set.
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java