hachikuji commented on code in PR #12250: URL: https://github.com/apache/kafka/pull/12250#discussion_r892713477
########## core/src/main/scala/kafka/server/ReplicaFetcherManager.scala: ########## @@ -21,13 +21,15 @@ import kafka.cluster.BrokerEndPoint import org.apache.kafka.clients.FetchSessionHandler import org.apache.kafka.common.metrics.Metrics import org.apache.kafka.common.utils.{LogContext, Time} +import org.apache.kafka.server.common.MetadataVersion class ReplicaFetcherManager(brokerConfig: KafkaConfig, protected val replicaManager: ReplicaManager, metrics: Metrics, time: Time, threadNamePrefix: Option[String] = None, - quotaManager: ReplicationQuotaManager) + quotaManager: ReplicationQuotaManager, + val metadataVersionSupplier: () => MetadataVersion) Review Comment: nit: why is this a `val`? ########## core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala: ########## @@ -132,10 +131,7 @@ class BrokerMetadataPublisher(conf: KafkaConfig, // Publish the new metadata image to the metadata cache. metadataCache.setImage(newImage) - val metadataVersionLogMsg = newImage.features().metadataVersion() match { - case MetadataVersion.UNINITIALIZED => "un-initialized metadata.version" Review Comment: Does broker startup have any guarantees with respect to the metadata version in the log? For example, does the controller require the latest metadata feature record to be consumed before a broker can be unfenced? I'm mainly trying to understand if we might revert to an older version and what the consequences of that would be. ########## core/src/main/scala/kafka/server/ReplicaFetcherThread.scala: ########## @@ -135,7 +137,7 @@ class ReplicaFetcherThread(name: String, def maybeWarnIfOversizedRecords(records: MemoryRecords, topicPartition: TopicPartition): Unit = { // oversized messages don't cause replication to fail from fetch request version 3 (KIP-74) - if (brokerConfig.fetchRequestVersion <= 2 && records.sizeInBytes > 0 && records.validBytes <= 0) + if (metadataVersionSupplier().fetchRequestVersion() <= 2 && records.sizeInBytes > 0 && records.validBytes <= 0) Review Comment: nit: are the parenthesis required for `fetchRequestVersion`? ########## core/src/main/scala/kafka/server/KafkaRaftServer.scala: ########## @@ -185,7 +185,7 @@ object KafkaRaftServer { val bootstrapMetadata = if (config.originals.containsKey(KafkaConfig.InterBrokerProtocolVersionProp)) { BootstrapMetadata.load(Paths.get(config.metadataLogDir), config.interBrokerProtocolVersion) } else { - BootstrapMetadata.load(Paths.get(config.metadataLogDir), MetadataVersion.IBP_3_0_IV0) + BootstrapMetadata.load(Paths.get(config.metadataLogDir), MetadataVersion.MINIMUM_KRAFT_VERSION) Review Comment: nit: the use of "preview" in some of the comments and code seems unnecessary and likely to cause confusion. Could we just say older versions? Instead of `fallbackPreviewVersion`, maybe just `fallbackVersion`? ########## server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java: ########## @@ -43,7 +43,6 @@ * released version, they can use "0.10.0" when upgrading to the 0.10.0 release. */ public enum MetadataVersion { - UNINITIALIZED(-1, "0.0", ""), Review Comment: nit: There is a comment below about omitting UNINITIALIZED that we need to update. -- 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