@Test
    public void testActivationMessageForNonEmptyLogWithMigrations() {
        ControllerResult<Void> result;

        assertEquals(
            "Should not have ZK migrations enabled on a cluster running metadata.version 3.3-IV0",
            assertThrows(RuntimeException.class, () ->
                ActivationRecordsGenerator.recordsForNonEmptyLog(
                    logMsg -> fail(),
                    -1L,
                    true,
                    buildFeatureControl(MetadataVersion.IBP_3_3_IV0, Optional.empty()),
                    MetadataVersion.IBP_3_3_IV0
                )).getMessage()
        );

        assertEquals(
            "Should not have ZK migrations enabled on a cluster that was created in KRaft mode.",
            assertThrows(RuntimeException.class, () ->
                ActivationRecordsGenerator.recordsForNonEmptyLog(
                    logMsg -> fail(),
                    -1L,
                    true,
                    buildFeatureControl(MetadataVersion.IBP_3_4_IV0, Optional.empty()),
                    MetadataVersion.IBP_3_4_IV0
                )
            ).getMessage()
        );

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Loaded ZK migration state of " +
                "PRE_MIGRATION. Activating pre-migration controller without empty log. There may be a partial " +
                "migration.", logMsg),
            -1L,
            true,
            buildFeatureControl(MetadataVersion.IBP_3_4_IV0, Optional.of(ZkMigrationState.PRE_MIGRATION)),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(0, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Loaded ZK migration state of " +
                "PRE_MIGRATION.", logMsg),
            -1L,
            true,
            buildFeatureControl(MetadataVersion.IBP_3_6_IV1, Optional.of(ZkMigrationState.PRE_MIGRATION)),
            MetadataVersion.IBP_3_6_IV1
        );
        assertTrue(result.isAtomic());
        assertEquals(0, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Loaded ZK migration state of MIGRATION. " +
                "Staying in ZK migration mode since 'zookeeper.metadata.migration.enable' is still 'true'.", logMsg),
            -1L,
            true,
            buildFeatureControl(MetadataVersion.IBP_3_4_IV0, Optional.of(ZkMigrationState.MIGRATION)),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(0, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Loaded ZK migration state of MIGRATION. " +
                "Completing the ZK migration since this controller was configured with " +
                "'zookeeper.metadata.migration.enable' set to 'false'.", logMsg),
            -1L,
            false,
            buildFeatureControl(MetadataVersion.IBP_3_4_IV0, Optional.of(ZkMigrationState.MIGRATION)),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(1, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Aborting in-progress metadata " +
                "transaction at offset 42. Loaded ZK migration state of MIGRATION. Completing the ZK migration " +
                "since this controller was configured with 'zookeeper.metadata.migration.enable' set to 'false'.", logMsg),
            42L,
            false,
            buildFeatureControl(MetadataVersion.IBP_3_6_IV1, Optional.of(ZkMigrationState.MIGRATION)),
            MetadataVersion.IBP_3_6_IV1
        );
        assertTrue(result.isAtomic());
        assertEquals(2, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Loaded ZK migration state of " +
                "POST_MIGRATION.", logMsg),
            -1L,
            false,
            buildFeatureControl(MetadataVersion.IBP_3_4_IV0, Optional.of(ZkMigrationState.POST_MIGRATION)),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(0, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Aborting in-progress metadata " +
                "transaction at offset 42. Loaded ZK migration state of POST_MIGRATION.", logMsg),
            42L,
            false,
            buildFeatureControl(MetadataVersion.IBP_3_6_IV1, Optional.of(ZkMigrationState.POST_MIGRATION)),
            MetadataVersion.IBP_3_6_IV1
        );
        assertTrue(result.isAtomic());
        assertEquals(1, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Loaded ZK migration state of " +
                "POST_MIGRATION. Ignoring 'zookeeper.metadata.migration.enable' value of 'true' since the " +
                "ZK migration has been completed.", logMsg),
            -1L,
            true,
            buildFeatureControl(MetadataVersion.IBP_3_4_IV0, Optional.of(ZkMigrationState.POST_MIGRATION)),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(0, result.records().size());

        result = ActivationRecordsGenerator.recordsForNonEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Aborting in-progress metadata " +
                "transaction at offset 42. Loaded ZK migration state of POST_MIGRATION. Ignoring " +
                "'zookeeper.metadata.migration.enable' value of 'true' since the ZK migration has been completed.", logMsg),
            42L,
            true,
            buildFeatureControl(MetadataVersion.IBP_3_6_IV1, Optional.of(ZkMigrationState.POST_MIGRATION)),
            MetadataVersion.IBP_3_6_IV1
        );
        assertTrue(result.isAtomic());
        assertEquals(1, result.records().size());
    }
This class is for testing the log message or exception produced by ActivationRecordsGenerator. For tests that
verify the semantics of the returned records, see QuorumControllerTest.
metadata/src/test/java/org/apache/kafka/controller/ActivationRecordsGeneratorTest.java