public ControllerResult<BrokerRegistrationReply> registerBroker(
        BrokerRegistrationRequestData request,
        long newBrokerEpoch,
        FinalizedControllerFeatures finalizedFeatures
    ) {
        if (heartbeatManager == null) {
            throw new RuntimeException("ClusterControlManager is not active.");
        }
        if (!clusterId.equals(request.clusterId())) {
            throw new InconsistentClusterIdException("Expected cluster ID " + clusterId +
                ", but got cluster ID " + request.clusterId());
        }
        int brokerId = request.brokerId();
        List<ApiMessageAndVersion> records = new ArrayList<>();
        BrokerRegistration existing = brokerRegistrations.get(brokerId);
        Uuid prevIncarnationId = null;
        if (existing != null) {
            prevIncarnationId = existing.incarnationId();
            if (heartbeatManager.hasValidSession(brokerId)) {
                if (!request.incarnationId().equals(prevIncarnationId)) {
                    throw new DuplicateBrokerRegistrationException("Another broker is " +
                        "registered with that broker id.");
                }
            }
        }

        if (request.isMigratingZkBroker() && !zkRegistrationAllowed()) {
            throw new BrokerIdNotRegisteredException("Controller does not support registering ZK brokers.");
        }

        if (!request.isMigratingZkBroker() && featureControl.inPreMigrationMode()) {
            throw new BrokerIdNotRegisteredException("Controller is in pre-migration mode and cannot register KRaft " +
                "brokers until the metadata migration is complete.");
        }

        if (featureControl.metadataVersion().isDirectoryAssignmentSupported()) {
            if (request.logDirs().isEmpty()) {
                throw new InvalidRegistrationException("No directories specified in request");
            }
            if (request.logDirs().stream().anyMatch(DirectoryId::reserved)) {
                throw new InvalidRegistrationException("Reserved directory ID in request");
            }
            Set<Uuid> set = new HashSet<>(request.logDirs());
            if (set.size() != request.logDirs().size()) {
                throw new InvalidRegistrationException("Duplicate directory ID in request");
            }
            for (Uuid directory : request.logDirs()) {
                Integer dirBrokerId = directoryToBroker.get(directory);
                if (dirBrokerId != null && dirBrokerId != brokerId) {
                    throw new InvalidRegistrationException("Broker " + dirBrokerId +
                            " is already registered with directory " + directory);
                }
            }
        }

        ListenerInfo listenerInfo = ListenerInfo.fromBrokerRegistrationRequest(request.listeners());
        RegisterBrokerRecord record = new RegisterBrokerRecord().
            setBrokerId(brokerId).
            setIsMigratingZkBroker(request.isMigratingZkBroker()).
            setIncarnationId(request.incarnationId()).
            setRack(request.rack()).
            setEndPoints(listenerInfo.toBrokerRegistrationRecord());

        for (BrokerRegistrationRequestData.Feature feature : request.features()) {
            record.features().add(processRegistrationFeature(brokerId, finalizedFeatures, feature));
        }
        if (request.features().find(MetadataVersion.FEATURE_NAME) == null) {
            // Brokers that don't send a supported metadata.version range are assumed to only
            // support the original metadata.version.
            record.features().add(processRegistrationFeature(brokerId, finalizedFeatures,
                    new BrokerRegistrationRequestData.Feature().
                            setName(MetadataVersion.FEATURE_NAME).
                            setMinSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel()).
                            setMaxSupportedVersion(MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel())));
        }
        if (featureControl.metadataVersion().isDirectoryAssignmentSupported()) {
            record.setLogDirs(request.logDirs());
        }

        if (!request.incarnationId().equals(prevIncarnationId)) {
            int prevNumRecords = records.size();
            brokerUncleanShutdownHandler.addRecordsForShutdown(request.brokerId(), records);
            int numRecordsAdded = records.size() - prevNumRecords;
            if (existing == null) {
                log.info("No previous registration found for broker {}. New incarnation ID is " +
                        "{}.  Generated {} record(s) to clean up previous incarnations. New broker " +
                        "epoch is {}.", brokerId, request.incarnationId(), numRecordsAdded, newBrokerEpoch);
            } else {
                log.info("Registering a new incarnation of broker {}. Previous incarnation ID " +
                        "was {}; new incarnation ID is {}. Generated {} record(s) to clean up " +
                        "previous incarnations. Broker epoch will become {}.", brokerId,
                        existing.incarnationId(), request.incarnationId(), numRecordsAdded,
                        newBrokerEpoch);
            }
            record.setBrokerEpoch(newBrokerEpoch);
        } else {
            log.info("Amending registration of broker {}, incarnation ID {}. Broker epoch remains {}.",
                    request.brokerId(), request.incarnationId(), existing.epoch());
            record.setFenced(existing.fenced());
            record.setInControlledShutdown(existing.inControlledShutdown());
            record.setBrokerEpoch(existing.epoch());
        }
        records.add(new ApiMessageAndVersion(record, featureControl.metadataVersion().
            registerBrokerRecordVersion()));

        if (!request.incarnationId().equals(prevIncarnationId)) {
            // Remove any existing session for the old broker incarnation.
            heartbeatManager.remove(brokerId);
        }
        heartbeatManager.register(brokerId, record.fenced());

        // AutoMQ for Kafka inject start
        records.addAll(registerBrokerRecords(brokerId));
        // AutoMQ for Kafka inject end

        return ControllerResult.atomicOf(records, new BrokerRegistrationReply(record.brokerEpoch()));
    }
Process an incoming broker registration request.
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java