@Override
        public void handleLoadSnapshot(SnapshotReader<ApiMessageAndVersion> reader) {
            appendRaftEvent(String.format("handleLoadSnapshot[snapshotId=%s]", reader.snapshotId()), () -> {
                try {
                    String snapshotName = Snapshots.filenameFromSnapshotId(reader.snapshotId());
                    if (isActiveController()) {
                        throw fatalFaultHandler.handleFault("Asked to load snapshot " + snapshotName +
                                ", but we are the active controller at epoch " + curClaimEpoch);
                    }
                    offsetControl.beginLoadSnapshot(reader.snapshotId());
                    while (reader.hasNext()) {
                        Batch<ApiMessageAndVersion> batch = reader.next();
                        long offset = batch.lastOffset();
                        List<ApiMessageAndVersion> messages = batch.records();

                        log.debug("Replaying snapshot {} batch with last offset of {}",
                                snapshotName, offset);

                        int i = 1;
                        for (ApiMessageAndVersion message : messages) {
                            try {
                                replay(message.message(), Optional.of(reader.snapshotId()),
                                        reader.lastContainedLogOffset());
                            } catch (Throwable e) {
                                String failureMessage = String.format("Unable to apply %s record " +
                                    "from snapshot %s on standby controller, which was %d of " +
                                    "%d record(s) in the batch with baseOffset %d.",
                                    message.message().getClass().getSimpleName(), reader.snapshotId(),
                                    i, messages.size(), batch.baseOffset());
                                throw fatalFaultHandler.handleFault(failureMessage, e);
                            }
                            i++;
                        }
                    }
                    offsetControl.endLoadSnapshot(reader.lastContainedLogTimestamp());
                } catch (FaultHandlerException e) {
                    throw e;
                } catch (Throwable e) {
                    throw fatalFaultHandler.handleFault("Error while loading snapshot " +
                            reader.snapshotId(), e);
                } 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