You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "CalvinConfluent (via GitHub)" <gi...@apache.org> on 2023/10/19 18:50:55 UTC
[PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
CalvinConfluent opened a new pull request, #14594:
URL: https://github.com/apache/kafka/pull/14594
Enforce a new rule of the HWM advance. Only advance HWM if
- Using ELR
- ISR size >= min ISR
https://issues.apache.org/jira/browse/KAFKA-15583
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r2867528688
##########
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:
opened https://issues.apache.org/jira/browse/KAFKA-20235 as a follow-up
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r2620028116
##########
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:
> When modifying min.insync.replicas or adjusting the replica count, we could check whether the change is valid.
It would be a useful change for the user experience. But it is not included in the KIP. The KIP utilize an workaround as @artemlivshits quoted above.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on PR #14594:
URL: https://github.com/apache/kafka/pull/14594#issuecomment-1793174790
The failing tests are irrelevant.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1380240564
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -359,10 +359,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 = {
Review Comment:
Can we add a short doc string for this method?
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1106,6 +1112,10 @@ 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 (isUnderMinIsr) {
+ trace(s"Skip checking whether HWM can advance because partition is under min ISR(ISR=${partitionState.isr}")
Review Comment:
Should we word this more like "Not increasing HWM..." so it's explicit that the HWM is not increasing?
##########
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:
Yea, it makes sense to constrain the min ISR like this, I was just wondering if we actually saw this problem in practice. Either way, I'm fine with fixing this here.
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -359,10 +359,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)
+ }
+
+ def isAtMinIsr: Boolean = leaderLogIfLocal.exists { partitionState.isr.size == effectiveMinIsr(_) }
Review Comment:
Can we add a short doc string for this that mentions the constraint on 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1380568178
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -359,10 +359,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)
+ }
+
+ def isAtMinIsr: Boolean = leaderLogIfLocal.exists { partitionState.isr.size == effectiveMinIsr(_) }
Review Comment:
Done
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373427855
##########
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:
I guess ZK can also upgrade to 3.7IV1 and the elrEnabled is a static config shared between Kraft and ZK?
##########
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:
Thanks! Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373421615
##########
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:
The removed UT tries to set the min ISR larger than the replication factor.
Basically, we don't have any restraints over the minISR value. If someone set or has set the minISR wrongly, it will be wield availability issue after the upgrade.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373424287
##########
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:
When ISR size is equal to min ISR, the HW can advance.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r2606499739
##########
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:
Restricting `min.insync.replicas` to not exceed the replica count may have more practical significance, but it could also lead to some compatibility issues.
When modifying `min.insync.replicas` or adjusting the replica count, we could check whether the change is valid. If it is not, a warning log could be printed to remind users that such an invalid configuration will be prohibited in a future release (e.g., version 5.0). What do you think of this approach?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r2606393474
##########
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:
Sorry for the noise on this old PR. Was this behavior change documented? Crucially, this modification seems to let users produce with `acks=all` even if the replicas count is below `min.insync.replicas`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1380567962
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1106,6 +1112,10 @@ 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 (isUnderMinIsr) {
+ trace(s"Skip checking whether HWM can advance because partition is under min ISR(ISR=${partitionState.isr}")
Review Comment:
Done.
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -359,10 +359,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 = {
Review Comment:
Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373418439
##########
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:
Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah merged PR #14594:
URL: https://github.com/apache/kafka/pull/14594
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373421615
##########
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:
The removed UT tries to set the min ISR larger than the replication factor.
Basically, we don't have any requirements over the minISR value. If someone set or has set the minISR wrongly, it will be wield availability issue after the upgrade.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373832091
##########
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:
Agree. Now make it a general requirement.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR With ELR [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r1373758559
##########
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:
I have a higher-level question -- do we actually care if ELR is enabled for this logic? It is my understanding that the logic on the brokers doesn't use ELR, it just has some changes that seem to improve the protocol even if ELR is not used / implemented:
- strict min ISR
- advance HWM only only if min ISR known to the controller
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r2606515187
##########
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:
> If it is not, a warning log could be printed to remind users that such an invalid configuration will be prohibited in a future release (e.g., version 5.0). What do you think of this approach?
As this behaviour has not been discussed via a KIP, we should limit ourself to explaining the potential issues users could face by simply issuing a warning message
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15583: Enforce HWM advance only if partition is not under-min-ISR [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #14594:
URL: https://github.com/apache/kafka/pull/14594#discussion_r2615974040
##########
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:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=263426970#KIP966:EligibleLeaderReplicas-Compatibility,Deprecation,andMigrationPlan says
- min.insync.replicas will no longer be effective to be larger than the replication factor. For existing configs, the min.insync.replicas will be min(min.insync.replicas, replication factor).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org