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
