public void appendLeaderChangeMessageAndBootstrapRecords(long currentTimeMs) {
        List<Voter> voters = convertToVoters(voterStates.keySet());
        List<Voter> grantingVoters = convertToVoters(this.grantingVoters());

        LeaderChangeMessage leaderChangeMessage = new LeaderChangeMessage()
            .setVersion(ControlRecordUtils.LEADER_CHANGE_CURRENT_VERSION)
            .setLeaderId(this.election().leaderId())
            .setVoters(voters)
            .setGrantingVoters(grantingVoters);

        accumulator.appendControlMessages((baseOffset, epoch, compression, buffer) -> {
            try (MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
                    buffer,
                    RecordBatch.CURRENT_MAGIC_VALUE,
                    compression,
                    TimestampType.CREATE_TIME,
                    baseOffset,
                    currentTimeMs,
                    RecordBatch.NO_PRODUCER_ID,
                    RecordBatch.NO_PRODUCER_EPOCH,
                    RecordBatch.NO_SEQUENCE,
                    false, // isTransactional
                    true,  // isControlBatch
                    epoch,
                    buffer.capacity()
                )
            ) {
                builder.appendLeaderChangeMessage(currentTimeMs, leaderChangeMessage);

                offsetOfVotersAtEpochStart.ifPresent(offset -> {
                    if (offset == -1) {
                        // Latest voter set came from the bootstrap checkpoint (0-0.checkpoint)
                        // rewrite the voter set to the log so that it is replicated to the replicas.
                        if (!kraftVersionAtEpochStart.isReconfigSupported()) {
                            throw new IllegalStateException(
                                String.format(
                                    "The bootstrap checkpoint contains a set of voters %s at %s " +
                                    "and the KRaft version is %s",
                                    voterSetAtEpochStart,
                                    offset,
                                    kraftVersionAtEpochStart
                                )
                            );
                        } else {
                            builder.appendKRaftVersionMessage(
                                currentTimeMs,
                                new KRaftVersionRecord()
                                    .setVersion(ControlRecordUtils.KRAFT_VERSION_CURRENT_VERSION)
                                    .setKRaftVersion(kraftVersionAtEpochStart.featureLevel())
                            );
                            builder.appendVotersMessage(
                                currentTimeMs,
                                voterSetAtEpochStart.toVotersRecord(
                                    ControlRecordUtils.KRAFT_VOTERS_CURRENT_VERSION
                                )
                            );
                        }
                    }
                });

                return builder.build();
            }
        });
        accumulator.forceDrain();
    }
Reset the checkQuorumTimer if we've received fetch/fetchSnapshot request from the majority of the voter

@param replicaKey the replica key of the voter
@param currentTimeMs the current timestamp in millisecond
raft/src/main/java/org/apache/kafka/raft/LeaderState.java