BrokerFeature processRegistrationFeature(
        int brokerId,
        FinalizedControllerFeatures finalizedFeatures,
        BrokerRegistrationRequestData.Feature feature
    ) {
        int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
        short finalized = finalizedFeatures.versionOrDefault(feature.name(), (short) defaultVersion);
        if (!VersionRange.of(feature.minSupportedVersion(), feature.maxSupportedVersion()).contains(finalized)) {
            throw new UnsupportedVersionException("Unable to register because the broker " +
                "does not support version " + finalized + " of " + feature.name() +
                    ". It wants a version between " + feature.minSupportedVersion() + " and " +
                    feature.maxSupportedVersion() + ", inclusive.");
        }
        // A feature is not found in the finalizedFeature map if it is unknown to the controller or set to 0 (feature not enabled).
        // Only log if the feature name is not known by the controller.
        if (!Features.PRODUCTION_FEATURE_NAMES.contains(feature.name()))
            log.warn("Broker {} registered with feature {} that is unknown to the controller",
                    brokerId, feature.name());
        return new BrokerFeature().
                setName(feature.name()).
                setMinSupportedVersion(feature.minSupportedVersion()).
                setMaxSupportedVersion(feature.maxSupportedVersion());
    }
Process an incoming broker registration request.
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java