static ControllerResult<Void> recordsForEmptyLog(
        Consumer<String> activationMessageConsumer,
        long transactionStartOffset,
        boolean zkMigrationEnabled,
        BootstrapMetadata bootstrapMetadata,
        MetadataVersion metadataVersion
    ) {
        StringBuilder logMessageBuilder = new StringBuilder("Performing controller activation. ");
        List<ApiMessageAndVersion> records = new ArrayList<>();

        if (transactionStartOffset != -1L) {
            // In-flight bootstrap transaction
            if (!metadataVersion.isMetadataTransactionSupported()) {
                throw new RuntimeException("Detected partial bootstrap records transaction at " +
                    transactionStartOffset + ", but the metadata.version " + metadataVersion +
                    " does not support transactions. Cannot continue.");
            } else {
                logMessageBuilder
                    .append("Aborting partial bootstrap records transaction at offset ")
                    .append(transactionStartOffset)
                    .append(". Re-appending ")
                    .append(bootstrapMetadata.records().size())
                    .append(" bootstrap record(s) in new metadata transaction at metadata.version ")
                    .append(metadataVersion)
                    .append(" from bootstrap source '")
                    .append(bootstrapMetadata.source())
                    .append("'. ");
                records.add(new ApiMessageAndVersion(
                    new AbortTransactionRecord().setReason("Controller failover"), (short) 0));
                records.add(new ApiMessageAndVersion(
                    new BeginTransactionRecord().setName("Bootstrap records"), (short) 0));
            }
        } else {
            // No in-flight transaction
            logMessageBuilder
                .append("The metadata log appears to be empty. ")
                .append("Appending ")
                .append(bootstrapMetadata.records().size())
                .append(" bootstrap record(s) ");
            if (metadataVersion.isMetadataTransactionSupported()) {
                records.add(new ApiMessageAndVersion(
                    new BeginTransactionRecord().setName("Bootstrap records"), (short) 0));
                logMessageBuilder.append("in metadata transaction ");
            }
            logMessageBuilder
                .append("at metadata.version ")
                .append(metadataVersion)
                .append(" from bootstrap source '")
                .append(bootstrapMetadata.source())
                .append("'. ");
        }

        // If no records have been replayed, we need to write out the bootstrap records.
        // This will include the new metadata.version, as well as things like SCRAM
        // initialization, etc.
        records.addAll(bootstrapMetadata.records());

        if (metadataVersion.isMigrationSupported()) {
            if (zkMigrationEnabled) {
                logMessageBuilder.append("Putting the controller into pre-migration mode. No metadata updates " +
                    "will be allowed until the ZK metadata has been migrated. ");
                records.add(ZkMigrationState.PRE_MIGRATION.toRecord());
            } else {
                logMessageBuilder.append("Setting the ZK migration state to NONE since this is a de-novo " +
                    "KRaft cluster. ");
                records.add(ZkMigrationState.NONE.toRecord());
            }
        } else {
            if (zkMigrationEnabled) {
                throw new RuntimeException("The bootstrap metadata.version " + bootstrapMetadata.metadataVersion() +
                    " does not support ZK migrations. Cannot continue with ZK migrations enabled.");
            }
        }

        activationMessageConsumer.accept(logMessageBuilder.toString().trim());
        if (metadataVersion.isMetadataTransactionSupported()) {
            records.add(new ApiMessageAndVersion(new EndTransactionRecord(), (short) 0));
            return ControllerResult.of(records, null);
        } else {
            return ControllerResult.atomicOf(records, null);
        }
    }
@Test
    public void testActivationMessageForEmptyLog() {
        ControllerResult<Void> result;
        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. The metadata log appears to be empty. " +
                "Appending 1 bootstrap record(s) at metadata.version 3.0-IV1 from bootstrap source 'test'.", logMsg),
            -1L,
            false,
            BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION, "test"),
            MetadataVersion.MINIMUM_KRAFT_VERSION
        );
        assertTrue(result.isAtomic());
        assertEquals(1, result.records().size());

        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. The metadata log appears to be empty. " +
                "Appending 1 bootstrap record(s) at metadata.version 3.4-IV0 from bootstrap " +
                "source 'test'. Setting the ZK migration state to NONE since this is a de-novo KRaft cluster.", logMsg),
            -1L,
            false,
            BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_4_IV0, "test"),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(2, result.records().size());


        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. The metadata log appears to be empty. " +
                "Appending 1 bootstrap record(s) at metadata.version 3.4-IV0 from bootstrap " +
                "source 'test'. Putting the controller into pre-migration mode. No metadata updates will be allowed " +
                "until the ZK metadata has been migrated.", logMsg),
            -1L,
            true,
            BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_4_IV0, "test"),
            MetadataVersion.IBP_3_4_IV0
        );
        assertTrue(result.isAtomic());
        assertEquals(2, result.records().size());

        assertEquals(
            "The bootstrap metadata.version 3.3-IV2 does not support ZK migrations. Cannot continue with ZK migrations enabled.",
            assertThrows(RuntimeException.class, () ->
                ActivationRecordsGenerator.recordsForEmptyLog(
                    logMsg -> fail(),
                    -1L,
                    true,
                    BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV2, "test"),
                    MetadataVersion.IBP_3_3_IV2
                )).getMessage()
        );

        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. The metadata log appears to be empty. " +
                "Appending 1 bootstrap record(s) in metadata transaction at metadata.version 3.6-IV1 from bootstrap " +
                "source 'test'. Setting the ZK migration state to NONE since this is a de-novo KRaft cluster.", logMsg),
            -1L,
            false,
            BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test"),
            MetadataVersion.IBP_3_6_IV1
        );
        assertFalse(result.isAtomic());
        assertEquals(4, result.records().size());

        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. The metadata log appears to be empty. " +
                "Appending 1 bootstrap record(s) in metadata transaction at metadata.version 3.6-IV1 from bootstrap " +
                "source 'test'. Putting the controller into pre-migration mode. No metadata updates will be allowed " +
                "until the ZK metadata has been migrated.", logMsg),
            -1L,
            true,
            BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test"),
            MetadataVersion.IBP_3_6_IV1
        );
        assertFalse(result.isAtomic());
        assertEquals(4, result.records().size());

        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Aborting partial bootstrap records " +
                "transaction at offset 0. Re-appending 1 bootstrap record(s) in new metadata transaction at " +
                "metadata.version 3.6-IV1 from bootstrap source 'test'. Setting the ZK migration state to NONE " +
                "since this is a de-novo KRaft cluster.", logMsg),
            0L,
            false,
            BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test"),
            MetadataVersion.IBP_3_6_IV1
        );
        assertFalse(result.isAtomic());
        assertEquals(5, result.records().size());

        result = ActivationRecordsGenerator.recordsForEmptyLog(
            logMsg -> assertEquals("Performing controller activation. Aborting partial bootstrap records " +
                "transaction at offset 0. Re-appending 1 bootstrap record(s) in new metadata transaction at " +
                "metadata.version 3.6-IV1 from bootstrap source 'test'. Putting the controller into pre-migration " +
                "mode. No metadata updates will be allowed until the ZK metadata has been migrated.", logMsg),
            0L,
            true,
            BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test"),
            MetadataVersion.IBP_3_6_IV1
        );
        assertFalse(result.isAtomic());
        assertEquals(5, result.records().size());

        assertEquals(
            "Detected partial bootstrap records transaction at 0, but the metadata.version 3.6-IV0 does not " +
                "support transactions. Cannot continue.",
            assertThrows(RuntimeException.class, () ->
                ActivationRecordsGenerator.recordsForEmptyLog(
                    logMsg -> assertEquals("", logMsg),
                    0L,
                    true,
                    BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV0, "test"),
                    MetadataVersion.IBP_3_6_IV0
                )).getMessage()
        );
    }