Repository: apache/kafka
File: metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java (Java)

Code before:
    private final List<Integer> quorumNodeIds;

    public static Optional<String> reasonNotSupported(
        short newVersion,
        String what,
        VersionRange range
    ) {
        if (!range.contains(newVersion)) {
            if (range.max() == (short) 0) {
                return Optional.of(what + " does not support this feature.");
            } else {
                return Optional.of(what + " only supports versions " + range);
            }
        }
        return Optional.empty();
    }

    public static Map<String, VersionRange> defaultFeatureMap(boolean enableUnstable) {
        Map<String, VersionRange> features = new HashMap<>(1);
        features.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
                MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(),
                enableUnstable ?
                    MetadataVersion.latestTesting().featureLevel() :
                    MetadataVersion.latestProduction().featureLevel()));
        for (Feature feature : Feature.PRODUCTION_FEATURES) {
            short maxVersion = enableUnstable ? feature.latestTesting() : feature.defaultLevel(MetadataVersion.LATEST_PRODUCTION);
            if (maxVersion > 0) {
                features.put(feature.featureName(), VersionRange.of(feature.minimumProduction(), maxVersion));
            }
        }
        return features;
    }

    public QuorumFeatures(
        int nodeId,
        Map<String, VersionRange> localSupportedFeatures,
        List<Integer> quorumNodeIds
    ) {
        this.nodeId = nodeId;
        this.localSupportedFeatures = Collections.unmodifiableMap(localSupportedFeatures);
        this.quorumNodeIds = Collections.unmodifiableList(quorumNodeIds);
    }

    public int nodeId() {
        return nodeId;
    }

    public Map<String, VersionRange> localSupportedFeatures() {
        return localSupportedFeatures;
    }


Review comment (bug):
@junrao @dongnuo123 I noticed we didn't change the defaults here on the previous PR. I have done so here. A test was failing since the production version for transaction version is now not the same as the default based on the latest production MV.

Code after:

    public static Optional<String> reasonNotSupported(
        short newVersion,
        String what,
        VersionRange range
    ) {
        if (!range.contains(newVersion)) {
            if (range.max() == (short) 0) {
                return Optional.of(what + " does not support this feature.");
            } else {
                return Optional.of(what + " only supports versions " + range);
            }
        }
        return Optional.empty();
    }

    public static Map<String, VersionRange> defaultSupportedFeatureMap(boolean enableUnstable) {
        Map<String, VersionRange> features = new HashMap<>(1);
        features.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
                MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(),
                enableUnstable ?
                    MetadataVersion.latestTesting().featureLevel() :
                    MetadataVersion.latestProduction().featureLevel()));
        for (Feature feature : Feature.PRODUCTION_FEATURES) {
            short maxVersion = enableUnstable ? feature.latestTesting() : feature.latestProduction();
            if (maxVersion > 0) {
                features.put(feature.featureName(), VersionRange.of(feature.minimumProduction(), maxVersion));
            }
        }
        return features;
    }

    public QuorumFeatures(
        int nodeId,
        Map<String, VersionRange> localSupportedFeatures,
        List<Integer> quorumNodeIds
    ) {
        this.nodeId = nodeId;
        this.localSupportedFeatures = Collections.unmodifiableMap(localSupportedFeatures);
        this.quorumNodeIds = Collections.unmodifiableList(quorumNodeIds);
    }

    public int nodeId() {
        return nodeId;
    }

    public Map<String, VersionRange> localSupportedFeatures() {
        return localSupportedFeatures;
    }

    public List<Integer> quorumNodeIds() {