content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.server;
  
  import org.apache.kafka.common.feature.Features;
  import org.apache.kafka.common.feature.SupportedVersionRange;
  import org.apache.kafka.metadata.VersionRange;
  import org.apache.kafka.server.common.KRaftVersion;
  import org.apache.kafka.server.common.MetadataVersion;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.HashMap;
  import java.util.Map;
  import java.util.stream.Collectors;
  
  import static org.apache.kafka.server.common.Feature.PRODUCTION_FEATURES;
  
  /**
   * A class that encapsulates the latest features supported by the Broker and also provides APIs to
   * check for incompatibilities between the features supported by the Broker and finalized features.
   * This class is immutable in production. It provides few APIs to mutate state only for the purpose
   * of testing.
   */
  public class BrokerFeatures {
  
      private final Features<SupportedVersionRange> supportedFeatures;
      private static final Logger log = LoggerFactory.getLogger(BrokerFeatures.class);
  
      private BrokerFeatures(Features<SupportedVersionRange> supportedFeatures) {
          this.supportedFeatures = supportedFeatures;
      }
  
      public static BrokerFeatures createDefault(boolean unstableFeatureVersionsEnabled) {
          return new BrokerFeatures(defaultSupportedFeatures(unstableFeatureVersionsEnabled));
      }
      
      // only 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));
      }
  
      public static Map<String, VersionRange> createDefaultFeatureMap(BrokerFeatures features) {
          return features.supportedFeatures.features()
                  .entrySet()
                  .stream()
                  .collect(Collectors.toMap(Map.Entry::getKey, e -> VersionRange.of(e.getValue().min(), e.getValue().max())));
      }
  
      public static Features<SupportedVersionRange> defaultSupportedFeatures(boolean unstableFeatureVersionsEnabled) {
          Map<String, SupportedVersionRange> features = new HashMap<>();
          features.put(MetadataVersion.FEATURE_NAME,
                  new SupportedVersionRange(
                          MetadataVersion.MINIMUM_VERSION.featureLevel(),
                          unstableFeatureVersionsEnabled ? MetadataVersion.latestTesting().featureLevel()
                                  : MetadataVersion.latestProduction().featureLevel()));
          PRODUCTION_FEATURES.forEach(feature -> {
              int maxVersion = unstableFeatureVersionsEnabled ? feature.latestTesting() : feature.latestProduction();
              if (maxVersion > 0) {
                  features.put(feature.featureName(), new SupportedVersionRange(feature.minimumProduction(), (short) maxVersion));
              }
          });
          return Features.supportedFeatures(features);
      }
  
      public static BrokerFeatures createEmpty() {
          return new BrokerFeatures(Features.emptySupportedFeatures());
      }
  
      /**
       * Returns true if any of the provided finalized features are incompatible with the provided
       * supported features.
       *
       * @param supportedFeatures The supported features to be compared
       * @param finalizedFeatures The finalized features to be compared
       * @return - True if there are any feature incompatibilities found.
       * - False otherwise.
       */
      public static boolean hasIncompatibleFeatures(Features<SupportedVersionRange> supportedFeatures,
                                                    Map<String, Short> finalizedFeatures) {
          return !incompatibleFeatures(supportedFeatures, finalizedFeatures, false).isEmpty();
      }
  
      /**
       * Returns the default finalized features that a new Kafka cluster with IBP config >= IBP_2_7_IV0
       * needs to be bootstrapped with.
       */
      public Map<String, Short> defaultFinalizedFeatures() {
          return supportedFeatures.features().entrySet()
                  .stream()
                  .collect(Collectors.toMap(Map.Entry::getKey,
                          e -> e.getKey().equals(KRaftVersion.FEATURE_NAME) ? 0 : e.getValue().max()));
      }
  
      /**
       * Returns the set of feature names found to be incompatible.
       * A feature incompatibility is a version mismatch between the latest feature supported by the
       * Broker, and a provided finalized feature. This can happen because a provided finalized
       * feature:
       * 1) Does not exist in the Broker (i.e. it is unknown to the Broker).
       * [OR]
       * 2) Exists but the FinalizedVersionRange does not match with the SupportedVersionRange
       * of the supported feature.
       *
       * @param finalized The finalized features against which incompatibilities need to be checked for.
       * @return The subset of input features which are incompatible. If the returned object
       * is empty, it means there were no feature incompatibilities found.
       */
      public Map<String, Short> incompatibleFeatures(Map<String, Short> finalized) {
          return BrokerFeatures.incompatibleFeatures(supportedFeatures, finalized, true);
      }
  
      public Features<SupportedVersionRange> supportedFeatures() {
          return supportedFeatures;
      }
  
      private static Map<String, Short> incompatibleFeatures(Features<SupportedVersionRange> supportedFeatures,
                                                             Map<String, Short> finalizedFeatures,
                                                             boolean logIncompatibilities) {
          Map<String, Short> incompatibleFeaturesInfo = new HashMap<>();
          finalizedFeatures.forEach((feature, versionLevels) -> {
              SupportedVersionRange supportedVersions = supportedFeatures.get(feature);
              if (supportedVersions == null) {
                  incompatibleFeaturesInfo.put(feature, versionLevels);
                  if (logIncompatibilities) {
                      log.warn("Feature incompatibilities seen: {feature={}, reason='Unknown feature'}", feature);
                  }
              } else if (supportedVersions.isIncompatibleWith(versionLevels)) {
                  incompatibleFeaturesInfo.put(feature, versionLevels);
                  if (logIncompatibilities) {
                      log.warn("Feature incompatibilities seen: {feature={}, reason='{} is incompatible with {}'}",
                              feature, versionLevels, supportedVersions);
                  }
              }
          });
          return incompatibleFeaturesInfo;
      }
  }
  