ControllerResult<AlterPartitionResponseData> alterPartition(
        ControllerRequestContext context,
        AlterPartitionRequestData request
    ) {
        short requestVersion = context.requestHeader().requestApiVersion();
        clusterControl.checkBrokerEpoch(request.brokerId(), request.brokerEpoch());
        AlterPartitionResponseData response = new AlterPartitionResponseData();
        List<ApiMessageAndVersion> records = new ArrayList<>();
        for (AlterPartitionRequestData.TopicData topicData : request.topics()) {
            AlterPartitionResponseData.TopicData responseTopicData =
                new AlterPartitionResponseData.TopicData().
                    setTopicName(topicData.topicName()).
                    setTopicId(topicData.topicId());
            response.topics().add(responseTopicData);

            Uuid topicId = requestVersion > 1 ? topicData.topicId() : topicsByName.get(topicData.topicName());
            if (topicId == null || topicId.equals(Uuid.ZERO_UUID) || !topics.containsKey(topicId)) {
                Errors error = requestVersion > 1 ? UNKNOWN_TOPIC_ID : UNKNOWN_TOPIC_OR_PARTITION;
                for (AlterPartitionRequestData.PartitionData partitionData : topicData.partitions()) {
                    responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                        setPartitionIndex(partitionData.partitionIndex()).
                        setErrorCode(error.code()));
                }
                log.info("Rejecting AlterPartition request for unknown topic ID {} or name {}.",
                    topicData.topicId(), topicData.topicName());
                continue;
            }

            TopicControlInfo topic = topics.get(topicId);
            for (AlterPartitionRequestData.PartitionData partitionData : topicData.partitions()) {
                if (requestVersion < 3) {
                    partitionData.setNewIsrWithEpochs(
                        AlterPartitionRequest.newIsrToSimpleNewIsrWithBrokerEpochs(partitionData.newIsr())
                    );
                }

                int partitionId = partitionData.partitionIndex();
                PartitionRegistration partition = topic.parts.get(partitionId);

                Errors validationError = validateAlterPartitionData(
                    request.brokerId(),
                    topic,
                    partitionId,
                    partition,
                    context.requestHeader().requestApiVersion(),
                    partitionData);

                if (validationError != Errors.NONE) {
                    responseTopicData.partitions().add(
                        new AlterPartitionResponseData.PartitionData()
                            .setPartitionIndex(partitionId)
                            .setErrorCode(validationError.code())
                    );

                    continue;
                }

                PartitionChangeBuilder builder = new PartitionChangeBuilder(
                    partition,
                    topic.id,
                    partitionId,
                    new LeaderAcceptor(clusterControl, partition),
                    featureControl.metadataVersion(),
                    getTopicEffectiveMinIsr(topic.name)
                )
                    .setZkMigrationEnabled(clusterControl.zkRegistrationAllowed())
                    .setEligibleLeaderReplicasEnabled(isElrEnabled());
                if (configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name())) {
                    builder.setElection(PartitionChangeBuilder.Election.UNCLEAN);
                }

                // AutoMQ for Kafka inject start
                if (ElasticStreamSwitch.isEnabled()) {
                    List<Integer> newIsr = partitionData.newIsr();
                    if (newIsr.size() != 1) {
                        throw new InvalidReplicaAssignmentException("elastic stream not support isr != 1");
                    }
                    builder.setTargetNode(newIsr.get(0));
                } else {
                    builder.setTargetIsr(partitionData.newIsr());
                }

                Optional<ApiMessageAndVersion> record = builder
//                    .setTargetIsrWithBrokerStates(partitionData.newIsrWithEpochs())
                    .setTargetLeaderRecoveryState(LeaderRecoveryState.of(partitionData.leaderRecoveryState()))
                    .setDefaultDirProvider(clusterDescriber)
                    .build();
                // AutoMQ for Kafka inject end

                if (record.isPresent()) {
                    records.add(record.get());
                    PartitionChangeRecord change = (PartitionChangeRecord) record.get().message();
                    partition = partition.merge(change);
                    if (log.isDebugEnabled()) {
                        log.debug("Node {} has altered ISR for {}-{} to {}.",
                            request.brokerId(), topic.name, partitionId, change.isr());
                    }
                    if (change.leader() != request.brokerId() &&
                            change.leader() != NO_LEADER_CHANGE) {
                        // Normally, an AlterPartition request, which is made by the partition
                        // leader itself, is not allowed to modify the partition leader.
                        // However, if there is an ongoing partition reassignment and the
                        // ISR change completes it, then the leader may change as part of
                        // the changes made during reassignment cleanup.
                        //
                        // In this case, we report back NEW_LEADER_ELECTED to the leader
                        // which made the AlterPartition request. This lets it know that it must
                        // fetch new metadata before trying again. This return code is
                        // unusual because we both return an error and generate a new
                        // metadata record. We usually only do one or the other.
                        // FENCED_LEADER_EPOCH is used for request version below or equal to 1.
                        Errors error = requestVersion > 1 ? NEW_LEADER_ELECTED : FENCED_LEADER_EPOCH;
                        log.info("AlterPartition request from node {} for {}-{} completed " +
                            "the ongoing partition reassignment and triggered a " +
                            "leadership change. Returning {}.",
                            request.brokerId(), topic.name, partitionId, error);
                        responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                            setPartitionIndex(partitionId).
                            setErrorCode(error.code()));
                        continue;
                    } else if (isReassignmentInProgress(partition)) {
                        log.info("AlterPartition request from node {} for {}-{} completed " +
                            "the ongoing partition reassignment.", request.brokerId(),
                            topic.name, partitionId);
                    }
                }

                /* Setting the LeaderRecoveryState field is always safe because it will always be the
                 * same as the value set in the request. For version 0, that is always the default
                 * RECOVERED which is ignored when serializing to version 0. For any other version, the
                 * LeaderRecoveryState field is supported.
                 */
                responseTopicData.partitions().add(new AlterPartitionResponseData.PartitionData().
                    setPartitionIndex(partitionId).
                    setErrorCode(Errors.NONE.code()).
                    setLeaderId(partition.leader).
                    setIsr(Replicas.toList(partition.isr)).
                    setLeaderRecoveryState(partition.leaderRecoveryState.value()).
                    setLeaderEpoch(partition.leaderEpoch).
                    setPartitionEpoch(partition.partitionEpoch));
            }
        }

        return ControllerResult.of(records, response);
    }
Timer for partition reassignment elect timeout.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java