public void initialize(
        Map<Integer, InetSocketAddress> voterAddresses,
        QuorumStateStore quorumStateStore,
        Metrics metrics
    ) {
        Optional<VoterSet> staticVoters = voterAddresses.isEmpty() ?
            Optional.empty() :
            Optional.of(VoterSet.fromInetSocketAddresses(channel.listenerName(), voterAddresses));

        partitionState = new KRaftControlRecordStateMachine(
            staticVoters,
            log,
            serde,
            // AutoMQ for Kafka inject start
            // BufferSupplier.create(),
            new BufferSupplier.GrowableMultiBufferSupplier(),
            // AutoMQ for Kafka inject end
            MAX_BATCH_SIZE_BYTES,
            logContext
        );
        // Read the entire log
        logger.info("Reading KRaft snapshot and log as part of the initialization");
        partitionState.updateState();

        if (requestManager == null) {
            // The request manager wasn't created using the bootstrap servers
            // create it using the voters static configuration
            List<Node> bootstrapNodes = voterAddresses
                .entrySet()
                .stream()
                .map(entry ->
                    new Node(
                        entry.getKey(),
                        entry.getValue().getHostString(),
                        entry.getValue().getPort()
                    )
                )
                .collect(Collectors.toList());

            logger.info("Starting request manager with static voters: {}", bootstrapNodes);

            requestManager = new RequestManager(
                bootstrapNodes,
                quorumConfig.retryBackoffMs(),
                quorumConfig.requestTimeoutMs(),
                random
            );
        }

        quorum = new QuorumState(
            nodeId,
            nodeDirectoryId,
            partitionState,
            localListeners,
            quorumConfig.electionTimeoutMs(),
            quorumConfig.fetchTimeoutMs(),
            quorumStateStore,
            time,
            logContext,
            random
        );

        kafkaRaftMetrics = new KafkaRaftMetrics(metrics, "raft", quorum);
        // All Raft voters are statically configured and known at startup
        // so there are no unknown voter connections. Report this metric as 0.
        kafkaRaftMetrics.updateNumUnknownVoterConnections(0);

        quorum.initialize(new OffsetAndEpoch(log.endOffset().offset(), log.lastFetchedEpoch()));

        long currentTimeMs = time.milliseconds();
        if (quorum.isLeader()) {
            throw new IllegalStateException("Voter cannot initialize as a Leader");
        } else if (quorum.isCandidate()) {
            onBecomeCandidate(currentTimeMs);
        } else if (quorum.isFollower()) {
            onBecomeFollower(currentTimeMs);
        }

        // When there is only a single voter, become candidate immediately
        if (quorum.isOnlyVoter() && !quorum.isCandidate()) {
            transitionToCandidate(currentTimeMs);
        }

        // Specialized add voter handler
        this.addVoterHandler = new AddVoterHandler(
            partitionState,
            new DefaultRequestSender(
                requestManager,
                channel,
                messageQueue,
                logContext
            ),
            time,
            logContext
        );

        // Specialized remove voter handler
        this.removeVoterHandler = new RemoveVoterHandler(
            nodeId,
            nodeDirectoryId,
            partitionState,
            time,
            quorumConfig.requestTimeoutMs(),
            logContext
        );
    }
Create a new instance.

Note that if the node ID is empty, then the client will behave as a
non-participating observer.
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java