@Override
        public CompletableFuture<?> beginMigration() {
            if (featureControl.metadataVersion().isMetadataTransactionSupported()) {
                log.info("Starting migration of ZooKeeper metadata to KRaft.");
                ControllerWriteEvent<Void> batchEvent = new ControllerWriteEvent<>(
                    "Begin ZK Migration Transaction",
                    new MigrationWriteOperation(Collections.singletonList(
                        new ApiMessageAndVersion(
                            new BeginTransactionRecord().setName("ZK Migration"), (short) 0))
                    ), eventFlags);
                queue.append(batchEvent);
                return batchEvent.future;
            } else {
                log.warn("Starting ZK Migration without metadata transactions enabled. This is not safe since " +
                    "a controller failover or processing error may lead to partially migrated metadata.");
                return CompletableFuture.completedFuture(null);
            }
        }
Append records to the Raft log. They will be written out asynchronously.

@param log                   The log4j logger.
@param result                The controller result we are writing out.
@param maxRecordsPerBatch    The maximum number of records to allow in a batch.
@param appender              The callback to invoke for each batch. The arguments are last
                             write offset, record list, and the return result is the new
                             last write offset.
@return                      The final offset that was returned from the Raft layer.
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java