ahuang98 commented on code in PR #12050: URL: https://github.com/apache/kafka/pull/12050#discussion_r869745683
########## core/src/test/scala/unit/kafka/server/BrokerFeaturesTest.scala: ########## @@ -25,10 +25,6 @@ import scala.jdk.CollectionConverters._ class BrokerFeaturesTest { - @Test Review Comment: nit: Instead of removing, should we replace with `assertTrue(BrokerFeatures.createEmpty().supportedFeatures.empty)` ########## core/src/main/scala/kafka/raft/RaftManager.scala: ########## @@ -108,7 +108,8 @@ class KafkaRaftManager[T]( time: Time, metrics: Metrics, threadNamePrefixOpt: Option[String], - val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]] + val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]], + apiVersions: ApiVersions Review Comment: Not sure if it makes a difference, but it seems that all calls to create a new `KafkaRaftManager` use a newly created `ApiVersions`, should we just drop this from the list of constructor arguments and instantiate within the class instead? ``` private val apiVersions = new ApiVersions() ``` ########## core/src/main/scala/kafka/tools/StorageTool.scala: ########## @@ -189,21 +207,24 @@ object StorageTool extends Logging { def buildMetadataProperties( clusterIdStr: String, - config: KafkaConfig + config: KafkaConfig, Review Comment: nit: extra comma? ########## core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala: ########## @@ -127,19 +128,26 @@ class BrokerMetadataPublisher(conf: KafkaConfig, // Publish the new metadata image to the metadata cache. metadataCache.setImage(newImage) + val metadataVersion: Option[Short] = Option(newImage.features().metadataVersion()) Review Comment: Is it expected that we might end up with an empty Option here? Wondering if we need any error logging for this ########## core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala: ########## @@ -127,19 +128,26 @@ class BrokerMetadataPublisher(conf: KafkaConfig, // Publish the new metadata image to the metadata cache. metadataCache.setImage(newImage) + val metadataVersion: Option[Short] = Option(newImage.features().metadataVersion()) + .filterNot(mv => mv.equals(MetadataVersion.UNINITIALIZED)) + .map(_.featureLevel()) + if (_firstPublish) { - info(s"Publishing initial metadata at offset $highestOffsetAndEpoch.") + info(s"Publishing initial metadata at offset $highestOffsetAndEpoch with metadata.version $metadataVersion.") Review Comment: nit: remove extra space before `with` -- 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. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org