static ControllerResult<Void> recordsForNonEmptyLog(
        Consumer<String> activationMessageConsumer,
        long transactionStartOffset,
        boolean zkMigrationEnabled,
        FeatureControlManager featureControl,
        MetadataVersion metadataVersion
    ) {
        StringBuilder logMessageBuilder = new StringBuilder("Performing controller activation. ");

        // Logs have been replayed. We need to initialize some things here if upgrading from older KRaft versions
        List<ApiMessageAndVersion> records = new ArrayList<>();

        // Check for in-flight transaction
        if (transactionStartOffset != -1L) {
            if (!metadataVersion.isMetadataTransactionSupported()) {
                throw new RuntimeException("Detected in-progress transaction at offset " + transactionStartOffset +
                    ", but the metadata.version " + metadataVersion +
                    " does not support transactions. Cannot continue.");
            } else {
                logMessageBuilder
                    .append("Aborting in-progress metadata transaction at offset ")
                    .append(transactionStartOffset)
                    .append(". ");
                records.add(new ApiMessageAndVersion(
                    new AbortTransactionRecord().setReason("Controller failover"), (short) 0));
            }
        }

        if (metadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
            logMessageBuilder.append("No metadata.version feature level record was found in the log. ")
                .append("Treating the log as version ")
                .append(MetadataVersion.MINIMUM_KRAFT_VERSION)
                .append(". ");
        }

        if (zkMigrationEnabled && !metadataVersion.isMigrationSupported()) {
            throw new RuntimeException("Should not have ZK migrations enabled on a cluster running " +
                "metadata.version " + featureControl.metadataVersion());
        } else if (metadataVersion.isMigrationSupported()) {
            logMessageBuilder
                .append("Loaded ZK migration state of ")
                .append(featureControl.zkMigrationState())
                .append(". ");
            switch (featureControl.zkMigrationState()) {
                case NONE:
                    // Since this is the default state there may or may not be an actual NONE in the log. Regardless,
                    // it will eventually be persisted in a snapshot, so we don't need to explicitly write it here.
                    if (zkMigrationEnabled) {
                        throw new RuntimeException("Should not have ZK migrations enabled on a cluster that was " +
                            "created in KRaft mode.");
                    }
                    logMessageBuilder.append("This is expected because this is a de-novo KRaft cluster.");
                    break;
                case PRE_MIGRATION:
                    if (!metadataVersion.isMetadataTransactionSupported()) {
                        logMessageBuilder
                            .append("Activating pre-migration controller without empty log. ")
                            .append("There may be a partial migration. ");
                    }
                    break;
                case MIGRATION:
                    if (!zkMigrationEnabled) {
                        // This can happen if controller leadership transfers to a controller with migrations enabled
                        // after another controller had finalized the migration. For example, during a rolling restart
                        // of the controller quorum during which the migration config is being set to false.
                        logMessageBuilder
                            .append("Completing the ZK migration since this controller was configured with ")
                            .append("'zookeeper.metadata.migration.enable' set to 'false'. ");
                        records.add(ZkMigrationState.POST_MIGRATION.toRecord());
                    } else {
                        // This log message is used in zookeeper_migration_test.py
                        logMessageBuilder
                            .append("Staying in ZK migration mode since 'zookeeper.metadata.migration.enable' ")
                            .append("is still 'true'. ");
                    }
                    break;
                case POST_MIGRATION:
                    if (zkMigrationEnabled) {
                        logMessageBuilder
                            .append("Ignoring 'zookeeper.metadata.migration.enable' value of 'true' since ")
                            .append("the ZK migration has been completed. ");
                    }
                    break;
                default:
                    throw new IllegalStateException("Unsupported ZkMigrationState " + featureControl.zkMigrationState());
            }
        }

        activationMessageConsumer.accept(logMessageBuilder.toString().trim());
        return ControllerResult.atomicOf(records, null);
    }
@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());
    }