@ClusterTest
    public void testCancellation() throws Exception {
        createTopics();
        TopicPartition foo0 = new TopicPartition("foo", 0);
        TopicPartition baz1 = new TopicPartition("baz", 1);

        produceMessages(foo0.topic(), foo0.partition(), 200);
        produceMessages(baz1.topic(), baz1.partition(), 200);
        String assignment = "{\"version\":1,\"partitions\":" +
                "[{\"topic\":\"foo\",\"partition\":0,\"replicas\":[0,1,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}," +
                "{\"topic\":\"baz\",\"partition\":1,\"replicas\":[0,2,3],\"log_dirs\":[\"any\",\"any\",\"any\"]}" +
                "]}";
        try (Admin admin = Admin.create(Collections.singletonMap(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
            assertEquals(unthrottledBrokerConfigs,
                    describeBrokerLevelThrottles(admin, unthrottledBrokerConfigs.keySet()));
            long interBrokerThrottle = 1L;
            runExecuteAssignment(false, assignment, interBrokerThrottle, -1L);
            waitForInterBrokerThrottle(admin, asList(0, 1, 2, 3), interBrokerThrottle);

            Map<TopicPartition, PartitionReassignmentState> partStates = new HashMap<>();

            partStates.put(foo0, new PartitionReassignmentState(asList(0, 1, 3, 2), asList(0, 1, 3), false));
            partStates.put(baz1, new PartitionReassignmentState(asList(0, 2, 3, 1), asList(0, 2, 3), false));

            // Verify that the reassignment is running.  The very low throttle should keep it
            // from completing before this runs.
            waitForVerifyAssignment(admin, assignment, true,
                    new VerifyAssignmentResult(partStates, true, Collections.emptyMap(), false));
            // Cancel the reassignment.
            assertEquals(new AbstractMap.SimpleImmutableEntry<>(new HashSet<>(asList(foo0, baz1)), Collections.emptySet()), runCancelAssignment(assignment, true));
            // Broker throttles are still active because we passed --preserve-throttles
            waitForInterBrokerThrottle(admin, asList(0, 1, 2, 3), interBrokerThrottle);
            // Cancelling the reassignment again should reveal nothing to cancel.
            assertEquals(new AbstractMap.SimpleImmutableEntry<>(Collections.emptySet(), Collections.emptySet()), runCancelAssignment(assignment, false));
            // This time, the broker throttles were removed.
            waitForBrokerLevelThrottles(admin, unthrottledBrokerConfigs);
            // Verify that there are no ongoing reassignments.
            assertFalse(runVerifyAssignment(admin, assignment, false).partsOngoing);
        }
        // Verify that the partition is removed from cancelled replicas
        verifyReplicaDeleted(new TopicPartitionReplica(foo0.topic(), foo0.partition(), 3));
        verifyReplicaDeleted(new TopicPartitionReplica(baz1.topic(), baz1.partition(), 3));
    }
Test running a reassignment and then cancelling it.
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java