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/08/15 14:48:51 UTC
[GitHub] [kafka] jsancio opened a new pull request, #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
jsancio opened a new pull request, #14213:
URL: https://github.com/apache/kafka/pull/14213
In a non-empty log the KRaft leader only notifies the listener of leadership when it has read to the leader's epoch start offset. This guarantees that the leader epoch has been committed and that the listener has read all committed offset/records.
Unfortunately, the KRaft leader doesn't do this when the log is empty. When the log is empty the listener is notified immediately when it has become leader. This makes the API inconsistent and harder to program against.
This change fixes that by having the KRaft leader wait for the listener's nextOffset to be greater than the leader's epoch start offset before calling handleLeaderChange.
The RecordsBatchReader implementation is also changed to include control records. This makes it possible for the state machine learn about committed control records. This additional information can be used to compute the committed offset or for counting those bytes when determining when to snapshot the partition.
### 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] cmccabe commented on pull request #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #14213:
URL: https://github.com/apache/kafka/pull/14213#issuecomment-1679620704
> The RecordsBatchReader implementation is also changed to include control records. This makes it possible for the state machine learn about committed control records. This additional information can be used to compute the committed offset or for counting those bytes when determining when to snapshot the partition.
I think we should do this change separately, since it may require downstream changes. I think a few implementations assume they don't get "empty" batches (that don't have records), which would no longer be true if control record batches were sent.
--
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 #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14213:
URL: https://github.com/apache/kafka/pull/14213#discussion_r1297478095
##########
raft/src/main/java/org/apache/kafka/raft/internals/RecordsBatchReader.java:
##########
@@ -117,14 +117,8 @@ private void ensureOpen() {
}
private Optional<Batch<T>> nextBatch() {
- while (iterator.hasNext()) {
- Batch<T> batch = iterator.next();
-
- if (batch.records().isEmpty()) {
- lastReturnedOffset = batch.lastOffset();
- } else {
- return Optional.of(batch);
- }
+ if (iterator.hasNext()) {
Review Comment:
I agree. I added a couple of tests for 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] hachikuji commented on a diff in pull request #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "hachikuji (via GitHub)" <gi...@apache.org>.
hachikuji commented on code in PR #14213:
URL: https://github.com/apache/kafka/pull/14213#discussion_r1296563672
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2653,11 +2652,11 @@ private boolean shouldFireLeaderChange(LeaderAndEpoch leaderAndEpoch) {
}
private void maybeFireLeaderChange(LeaderAndEpoch leaderAndEpoch, long epochStartOffset) {
- // If this node is becoming the leader, then we can fire `handleClaim` as soon
+ // If this node is becoming the leader, then we can fire `handleLeaderChange` as soon
Review Comment:
Maybe useful to mention the point about the high watermark as well since it is a little subtle.
##########
raft/src/main/java/org/apache/kafka/raft/internals/RecordsBatchReader.java:
##########
@@ -117,14 +117,8 @@ private void ensureOpen() {
}
private Optional<Batch<T>> nextBatch() {
- while (iterator.hasNext()) {
- Batch<T> batch = iterator.next();
-
- if (batch.records().isEmpty()) {
- lastReturnedOffset = batch.lastOffset();
- } else {
- return Optional.of(batch);
- }
+ if (iterator.hasNext()) {
Review Comment:
We're testing this change indirectly, but perhaps it is useful to have an explicit test in `RecordsBatchReaderTest` as well.
--
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 #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #14213:
URL: https://github.com/apache/kafka/pull/14213#discussion_r1297478419
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2653,11 +2652,11 @@ private boolean shouldFireLeaderChange(LeaderAndEpoch leaderAndEpoch) {
}
private void maybeFireLeaderChange(LeaderAndEpoch leaderAndEpoch, long epochStartOffset) {
- // If this node is becoming the leader, then we can fire `handleClaim` as soon
+ // If this node is becoming the leader, then we can fire `handleLeaderChange` as soon
Review Comment:
I added a few more sentences to this comment.
--
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 #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "hachikuji (via GitHub)" <gi...@apache.org>.
hachikuji commented on code in PR #14213:
URL: https://github.com/apache/kafka/pull/14213#discussion_r1297486408
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2656,6 +2656,12 @@ private void maybeFireLeaderChange(LeaderAndEpoch leaderAndEpoch, long epochStar
// as the listener has caught up to the start of the leader epoch. This guarantees
// that the state machine has seen the full committed state before it becomes
// leader and begins writing to the log.
+ //
+ // Note that the raft client doesn't need to compare nextOffset against the high-watermark
+ // to gurantee that the listener has caught up to the high-watermark. This is true because
Review Comment:
nit: guarantee
--
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 #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #14213:
URL: https://github.com/apache/kafka/pull/14213#issuecomment-1683208168
Unable to reproduce the test failures locally.
--
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 #14213: KAFKA-15345; KRaft leader notifies when listener reaches epoch start
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #14213:
URL: https://github.com/apache/kafka/pull/14213
--
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