@ClusterTest
    public void testProduceAndConsumeWithReassignmentInProgress() throws Exception {
        createTopics();
        produceMessages("baz", 2, 60);
        String assignment = "{\"version\":1,\"partitions\":" +
                "[{\"topic\":\"baz\",\"partition\":2,\"replicas\":[3,2,1],\"log_dirs\":[\"any\",\"any\",\"any\"]}" +
                "]}";
        runExecuteAssignment(false, assignment, 300L, -1L);
        produceMessages("baz", 2, 100);

        Properties consumerProps = new Properties();
        consumerProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers());
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        TopicPartition part = new TopicPartition("baz", 2);
        try (Consumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
            consumer.assign(singleton(part));
            List<ConsumerRecord<byte[], byte[]>> allRecords = new ArrayList<>();
            TestUtils.waitForCondition(() -> {
                ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100L));
                records.records("baz").forEach(allRecords::add);
                return allRecords.size() >= 100;
            }, "Timeout for waiting enough records");
        }
        removeReplicationThrottleForPartitions(part);
        Map<TopicPartition, PartitionReassignmentState> finalAssignment = singletonMap(part,
                new PartitionReassignmentState(asList(3, 2, 1), asList(3, 2, 1), true));
        try (Admin admin = Admin.create(Collections.singletonMap(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers())))  {
            waitForVerifyAssignment(admin, assignment, false,
                    new VerifyAssignmentResult(finalAssignment));
        }
    }
Test running a reassignment with the interBrokerThrottle set.
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java