kowshik commented on a change in pull request #9001:
URL: https://github.com/apache/kafka/pull/9001#discussion_r463935718
##########
File path: core/src/main/scala/kafka/controller/KafkaController.scala
##########
@@ -1647,6 +1844,185 @@ class KafkaController(val config: KafkaConfig,
}
}
+ /**
+ * Returns the new FinalizedVersionRange for the feature, if there are no
feature
+ * incompatibilities seen with all known brokers for the provided feature
update.
+ * Otherwise returns a suitable error.
+ *
+ * @param update the feature update to be processed (this can not be meant
to delete the feature)
+ *
+ * @return the new FinalizedVersionRange or error, as described
above.
+ */
+ private def newFinalizedVersionRangeOrIncompatibilityError(update:
UpdateFeaturesRequestData.FeatureUpdateKey): Either[FinalizedVersionRange,
ApiError] = {
+ if (UpdateFeaturesRequest.isDeleteRequest(update)) {
+ throw new IllegalArgumentException(s"Provided feature update can not be
meant to delete the feature: $update")
+ }
+ // NOTE: Below we set the finalized min version level to be the default
minimum version
+ // level. If the finalized feature already exists, then, this can cause
deprecation of all
+ // version levels in the closed range:
+ // [existingVersionRange.min(), defaultMinVersionLevel - 1].
+ val defaultMinVersionLevel =
brokerFeatures.defaultMinVersionLevel(update.feature)
+ val newVersionRange = new FinalizedVersionRange(defaultMinVersionLevel,
update.maxVersionLevel)
+ val numIncompatibleBrokers =
controllerContext.liveOrShuttingDownBrokers.count(broker => {
+ val singleFinalizedFeature =
+ Features.finalizedFeatures(Utils.mkMap(Utils.mkEntry(update.feature,
newVersionRange)))
+ BrokerFeatures.hasIncompatibleFeatures(broker.features,
singleFinalizedFeature)
+ })
+ if (numIncompatibleBrokers == 0) {
+ Left(newVersionRange)
+ } else {
+ Right(
+ new ApiError(Errors.INVALID_REQUEST,
+ s"Could not apply finalized feature update because
$numIncompatibleBrokers" +
+ " brokers were found to have incompatible features."))
+ }
+ }
+
+ /**
+ * Validate and process a finalized feature update on an existing
FinalizedVersionRange for the
+ * feature.
+ *
+ * If the processing is successful, then, the return value contains:
+ * 1. the new FinalizedVersionRange for the feature, if the feature update
was not meant to delete the feature.
+ * 2. Option.empty, if the feature update was meant to delete the feature.
+ *
+ * If the processing failed, then returned value contains a suitable
ApiError.
+ *
+ * @param update the feature update to be processed.
+ * @param existingVersionRange the existing FinalizedVersionRange which
can be empty when no
+ * FinalizedVersionRange exists for the
associated feature
+ *
+ * @return the new FinalizedVersionRange or error, as
described above.
+ */
+ private def processFeatureUpdate(update:
UpdateFeaturesRequestData.FeatureUpdateKey,
+ existingVersionRange:
Option[FinalizedVersionRange]): Either[Option[FinalizedVersionRange], ApiError]
= {
+ def newVersionRangeOrError(update:
UpdateFeaturesRequestData.FeatureUpdateKey):
Either[Option[FinalizedVersionRange], ApiError] = {
+ newFinalizedVersionRangeOrIncompatibilityError(update)
+ .fold(versionRange => Left(Some(versionRange)), error => Right(error))
+ }
+
+ if (update.feature.isEmpty) {
+ // Check that the feature name is not empty.
+ Right(new ApiError(Errors.INVALID_REQUEST, "Feature name can not be
empty."))
+ } else {
+ // We handle deletion requests separately from non-deletion requests.
+ if (UpdateFeaturesRequest.isDeleteRequest(update)) {
+ if (existingVersionRange.isEmpty) {
+ // Disallow deletion of a non-existing finalized feature.
+ Right(new ApiError(Errors.INVALID_REQUEST,
+ s"Can not delete non-existing finalized feature:
'${update.feature}'"))
+ } else {
+ Left(Option.empty)
+ }
+ } else if (update.maxVersionLevel() < 1) {
Review comment:
It is for returning the special error message that we handle it here
explicitly: `...less than 1 for feature...`.
A value < 1 is indicative of a deletion request (not purely a downgrade
request).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
