You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/05/28 19:16:09 UTC
[GitHub] [kafka] divijvaidya opened a new pull request, #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
divijvaidya opened a new pull request, #12224:
URL: https://github.com/apache/kafka/pull/12224
1\ As per the interface contract for `RaftClient.scheduleAtomicAppend()` [1], it should throw a `NotLeaderException` exception when the provided current leader epoch does not match the current epoch. However, the current `LocalLogManager`'s implementation of the API returns a LONG_MAX instead of throwing an exception. This change fixes the behaviour and makes it consistent with the interface contract.
2\ As per the interface contract for `RaftClient.resign(epoch)` [2] if the parameter epoch does not match the current epoch, this call will be ignored. But in the current `LocalLogManager` implementation the leader epoch might change when the thread if waiting to acquire a lock on `shared.tryAppend()` (note that tryAppend() is a synchronized method). In such a case, if a NotALeaderException is thrown (as per code change in 1\ above), then resign should be ignored.
[1] https://github.com/apache/kafka/blob/trunk/raft/src/main/java/org/apache/kafka/raft/RaftClient.java#L145
[2] https://github.com/apache/kafka/blob/trunk/raft/src/main/java/org/apache/kafka/raft/RaftClient.java#L208
--
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 a diff in pull request #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r892065895
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -227,14 +228,14 @@ synchronized long tryAppend(int nodeId, int epoch, List<ApiMessageAndVersion> ba
synchronized long tryAppend(int nodeId, int epoch, LocalBatch batch) {
if (epoch != leader.epoch()) {
- log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " +
+ log.debug("tryAppend(nodeId={}, epoch={}): the provided epoch does not " +
"match the current leader epoch of {}.", nodeId, epoch, leader.epoch());
- return Long.MAX_VALUE;
+ throw new NotLeaderException("Append failed because the replication is not the current leader");
Review Comment:
Thanks for pointing this out. I have updated LocalLogManager to adhere to the same.
--
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] tombentley commented on a diff in pull request #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
tombentley commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r901413660
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -226,16 +227,19 @@ synchronized long tryAppend(int nodeId, int epoch, List<ApiMessageAndVersion> ba
}
synchronized long tryAppend(int nodeId, int epoch, LocalBatch batch) {
- if (epoch != leader.epoch()) {
- log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " +
- "match the current leader epoch of {}.", nodeId, epoch, leader.epoch());
- return Long.MAX_VALUE;
- }
if (!leader.isLeader(nodeId)) {
- log.trace("tryAppend(nodeId={}, epoch={}): the given node id does not " +
- "match the current leader id of {}.", nodeId, epoch, leader.leaderId());
- return Long.MAX_VALUE;
+ log.debug("tryAppend(nodeId={}, epoch={}): the given node id does not " +
+ "match the current leader id of {}.", nodeId, epoch, leader.leaderId());
+ throw new NotLeaderException("Append failed because the replication is not the current leader");
+ }
+
+ if (epoch < leader.epoch()) {
+ throw new NotLeaderException("Append failed because the epoch doesn't match");
Review Comment:
'doesn't match' => `!=`, which isn't what the check's actually doing. Perhaps something like "... because the given epoch is stale" would be clearer.
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -723,9 +727,35 @@ public long scheduleAtomicAppend(int epoch, List<ApiMessageAndVersion> batch) {
@Override
public void resign(int epoch) {
- LeaderAndEpoch curLeader = leader;
- LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), curLeader.epoch() + 1);
- shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader));
+ if (epoch < 0) {
+ throw new IllegalArgumentException("Attempt to resign from an invalid negative epoch " + epoch);
+ }
+
+ LeaderAndEpoch leaderAndEpoch = leaderAndEpoch();
+ int currentEpoch = leaderAndEpoch.epoch();
+
+ if (epoch > currentEpoch) {
+ throw new IllegalArgumentException("Attempt to resign from epoch " + epoch +
+ " which is larger than the current epoch " + currentEpoch);
Review Comment:
The contract in `RaftClient` still says
```java
* @param epoch the epoch to resign from. If this does not match the current epoch, this
* call will be ignored.
```
Of course `KafkaRaftClient` is doing the same check, so I guess we need to update the `RaftClient` javadoc.
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -723,9 +727,35 @@ public long scheduleAtomicAppend(int epoch, List<ApiMessageAndVersion> batch) {
@Override
public void resign(int epoch) {
- LeaderAndEpoch curLeader = leader;
- LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), curLeader.epoch() + 1);
- shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader));
+ if (epoch < 0) {
+ throw new IllegalArgumentException("Attempt to resign from an invalid negative epoch " + epoch);
+ }
+
+ LeaderAndEpoch leaderAndEpoch = leaderAndEpoch();
+ int currentEpoch = leaderAndEpoch.epoch();
+
+ if (epoch > currentEpoch) {
+ throw new IllegalArgumentException("Attempt to resign from epoch " + epoch +
+ " which is larger than the current epoch " + currentEpoch);
+ } else if (epoch < currentEpoch) {
+ // If the passed epoch is smaller than the current epoch, then it might mean
+ // that the listener has not been notified about a leader change that already
+ // took place. In this case, we consider the call as already fulfilled and
+ // take no further action.
+ log.debug("Ignoring call to resign from epoch {} since it is smaller than the " +
+ "current epoch {}", epoch, currentEpoch);
+ return;
+ }
+
+ LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), currentEpoch + 1);
+ try {
+ shared.tryAppend(nodeId, currentEpoch, new LeaderChangeBatch(nextLeader));
+ } catch (NotLeaderException exp) {
+ // the leader epoch has already advanced. resign is a no op.
+ log.debug("Ignoring call to resign from epoch {} since it is smaller than the " +
+ "current epoch {}", epoch, currentEpoch);
Review Comment:
`tryAppend` can throw `NotLeaderException` when `nodeId` is not the current leader too, so the log message could be confusing.
--
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 #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on PR #12224:
URL: https://github.com/apache/kafka/pull/12224#issuecomment-1149992950
@jsancio this is ready for another review. Thanks for your time again.
--
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 #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r886118202
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -227,14 +228,14 @@ synchronized long tryAppend(int nodeId, int epoch, List<ApiMessageAndVersion> ba
synchronized long tryAppend(int nodeId, int epoch, LocalBatch batch) {
if (epoch != leader.epoch()) {
- log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " +
+ log.debug("tryAppend(nodeId={}, epoch={}): the provided epoch does not " +
"match the current leader epoch of {}.", nodeId, epoch, leader.epoch());
- return Long.MAX_VALUE;
+ throw new NotLeaderException("Append failed because the replication is not the current leader");
Review Comment:
Looks like `KafkaRaftClient` throws `IllegalArgumentException` when the passed `epoch` is greater than the current leader epoch. This is based on `BatchAccumulator`:
```java
if (epoch < this.epoch) {
throw new NotLeaderException("Append failed because the epoch doesn't match");
} else if (epoch > this.epoch) {
throw new IllegalArgumentException("Attempt to append from epoch " + epoch +
" which is larger than the current epoch " + this.epoch);
}
```
--
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 #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on PR #12224:
URL: https://github.com/apache/kafka/pull/12224#issuecomment-1149619718
> For my understanding, what is the motivation for this change?
Apologies for late response @jsancio (I was on leave) and thank you for your review.
It's interesting that you ask this question because there indeed is a motivation. I was trying to fix the flaky test described in https://issues.apache.org/jira/browse/KAFKA-13943. As you will observe from the stack trace, due to the bug fixed here in this PR, a snapshot was being created with LONG_MAX offset and thus leading to further cascading failures. The test is flaky because it only happens when an append request started off from a leader but when the request was waiting for synchronized lock in the tryAppend() method, the leader changed. Sometimes, the leader doesn't change while it is waiting and sometimes it does. With this fix, I hope to fix that bug.
Does this answer your question? Please feel free to add your thoughts to the JIRA directly.
--
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 a diff in pull request #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r903622295
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -226,16 +227,19 @@ synchronized long tryAppend(int nodeId, int epoch, List<ApiMessageAndVersion> ba
}
synchronized long tryAppend(int nodeId, int epoch, LocalBatch batch) {
- if (epoch != leader.epoch()) {
- log.trace("tryAppend(nodeId={}, epoch={}): the provided epoch does not " +
- "match the current leader epoch of {}.", nodeId, epoch, leader.epoch());
- return Long.MAX_VALUE;
- }
if (!leader.isLeader(nodeId)) {
- log.trace("tryAppend(nodeId={}, epoch={}): the given node id does not " +
- "match the current leader id of {}.", nodeId, epoch, leader.leaderId());
- return Long.MAX_VALUE;
+ log.debug("tryAppend(nodeId={}, epoch={}): the given node id does not " +
+ "match the current leader id of {}.", nodeId, epoch, leader.leaderId());
+ throw new NotLeaderException("Append failed because the replication is not the current leader");
+ }
+
+ if (epoch < leader.epoch()) {
+ throw new NotLeaderException("Append failed because the epoch doesn't match");
Review Comment:
Thanks for catching. I have fixed it as per your suggestion in the latest commit.
--
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 #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on PR #12224:
URL: https://github.com/apache/kafka/pull/12224#issuecomment-1156428420
@jsancio please review when you get a chance. Currently multiple tests in `QuorumControllerTest` are acting flaky because we are allowing to create a snapshot with LONG_MAX value. This is making it difficult to review PRs due to flakiness.
This code change fixes that.
--
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 a diff in pull request #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r914233354
##########
raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java:
##########
@@ -132,7 +134,7 @@ public long appendAtomic(int epoch, List<T> records) {
private long append(int epoch, List<T> records, boolean isAtomic) {
if (epoch < this.epoch) {
- throw new NotLeaderException("Append failed because the epoch doesn't match");
+ throw new NotLeaderException("Append failed because the given epoch is stale");
Review Comment:
done. Please review.
--
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 #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on PR #12224:
URL: https://github.com/apache/kafka/pull/12224#issuecomment-1140825549
cc: @jsancio perhaps you would be interested to look into this one since you reviewed https://github.com/apache/kafka/pull/10909
--
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 #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on PR #12224:
URL: https://github.com/apache/kafka/pull/12224#issuecomment-1168406453
@hachikuji requesting your on this PR when you get a chance!
--
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] hachikuji merged pull request #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
hachikuji merged PR #12224:
URL: https://github.com/apache/kafka/pull/12224
--
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 #12224: MINOR: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on PR #12224:
URL: https://github.com/apache/kafka/pull/12224#issuecomment-1140524281
@cmccabe as the primary author of `LocalLogManager`, perhaps you may want to review this change?
Note that the CI is successful for all tests except `testTopicIdUpgradeAfterReassigningPartitions` which is a known flaky test and is being fixed in https://github.com/apache/kafka/pull/11687
--
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 a diff in pull request #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r903622487
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -723,9 +727,35 @@ public long scheduleAtomicAppend(int epoch, List<ApiMessageAndVersion> batch) {
@Override
public void resign(int epoch) {
- LeaderAndEpoch curLeader = leader;
- LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), curLeader.epoch() + 1);
- shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader));
+ if (epoch < 0) {
+ throw new IllegalArgumentException("Attempt to resign from an invalid negative epoch " + epoch);
+ }
+
+ LeaderAndEpoch leaderAndEpoch = leaderAndEpoch();
+ int currentEpoch = leaderAndEpoch.epoch();
+
+ if (epoch > currentEpoch) {
+ throw new IllegalArgumentException("Attempt to resign from epoch " + epoch +
+ " which is larger than the current epoch " + currentEpoch);
Review Comment:
Thanks for catching. I have fixed it as per your suggestion in the latest commit.
##########
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java:
##########
@@ -723,9 +727,35 @@ public long scheduleAtomicAppend(int epoch, List<ApiMessageAndVersion> batch) {
@Override
public void resign(int epoch) {
- LeaderAndEpoch curLeader = leader;
- LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), curLeader.epoch() + 1);
- shared.tryAppend(nodeId, curLeader.epoch(), new LeaderChangeBatch(nextLeader));
+ if (epoch < 0) {
+ throw new IllegalArgumentException("Attempt to resign from an invalid negative epoch " + epoch);
+ }
+
+ LeaderAndEpoch leaderAndEpoch = leaderAndEpoch();
+ int currentEpoch = leaderAndEpoch.epoch();
+
+ if (epoch > currentEpoch) {
+ throw new IllegalArgumentException("Attempt to resign from epoch " + epoch +
+ " which is larger than the current epoch " + currentEpoch);
+ } else if (epoch < currentEpoch) {
+ // If the passed epoch is smaller than the current epoch, then it might mean
+ // that the listener has not been notified about a leader change that already
+ // took place. In this case, we consider the call as already fulfilled and
+ // take no further action.
+ log.debug("Ignoring call to resign from epoch {} since it is smaller than the " +
+ "current epoch {}", epoch, currentEpoch);
+ return;
+ }
+
+ LeaderAndEpoch nextLeader = new LeaderAndEpoch(OptionalInt.empty(), currentEpoch + 1);
+ try {
+ shared.tryAppend(nodeId, currentEpoch, new LeaderChangeBatch(nextLeader));
+ } catch (NotLeaderException exp) {
+ // the leader epoch has already advanced. resign is a no op.
+ log.debug("Ignoring call to resign from epoch {} since it is smaller than the " +
+ "current epoch {}", epoch, currentEpoch);
Review Comment:
Thanks for catching. I have fixed it as per your suggestion in the latest commit.
--
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] hachikuji commented on a diff in pull request #12224: KAFKA-13943: Make LocalLogManager implementation consistent with the RaftClient interface contract
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12224:
URL: https://github.com/apache/kafka/pull/12224#discussion_r914050203
##########
raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java:
##########
@@ -132,7 +134,7 @@ public long appendAtomic(int epoch, List<T> records) {
private long append(int epoch, List<T> records, boolean isAtomic) {
if (epoch < this.epoch) {
- throw new NotLeaderException("Append failed because the epoch doesn't match");
+ throw new NotLeaderException("Append failed because the given epoch is stale");
Review Comment:
nit: could we add the provided epoch and the current epoch to this 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