@Override
    public Optional<SnapshotWriter<T>> createSnapshot(
        OffsetAndEpoch snapshotId,
        long lastContainedLogTimestamp
    ) {
        if (!isInitialized()) {
            throw new IllegalStateException("Cannot create snapshot before the replica has been initialized");
        }

        return log.createNewSnapshot(snapshotId).map(writer -> {
            long lastContainedLogOffset = snapshotId.offset() - 1;

            RawSnapshotWriter wrappedWriter = new NotifyingRawSnapshotWriter(writer, offsetAndEpoch -> {
                // Trim the state in the internal listener up to the new snapshot
                partitionState.truncateOldEntries(offsetAndEpoch.offset());
            });

            return new RecordsSnapshotWriter.Builder()
                .setLastContainedLogTimestamp(lastContainedLogTimestamp)
                .setTime(time)
                .setMaxBatchSize(MAX_BATCH_SIZE_BYTES)
                .setMemoryPool(memoryPool)
                .setRawSnapshotWriter(wrappedWriter)
                .setKraftVersion(partitionState.kraftVersionAtOffset(lastContainedLogOffset))
                .setVoterSet(partitionState.voterSetAtOffset(lastContainedLogOffset))
                .build(serde);
        });
    }
Poll for new events. This allows the client to handle inbound
requests and send any needed outbound requests.
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java