@Override
        public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
            appendRaftEvent("handleCommit[baseOffset=" + reader.baseOffset() + "]", () -> {
                try {
                    boolean isActive = isActiveController();
                    while (reader.hasNext()) {
                        Batch<ApiMessageAndVersion> batch = reader.next();
                        long offset = batch.lastOffset();
                        int epoch = batch.epoch();
                        List<ApiMessageAndVersion> messages = batch.records();

                        if (isActive) {
                            // If the controller is active, the records were already replayed,
                            // so we don't need to do it here.
                            log.debug("Completing purgatory items up to offset {} and epoch {}.", offset, epoch);

                            // Advance the committed and stable offsets then complete any pending purgatory
                            // items that were waiting for these offsets.
                            offsetControl.handleCommitBatch(batch);
                            deferredEventQueue.completeUpTo(offsetControl.lastStableOffset());
                            deferredUnstableEventQueue.completeUpTo(offsetControl.lastCommittedOffset());

                            // The active controller can delete up to the current committed offset.
                            snapshotRegistry.deleteSnapshotsUpTo(offsetControl.lastStableOffset());
                        } else {
                            // If the controller is a standby, replay the records that were
                            // created by the active controller.
                            if (log.isDebugEnabled()) {
                                log.debug("Replaying commits from the active node up to " +
                                    "offset {} and epoch {}.", offset, epoch);
                            }
                            int recordIndex = 0;
                            for (ApiMessageAndVersion message : messages) {
                                long recordOffset = batch.baseOffset() + recordIndex;
                                try {
                                    replay(message.message(), Optional.empty(), recordOffset);
                                } catch (Throwable e) {
                                    String failureMessage = String.format("Unable to apply %s " +
                                        "record at offset %d on standby controller, from the " +
                                        "batch with baseOffset %d",
                                        message.message().getClass().getSimpleName(),
                                        recordOffset, batch.baseOffset());
                                    throw fatalFaultHandler.handleFault(failureMessage, e);
                                }
                                recordIndex++;
                            }
                            offsetControl.handleCommitBatch(batch);
                        }
                    }
                } finally {
                    reader.close();
                }
            });
        }
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