KAFKA-17564: Move BrokerFeatures to server module

[m1a2st]

Jira: https://issues.apache.org/jira/browse/KAFKA-17564

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[frankvicky]

Hi @m1a2st
I have few comments, PTAL

[m1a2st]

@frankvicky thanks for your comments, addressed them.

[chia7712]

It is weird to make supportedFeatures mutable for testing-only. Could you please add a new helper for testing?

    public static BrokerFeatures createDefault(boolean unstableFeatureVersionsEnabled, Features<SupportedVersionRange> newFeatures) {
        Map<String, SupportedVersionRange> combined = new HashMap<>(defaultSupportedFeatures(unstableFeatureVersionsEnabled).features());
        combined.putAll(newFeatures.features());
        return new BrokerFeatures(Features.supportedFeatures(combined));
    }

[m1a2st]

but the volatile can't use with final keyword, in this case I think we should reserve the volatile. If we can't make supportedFeatures immutable, Do we still need to use this helper?

[chia7712]

If we can't make supportedFeatures immutable, Do we still need to use this helper?

Yes, I proposed making supportedFeatures immutable. Are there any production use cases where supportedFeatures would need to be changed?

[m1a2st]

Yes, you are right, I will make it to final.

[chia7712]

@m1a2st thanks for your patch!

[chia7712]

        return supportedFeatures.features().entrySet()
                .stream()
                .collect(Collectors.toMap(Map.Entry::getKey,
                        e -> e.getKey().equals(KRaftVersion.FEATURE_NAME) ? 0 : e.getValue().max()));

[chia7712]

maybe transferToIncompatibleFeaturesInfo should return the feature with error to avoid extra filter?

[chia7712]

        return features.supportedFeatures.features()
                .entrySet()
                .stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> VersionRange.of(e.getValue().min(), e.getValue().max())));

[chia7712]

Could you please inline transferToIncompatibleFeaturesInfoToErrorCode? We can loop through the features and log any incompatible ones. This would simplify the code and provide a slight speed improvement.

[m1a2st]

If use StringBuilder to resolve this errorReason, we should deal with last two ", " so the code will be like

    private static Map<String, Short> incompatibleFeatures(Features<SupportedVersionRange> supportedFeatures,
                                                           Map<String, Short> finalizedFeatures,
                                                           boolean logIncompatibilities) {
        IncompatibleFeaturesInfosToErrorReasons incompatibleFeaturesInfoToErrorReason = 
                transferToIncompatibleFeaturesInfoToErrorCode(supportedFeatures, finalizedFeatures);
        if (logIncompatibilities && !incompatibleFeaturesInfoToErrorReason.errorReason.isEmpty()) {
            log.warn("Feature incompatibilities seen: {}", incompatibleFeaturesInfoToErrorReason.errorReason);
        }
        return incompatibleFeaturesInfoToErrorReason.incompatibleFeaturesInfo;
    }

    private static IncompatibleFeaturesInfosToErrorReasons transferToIncompatibleFeaturesInfoToErrorCode(
            Features<SupportedVersionRange> supportedFeatures,
            Map<String, Short> finalizedFeatures
    ) {
        Map<String, Short> incompatibleFeaturesInfo = new HashMap<>();
        StringBuilder errorReasons = new StringBuilder();
        finalizedFeatures.forEach((feature, versionLevels) -> {
            SupportedVersionRange supportedVersions = supportedFeatures.get(feature);
            if (supportedVersions == null) {
                incompatibleFeaturesInfo.put(feature, versionLevels);
                errorReasons.append(format("{feature=%s, reason='Unsupported feature'}, ", feature));
            } else if (supportedVersions.isIncompatibleWith(versionLevels)) {
                incompatibleFeaturesInfo.put(feature, versionLevels);
                errorReasons.append(format("{feature=%s, reason='%s is incompatible with %s'}, ", feature, versionLevels, supportedVersions));
            }
        });
        return new IncompatibleFeaturesInfosToErrorReasons(
                incompatibleFeaturesInfo,
                errorReasons.length() == 0 ? "" :
                errorReasons.delete(errorReasons.length() - 2, errorReasons.length()).toString()
        );
    }
    
    private static class IncompatibleFeaturesInfosToErrorReasons {
        final Map<String, Short> incompatibleFeaturesInfos;
        final String errorReason;
        
        IncompatibleFeaturesInfosToErrorReasons(Map<String, Short> incompatibleFeaturesInfos, String errorReason) {
            this.incompatibleFeaturesInfos = incompatibleFeaturesInfos;
            this.errorReason = errorReason;
        }
    }

I prefer to use the List to resolve this, It is more readable

    private static Map<String, Short> incompatibleFeatures(Features<SupportedVersionRange> supportedFeatures,
                                                           Map<String, Short> finalizedFeatures,
                                                           boolean logIncompatibilities) {
        IncompatibleFeaturesInfosToErrorReasons incompatibleFeaturesInfoToErrorReason = 
                transferToIncompatibleFeaturesInfoToErrorCode(supportedFeatures, finalizedFeatures);
        if (logIncompatibilities && !incompatibleFeaturesInfoToErrorReason.errorReason.isEmpty()) {
            log.warn("Feature incompatibilities seen: {}", incompatibleFeaturesInfoToErrorReason.errorReason);
        }
        return incompatibleFeaturesInfoToErrorReason.incompatibleFeaturesInfos;
    }

    private static IncompatibleFeaturesInfosToErrorReasons transferToIncompatibleFeaturesInfoToErrorCode(
            Features<SupportedVersionRange> supportedFeatures,
            Map<String, Short> finalizedFeatures
    ) {
        Map<String, Short> incompatibleFeaturesInfo = new HashMap<>();
        List<String> errorReasons = new ArrayList<>();
        finalizedFeatures.forEach((feature, versionLevels) -> {
            SupportedVersionRange supportedVersions = supportedFeatures.get(feature);
            if (supportedVersions == null) {
                incompatibleFeaturesInfo.put(feature, versionLevels);
                errorReasons.add(format("{feature=%s, reason='Unknown feature'}", feature));
            } else if (supportedVersions.isIncompatibleWith(versionLevels)) {
                incompatibleFeaturesInfo.put(feature, versionLevels);
                errorReasons.add(format("{feature=%s, reason='%s is incompatible with %s'}", feature, versionLevels, supportedVersions));
            }
        });
        return new IncompatibleFeaturesInfosToErrorReasons(
                incompatibleFeaturesInfo,
                String.join(", ", errorReasons)
        );
    }
    
    private static class IncompatibleFeaturesInfosToErrorReasons {
        final Map<String, Short> incompatibleFeaturesInfos;
        final String errorReason;
        
        IncompatibleFeaturesInfosToErrorReasons(Map<String, Short> incompatibleFeaturesInfos, String errorReason) {
            this.incompatibleFeaturesInfos = incompatibleFeaturesInfos;
            this.errorReason = errorReason;
        }
    }

WDYT?