ApiError electLeader(String topic, int partitionId, ElectionType electionType,
                         List<ApiMessageAndVersion> records) {
        // AutoMQ for Kafka inject start
        // cancel partition replica reassign leader elect timeout.
        Optional.ofNullable(partitionReElectTimeouts.remove(new TopicPartition(topic, partitionId))).ifPresent(Timeout::cancel);
        // AutoMQ for Kafka inject end

        Uuid topicId = topicsByName.get(topic);
        if (topicId == null) {
            return new ApiError(UNKNOWN_TOPIC_OR_PARTITION,
                "No such topic as " + topic);
        }
        TopicControlInfo topicInfo = topics.get(topicId);
        if (topicInfo == null) {
            return new ApiError(UNKNOWN_TOPIC_OR_PARTITION,
                "No such topic id as " + topicId);
        }
        PartitionRegistration partition = topicInfo.parts.get(partitionId);
        if (partition == null) {
            return new ApiError(UNKNOWN_TOPIC_OR_PARTITION,
                "No such partition as " + topic + "-" + partitionId);
        }
        if ((electionType == ElectionType.PREFERRED && partition.hasPreferredLeader())
            || (electionType == ElectionType.UNCLEAN && partition.hasLeader())) {
            return new ApiError(Errors.ELECTION_NOT_NEEDED);
        }

        PartitionChangeBuilder.Election election = PartitionChangeBuilder.Election.PREFERRED;
        if (electionType == ElectionType.UNCLEAN) {
            election = PartitionChangeBuilder.Election.UNCLEAN;
        }
        Optional<ApiMessageAndVersion> record = new PartitionChangeBuilder(
            partition,
            topicId,
            partitionId,
            new LeaderAcceptor(clusterControl, partition),
            featureControl.metadataVersion(),
            getTopicEffectiveMinIsr(topic)
        )
            .setElection(election)
            .setZkMigrationEnabled(clusterControl.zkRegistrationAllowed())
            .setEligibleLeaderReplicasEnabled(isElrEnabled())
            .setDefaultDirProvider(clusterDescriber)
            .build();
        if (!record.isPresent()) {
            if (electionType == ElectionType.PREFERRED) {
                return new ApiError(Errors.PREFERRED_LEADER_NOT_AVAILABLE);
            } else {
                return new ApiError(Errors.ELIGIBLE_LEADERS_NOT_AVAILABLE);
            }
        }
        records.add(record.get());
        return ApiError.NONE;
    }
Generates the appropriate records to handle a list of directories being reported offline.

If the reported directories include directories that were previously online, this includes
a BrokerRegistrationChangeRecord and any number of PartitionChangeRecord to update
leadership and ISR for partitions in those directories that were previously online.

@param brokerId    The broker id.
@param brokerEpoch The broker epoch.
@param offlineDirs The list of directories that are offline.
@param records     The record list to append to.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java