public <T> RecordsSnapshotWriter<T> build(RecordSerde<T> serde) {
            if (!rawSnapshotWriter.isPresent()) {
                throw new IllegalStateException("Builder::build called without a RawSnapshotWriter");
            } else if (rawSnapshotWriter.get().sizeInBytes() != 0) {
                throw new IllegalStateException(
                    String.format("Initializing writer with a non-empty snapshot: %s", rawSnapshotWriter.get().snapshotId())
                );
            } else if (kraftVersion == KRaftVersion.KRAFT_VERSION_0 && voterSet.isPresent()) {
                throw new IllegalStateException(
                    String.format("Voter set (%s) not expected when the kraft.version is 0", voterSet.get())
                );
            }

            RecordsSnapshotWriter<T> writer = new RecordsSnapshotWriter<>(
                rawSnapshotWriter.get(),
                maxBatchSize,
                memoryPool,
                time,
                compression,
                serde
            );

            writer.accumulator.appendControlMessages((baseOffset, epoch, compression, buffer) -> {
                long now = time.milliseconds();
                try (MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
                        buffer,
                        RecordBatch.CURRENT_MAGIC_VALUE,
                        compression,
                        TimestampType.CREATE_TIME,
                        baseOffset,
                        now,
                        RecordBatch.NO_PRODUCER_ID,
                        RecordBatch.NO_PRODUCER_EPOCH,
                        RecordBatch.NO_SEQUENCE,
                        false, // isTransactional
                        true,  // isControlBatch
                        epoch,
                        buffer.capacity()
                    )
                ) {
                    builder.appendSnapshotHeaderMessage(
                        now,
                        new SnapshotHeaderRecord()
                            .setVersion(ControlRecordUtils.SNAPSHOT_HEADER_CURRENT_VERSION)
                            .setLastContainedLogTimestamp(lastContainedLogTimestamp)
                    );

                    if (kraftVersion.isReconfigSupported()) {
                        builder.appendKRaftVersionMessage(
                            now,
                            new KRaftVersionRecord()
                                .setVersion(ControlRecordUtils.KRAFT_VERSION_CURRENT_VERSION)
                                .setKRaftVersion(kraftVersion.featureLevel())
                        );

                        if (voterSet.isPresent()) {
                            builder.appendVotersMessage(
                                now,
                                voterSet.get().toVotersRecord(ControlRecordUtils.KRAFT_VOTERS_CURRENT_VERSION)
                            );
                        }
                    }

                    return builder.build();
                }
            });

            return writer;
        }
Adds a {@link SnapshotFooterRecord} to the snapshot

No more records should be appended to the snapshot after calling this method
raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java