@ParameterizedTest
    @MethodSource("batchSizes")
    public void testCoalesceMigrationRecords(Optional<Integer> configBatchSize, List<Integer> batchSizes, int expectedBatchCount, int expectedRecordCount) throws Exception {
        List<List<ApiMessageAndVersion>> batchesPassedToController = new ArrayList<>();
        NoOpRecordConsumer recordConsumer = new NoOpRecordConsumer() {
            @Override
            public CompletableFuture<?> acceptBatch(List<ApiMessageAndVersion> recordBatch) {
                batchesPassedToController.add(recordBatch);
                return CompletableFuture.completedFuture(null);
            }
        };
        CountingMetadataPropagator metadataPropagator = new CountingMetadataPropagator();
        CapturingMigrationClient migrationClient = CapturingMigrationClient.newBuilder()
            .setBrokersInZk(1, 2, 3)
            .setBatchSupplier(new CapturingMigrationClient.MigrationBatchSupplier() {
                @Override
                public List<List<ApiMessageAndVersion>> recordBatches() {
                    List<List<ApiMessageAndVersion>> batches = new ArrayList<>();
                    for (int batchSize : batchSizes) {
                        batches.add(fillBatch(batchSize));
                    }
                    return batches;
                }
            })
            .build();
        MockFaultHandler faultHandler = new MockFaultHandler("testRebatchMigrationRecords");

        KRaftMigrationDriver.Builder builder = defaultTestBuilder()
                .setZkMigrationClient(migrationClient)
                .setZkRecordConsumer(recordConsumer)
                .setPropagator(metadataPropagator)
                .setFaultHandler(faultHandler);
        configBatchSize.ifPresent(builder::setMinMigrationBatchSize);
        try (KRaftMigrationDriver driver = builder.build()) {
            MetadataImage image = MetadataImage.EMPTY;
            MetadataDelta delta = new MetadataDelta(image);

            startAndWaitForRecoveringMigrationStateFromZK(driver);
            setupDeltaForMigration(delta, true);
            delta.replay(ZkMigrationState.PRE_MIGRATION.toRecord().message());
            delta.replay(zkBrokerRecord(1));
            delta.replay(zkBrokerRecord(2));
            delta.replay(zkBrokerRecord(3));
            MetadataProvenance provenance = new MetadataProvenance(100, 1, 1);
            image = delta.apply(provenance);

            driver.onControllerChange(new LeaderAndEpoch(OptionalInt.of(3000), 1));

            driver.onMetadataUpdate(delta, image, logDeltaManifestBuilder(provenance,
                    new LeaderAndEpoch(OptionalInt.of(3000), 1)).build());
            driver.onMetadataUpdate(delta, image, logDeltaManifestBuilder(provenance,
                    new LeaderAndEpoch(OptionalInt.of(3000), 1)).build());

            TestUtils.waitForCondition(() -> driver.migrationState().get(1, TimeUnit.MINUTES).equals(MigrationDriverState.DUAL_WRITE),
                    "Waiting for KRaftMigrationDriver to enter DUAL_WRITE state");

            assertEquals(expectedBatchCount, batchesPassedToController.size());
            assertEquals(expectedRecordCount, batchesPassedToController.stream().mapToInt(List::size).sum());
        }
    }
Don't send RPCs to brokers for every metadata change, only when brokers or topics change.
This is a regression test for KAFKA-14668
metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java