You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jsancio (via GitHub)" <gi...@apache.org> on 2023/05/25 17:28:56 UTC
[GitHub] [kafka] jsancio opened a new pull request, #13765: KAFKA-15021; Skip leader epoch bump
jsancio opened a new pull request, #13765:
URL: https://github.com/apache/kafka/pull/13765
TODO: Write description
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] jsancio commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1573978281
> 1. We change leader epoch when a replica is removed and shrinks the ISR (without a leader-re-election). Is that correct? Is yes, then should we also removing the logic to increment the epoch in such situations to keep the definition of leader epoch consistent?
Yes. That is what the change to `PartittionChangeBuilder` does. Can you point me to what code exactly are you referring to?
> 2. Is a similar change required for Zk code path?
We need to keep the old behavior in ZK deployments because the ZK controller doesn't implement KIP-841 which is required for this fix to work.
--
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
[GitHub] [kafka] divijvaidya commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1567981427
Hey @jsancio
With this change, we are changing the semantics of what a leadership epoch means. Prior to this change, leadership epoch is a version number representing membership of an ISR. As soon as membership changes, this version changes. After this change, the definition has changed to - leadership epoch is a version number that represents member of an ISR "in some cases". As you can see, the new definition has added ifs and buts to the simple definition above. Hence, I am not in favour of changing this.
To achieve the objective that you desired, there is another way without changing the definition, i.e. change how the components reacts when the version/epoch is changes. We can choose to not restart the fetcher threads on each replica when an shrink ISR with leadership epoch change arrives to it for processing.
Thoughts?
--
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
[GitHub] [kafka] splett2 commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "splett2 (via GitHub)" <gi...@apache.org>.
splett2 commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1205862927
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1087,12 +1087,14 @@ class Partition(val topicPartition: TopicPartition,
// avoid unnecessary collection generation
val leaderLogEndOffset = leaderLog.logEndOffsetMetadata
var newHighWatermark = leaderLogEndOffset
- remoteReplicasMap.values.foreach { replica =>
+ remoteReplicasMap.foreachEntry { (replicaId, replica) =>
Review Comment:
can we use `remoteReplicasMap.values` here and use the replica.brokerId similar to the maximalIsr.contains call?
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1218084219
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1091,8 +1091,10 @@ class Partition(val topicPartition: TopicPartition,
// Note here we are using the "maximal", see explanation above
val replicaState = replica.stateSnapshot
if (replicaState.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
- (replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs)
- || partitionState.maximalIsr.contains(replica.brokerId))) {
+ ((replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs) &&
+ isReplicaIsrEligible(replica.brokerId)) ||
Review Comment:
Agree. I called it `shouldWaitForReplica` since I think this is what the leader is trying to do.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1214750188
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1091,8 +1091,10 @@ class Partition(val topicPartition: TopicPartition,
// Note here we are using the "maximal", see explanation above
val replicaState = replica.stateSnapshot
if (replicaState.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
- (replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs)
- || partitionState.maximalIsr.contains(replica.brokerId))) {
+ ((replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs) &&
+ isReplicaIsrEligible(replica.brokerId)) ||
Review Comment:
Is it worth extracting this condition into an helper method (e.g. isIsrEligibleAndCaughtUp)? That would simplify the condition.
##########
core/src/test/scala/unit/kafka/cluster/PartitionTest.scala:
##########
@@ -1456,6 +1456,105 @@ class PartitionTest extends AbstractPartitionTest {
assertEquals(alterPartitionListener.failures.get, 1)
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("fenced", "shutdown", "unfenced"))
+ def testHWMIncreasesWithFencedOrShutdownFollower(brokerState: String): Unit = {
Review Comment:
nit: s/HWM/HighWatermark?
##########
core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala:
##########
@@ -357,6 +363,51 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
}
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testSendToPartitionWithFollowerShutdown(quorum: String): Unit = {
Review Comment:
nit: `*ShouldNotTimeout`? it would be great to capture the issue in the test name or to add a comment about it.
##########
metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java:
##########
@@ -260,11 +265,17 @@ private void tryElection(PartitionChangeRecord record) {
* NO_LEADER_CHANGE, a leader epoch bump will automatically occur. That takes care of
* case 1. In this function, we check for cases 2 and 3, and handle them by manually
* setting record.leader to the current leader.
+ *
+ * In MV before 3.6 there was a bug (KAFKA-15021) in the brokers' replica manager
+ * that required that the leader epoch be bump whenever the ISR shrank. In MV 3.6 this leader
+ * bump is not required when the ISR shrinks.
*/
void triggerLeaderEpochBumpIfNeeded(PartitionChangeRecord record) {
Review Comment:
For my understanding, do we bump the leader epoch when the ISR is expanded? My understanding is that we don't.
--
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
[GitHub] [kafka] splett2 commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "splett2 (via GitHub)" <gi...@apache.org>.
splett2 commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1207073563
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1087,12 +1087,14 @@ class Partition(val topicPartition: TopicPartition,
// avoid unnecessary collection generation
val leaderLogEndOffset = leaderLog.logEndOffsetMetadata
var newHighWatermark = leaderLogEndOffset
- remoteReplicasMap.values.foreach { replica =>
+ remoteReplicasMap.foreachEntry { (replicaId, replica) =>
Review Comment:
Should we have a test in `PartitionTest` to assert that the HWM is incremented when there is a replica that is fenced but caught 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
[GitHub] [kafka] dajac commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1568017910
> Yes, that is fair too. The definition of leadership epoch in which case changes to - it represents the version of the leader after a re-election. In this case, we should also remove the epoch change during ISR expansion as well. My point is, let's keep the definition as state of ISR (current) or state of leader (in which case we remove epoch change for both expansion and shrink).
Yeah, I agree that we need to do both in order to remain consistent.
> Aside for it, out of curiosity, is there any other version which represents the state of ISR in Kafka? Does replica epoch changes on every change to ISR?
There is the partition epoch which is incremented whenever the partition is updated. This includes ISR changes.
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1218088303
##########
core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala:
##########
@@ -357,6 +363,51 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
}
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("zk", "kraft"))
+ def testSendToPartitionWithFollowerShutdown(quorum: String): Unit = {
Review Comment:
Done and 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
[GitHub] [kafka] splett2 commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "splett2 (via GitHub)" <gi...@apache.org>.
splett2 commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1573900120
@jsancio
That makes sense. Sounds good to me in that case.
--
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
[GitHub] [kafka] divijvaidya commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1573957094
> the old code was increasing the leader epoch when with the ISR shrinks but not when the ISR expands
Thank you, I didn't realise that.
Next,
1. We change leader epoch when a replica is removed and shrinks the ISR (without a leader-re-election). Is that correct? Is yes, then should we also removing the logic to increment the epoch in such situations to keep the definition of leader epoch consistent?
2. Is a similar change required for Zk code path?
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1218098813
##########
metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java:
##########
@@ -260,11 +265,17 @@ private void tryElection(PartitionChangeRecord record) {
* NO_LEADER_CHANGE, a leader epoch bump will automatically occur. That takes care of
* case 1. In this function, we check for cases 2 and 3, and handle them by manually
* setting record.leader to the current leader.
+ *
+ * In MV before 3.6 there was a bug (KAFKA-15021) in the brokers' replica manager
+ * that required that the leader epoch be bump whenever the ISR shrank. In MV 3.6 this leader
+ * bump is not required when the ISR shrinks.
*/
void triggerLeaderEpochBumpIfNeeded(PartitionChangeRecord record) {
Review Comment:
Correct. The `Replica.contains` check is subtle but it returns `true` if the second list is a subset of the first list. I added a comment about this.
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1218084219
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1091,8 +1091,10 @@ class Partition(val topicPartition: TopicPartition,
// Note here we are using the "maximal", see explanation above
val replicaState = replica.stateSnapshot
if (replicaState.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
- (replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs)
- || partitionState.maximalIsr.contains(replica.brokerId))) {
+ ((replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs) &&
+ isReplicaIsrEligible(replica.brokerId)) ||
Review Comment:
I agree. I called it `shouldWaitForReplica` since I think this is what the leader is trying to do.
--
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
[GitHub] [kafka] dajac commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1568009841
> This change also fixes a bug when computing the HWM. When computing the HWM, replicas that are not eligible to join the ISR but are caught up should not be included in the computation. Otherwise, the HWM will never increase for replica.lag.time.max.ms because the shutting down replica is not sending FETCH request. Without this additional fix PRODUCE requests would timeout if the request timeout is greater than replica.lag.time.max.ms.
I think that the real issue is in `Partition.makeLeader`. As you can see there, we only reset the followers' states when the leader epoch is bumped. I suppose that this is why you stumbled upon this issue with having shutting down replicas holding back advancing the HWM. The issue is that the shutting down replica's state is not reset so it remains caught-up for `replica.lag.time.max.ms`. I think that we need to update `Partition.makeLeader` to always update the followers' states. Obviously, we also need your changes to not consider fenced and shutting down replicas in the HWM computation.
> Because of the bug above the KRaft controller needs to check the MV to guarantee that all brokers support this bug fix before skipping the leader epoch bump.
I wonder if we really need this if we change `Partition.makeLeader` as explained. It seems to me that the change in `Partition.makeLeader` and `Partition.maybeIncrementLeaderHW` together should be backward compatible. What do you think?
--
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
[GitHub] [kafka] jsancio commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1572275711
Thanks for your feedback @divijvaidya @dajac. I am replying to both your comments in you message.
> With this change, we are changing the semantics of what a leadership epoch means. Prior to this change, leadership epoch is a version number representing membership of an ISR. As soon as membership changes, this version changes.
@divijvaidya, The old code was increasing the leader epoch when with the ISR shrinks but not when the ISR expands. My understanding that we were doing this because the old replica manager used leader epoch bump to invalidate old fetchers. During shutdown the fetchers needed to be invalidated to avoid having them rejoin the ISR. With KIP-841, this is no longer necessary as we can reject brokers that are shutting down from joining the ISR and modifying the HWM.
Part of the code for doing this already exists, what we missing and what part of this PR fixes, is considering this state when advancing the HWM. The partition leader should not include shutting down replicas when determining the HWM.
> After this change, the definition has changed to - leadership epoch is a version number that represents member of an ISR "in some cases". As you can see, the new definition has added ifs and buts to the simple definition above. Hence, I am not in favour of changing this.
@divijvaidya, For correctness, the main requirement is that the leader epoch is increase whenever the leader changes. This is needed for log truncation and reconciliation. For log consistency, log truncation and reconciliation assumes that the tuples (offset, epoch) are unique per topic partition and that if the tuple (offset, epoch) match in two replicas then their log up to that offset also match. In my opinion, for correctness Kafka doesn't require that the leader epoch is increased when the ISR changes.
> As you can see there, we only reset the followers' states when the leader epoch is bumped. I suppose that this is why you stumbled upon this issue with having shutting down replicas holding back advancing the HWM. The issue is that the shutting down replica's state is not reset so it remains caught-up for replica.lag.time.max.ms. I think that we need to update Partition.makeLeader to always update the followers' states. Obviously, we also need your changes to not consider fenced and shutting down replicas in the HWM computation.
@dajac, Yes, I thought about this when I was implemented this feature. I decided against it because the follower (shutting down replica) is technically "caught up" to the leader we simply don't want the leader to wait for the replica when computing the HWM since we know it will soon be shutting down its fetchers.
> I wonder if we really need this if we change Partition.makeLeader as explained. It seems to me that the change in Partition.makeLeader and Partition.maybeIncrementLeaderHW together should be backward compatible. What do you think?
We need the MV check in the controller even with your suggestion. The question is "When is it beneficial for the controller to not increase the leader epoch when a replica is removed from the ISR because of shutdown?" This is only the case when the controller knows that the brokers have the replica manager fixes in this PR. That is guarantee to be the case if the MV is greater than the MV introduced in this PR.
If the brokers don't contain the fixes in this PR and the controller doesn't bump the leader epoch, PRODUCE requests will timeout because the HWM increase will be delayed.
--
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
[GitHub] [kafka] jsancio commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1573750489
> Actually, one thing I was wondering about this change since I am not that familiar with what the metadata version gates - does the change in the PR allow leader epochs to go backwards?
The important observation is that this PR doesn't change the semantic of replaying `PartitionRecord` and `PartitionChangeRecord` with with respect to leader epoch bump. When replaying `PartitionChangeRecord` the state machines (controller and broker) will increase the leader epoch if the field `Leader` is set. This holds true before and after this PR. What this PR changes is when the controller sets the `Leader` field in `PartitionChangeRecord`.
I should also point out that the MV check is not require for correctness. It is there for performance so that the PRODUCER request doesn't timeout and so that the Kafka producer doesn't have to retry the PRODUCE 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
[GitHub] [kafka] jsancio merged pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #13765:
URL: https://github.com/apache/kafka/pull/13765
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1213501789
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1087,12 +1087,14 @@ class Partition(val topicPartition: TopicPartition,
// avoid unnecessary collection generation
val leaderLogEndOffset = leaderLog.logEndOffsetMetadata
var newHighWatermark = leaderLogEndOffset
- remoteReplicasMap.values.foreach { replica =>
+ remoteReplicasMap.foreachEntry { (replicaId, replica) =>
Review Comment:
Done and Done. I confirmed that the test I added fails for the "fenced" and "shutdown" variant against trunk.
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1218084219
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1091,8 +1091,10 @@ class Partition(val topicPartition: TopicPartition,
// Note here we are using the "maximal", see explanation above
val replicaState = replica.stateSnapshot
if (replicaState.logEndOffsetMetadata.messageOffset < newHighWatermark.messageOffset &&
- (replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs)
- || partitionState.maximalIsr.contains(replica.brokerId))) {
+ ((replicaState.isCaughtUp(leaderLogEndOffset.messageOffset, currentTimeMs, replicaLagTimeMaxMs) &&
+ isReplicaIsrEligible(replica.brokerId)) ||
Review Comment:
I agree. I called it `shouldWaitForReplicaToJoinIsr` since I think this is what the leader is trying to do.
--
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
[GitHub] [kafka] jsancio commented on a diff in pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #13765:
URL: https://github.com/apache/kafka/pull/13765#discussion_r1218103438
##########
core/src/test/scala/unit/kafka/cluster/PartitionTest.scala:
##########
@@ -1456,6 +1456,105 @@ class PartitionTest extends AbstractPartitionTest {
assertEquals(alterPartitionListener.failures.get, 1)
}
+ @ParameterizedTest
+ @ValueSource(strings = Array("fenced", "shutdown", "unfenced"))
+ def testHWMIncreasesWithFencedOrShutdownFollower(brokerState: String): Unit = {
Review Comment:
Done and I added a comment to the last check in the test.
--
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
[GitHub] [kafka] dajac commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1568001597
> To achieve the objective that you desired, there is another way without changing the definition, i.e. change how the components reacts when the version/epoch is changes. We can choose to not restart the fetcher threads on each replica when an shrink ISR with leadership epoch change arrives to it for processing.
@divijvaidya This does not help. Restarting the fetcher threads is just a mean to provide them the new leader epoch that they have to use. Until they get it, they can't replicate. This is the annoying part. If you don't restart the fetchers and update the leader epoch "live", you still have that period of time during which the followers don't have the correct leader epoch. Note that the bump also have an impact on producers/consumers as they have to refresh their metadata. Overall, I think that the goal is to only bump the leader epoch on leadership changes to avoid all those disturbances.
--
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
[GitHub] [kafka] divijvaidya commented on pull request #13765: KAFKA-15021; Skip leader epoch bump on ISR shrink
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #13765:
URL: https://github.com/apache/kafka/pull/13765#issuecomment-1568008983
> Overall, I think that the goal is to only bump the leader epoch on leadership changes to avoid all those disturbances.
Yes, that is fair too. The definition of leadership epoch in which case changes to - it represents the version of the leader after a re-election. In this case, we should also remove the epoch change during ISR expansion as well. My point is, let's keep the definition as state of ISR (current) or state of leader (in which case we remove epoch change for both expansion and shrink).
Aside for it, out of curiosity, is there any other version which represents the state of ISR in Kafka? Does replica epoch changes on every change to 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