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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
