mumrah commented on code in PR #14594: URL: https://github.com/apache/kafka/pull/14594#discussion_r1373061403
########## core/src/main/scala/kafka/cluster/Partition.scala: ########## @@ -1106,6 +1115,12 @@ class Partition(val topicPartition: TopicPartition, * @return true if the HW was incremented, and false otherwise. */ private def maybeIncrementLeaderHW(leaderLog: UnifiedLog, currentTimeMs: Long = time.milliseconds): Boolean = { + if (metadataCache.isInstanceOf[KRaftMetadataCache] && interBrokerProtocolVersion.isElrSupported && eligibleLeaderReplicasEnabled) { Review Comment: Do we need the check on the metadata cache here? Shouldn't it be impossible to be running in ZK mode with ELR enabled? ########## core/src/main/scala/kafka/cluster/Partition.scala: ########## @@ -359,10 +361,14 @@ class Partition(val topicPartition: TopicPartition, // a false positive under min isr check, it has to check the leaderReplicaIdOpt again. Though it can still be affected // by ABA problems when leader->follower->leader, but it should be good enough for a metric. def isUnderMinIsr: Boolean = { - leaderLogIfLocal.exists { partitionState.isr.size < _.config.minInSyncReplicas } && isLeader + leaderLogIfLocal.exists{partitionState.isr.size < effectiveMinIsr(_) } && isLeader Review Comment: nit: keep the whitespace around the `{` ########## core/src/main/scala/kafka/cluster/Partition.scala: ########## @@ -359,10 +361,14 @@ class Partition(val topicPartition: TopicPartition, // a false positive under min isr check, it has to check the leaderReplicaIdOpt again. Though it can still be affected // by ABA problems when leader->follower->leader, but it should be good enough for a metric. def isUnderMinIsr: Boolean = { - leaderLogIfLocal.exists { partitionState.isr.size < _.config.minInSyncReplicas } && isLeader + leaderLogIfLocal.exists{partitionState.isr.size < effectiveMinIsr(_) } && isLeader } - def isAtMinIsr: Boolean = leaderLogIfLocal.exists { partitionState.isr.size == _.config.minInSyncReplicas } + private def effectiveMinIsr(leaderLog: UnifiedLog): Int = { + leaderLog.config.minInSyncReplicas.min(remoteReplicasMap.size + 1) Review Comment: Just curious, did we ever have a case where the min ISR was larger than the number of replicas? ########## core/src/main/scala/kafka/cluster/Partition.scala: ########## @@ -1106,6 +1115,12 @@ class Partition(val topicPartition: TopicPartition, * @return true if the HW was incremented, and false otherwise. */ private def maybeIncrementLeaderHW(leaderLog: UnifiedLog, currentTimeMs: Long = time.milliseconds): Boolean = { + if (metadataCache.isInstanceOf[KRaftMetadataCache] && interBrokerProtocolVersion.isElrSupported && eligibleLeaderReplicasEnabled) { + if (isUnderMinIsr) { + debug(s"Skip checking whether HWM can advance because partition=$topicPartition is under min ISR(ISR=${partitionState.isr}") Review Comment: Let's set this to TRACE. Also, we can omit the `$topicPartition` part since it's included in the logIdent for Partition ########## core/src/main/scala/kafka/cluster/Partition.scala: ########## @@ -1096,6 +1102,9 @@ class Partition(val topicPartition: TopicPartition, * advancing the HW, the follower's log end offset may keep falling behind the HW (determined by the leader's log end * offset) and therefore will never be added to ISR. * + * Note, If KIP-966 is enabled, the HW can only advance if the ISR size is equal or large than the + * min ISR(min.insync.replicas). Review Comment: how about "larger than the min ISR" -- 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