@Test
    public void testFailoverDuringMigrationTransaction() throws Exception {
        try (
            LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3).build()
        ) {
            QuorumControllerTestEnv.Builder controlEnvBuilder = new QuorumControllerTestEnv.Builder(logEnv).
                setControllerBuilderInitializer(controllerBuilder -> controllerBuilder.setZkMigrationEnabled(true)).
                setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test"));
            QuorumControllerTestEnv controlEnv = controlEnvBuilder.build();
            QuorumController active = controlEnv.activeController(true);
            ZkRecordConsumer migrationConsumer = active.zkRecordConsumer();
            migrationConsumer.beginMigration().get(30, TimeUnit.SECONDS);
            migrationConsumer.acceptBatch(ZK_MIGRATION_RECORDS).get(30, TimeUnit.SECONDS);
            forceRenounce(active);

            // Ensure next controller doesn't see the topic from partial migration
            QuorumController newActive = controlEnv.activeController(true);
            CompletableFuture<Map<String, ResultOrError<Uuid>>> results =
                newActive.findTopicIds(ANONYMOUS_CONTEXT, Collections.singleton("spam"));
            assertEquals(
                Errors.UNKNOWN_TOPIC_OR_PARTITION,
                results.get(30, TimeUnit.SECONDS).get("spam").error().error());

            assertEquals(
                ZkMigrationState.PRE_MIGRATION,
                newActive.appendReadEvent("read migration state", OptionalLong.empty(),
                    () -> newActive.featureControl().zkMigrationState()
                ).get(30, TimeUnit.SECONDS)
            );
            // Ensure the migration can happen on new active controller
            migrationConsumer = newActive.zkRecordConsumer();
            migrationConsumer.beginMigration().get(30, TimeUnit.SECONDS);
            migrationConsumer.acceptBatch(ZK_MIGRATION_RECORDS).get(30, TimeUnit.SECONDS);
            migrationConsumer.completeMigration().get(30, TimeUnit.SECONDS);

            results = newActive.findTopicIds(ANONYMOUS_CONTEXT, Collections.singleton("spam"));
            assertTrue(results.get(30, TimeUnit.SECONDS).get("spam").isResult());

            assertEquals(ZkMigrationState.MIGRATION, newActive.appendReadEvent("read migration state", OptionalLong.empty(),
                () -> newActive.featureControl().zkMigrationState()).get(30, TimeUnit.SECONDS));

        }
    }
Tests all intermediate images lead to the same final image for each image & delta type.
@param fromRecords
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java