KAFKA-10427: Fetch snapshot#9553
Conversation
986768f to
56dcc50
Compare
56dcc50 to
514a043
Compare
716d98b to
4609922
Compare
| if (snapshot.sizeInBytes() == partitionSnapshot.size()) { | ||
| // Finished fetching the snapshot. | ||
| snapshot.freeze(); | ||
| state.setFetchingSnapshot(Optional.empty()); |
There was a problem hiding this comment.
The follower needs to update the log start offset and log end offset after it has successfully fetched a snapshot. I want to implement this part in this JIRA: https://issues.apache.org/jira/browse/KAFKA-10820
| // The leader is asking us to fetch a snapshot | ||
|
|
||
| if (partitionResponse.snapshotId().epoch() < 0) { | ||
| throw new KafkaException( |
There was a problem hiding this comment.
Hmm.. The leader has sent a bad response. I think logging an error and retrying might be better than crashing the followers.
There was a problem hiding this comment.
Yeah. I am not sure. If these exception are thrown then it means that the epoch is "valid" but the end offset is not "valid" or vice versa. I think this could happen because of either a buggy remote replica or corrupted data in the remote replica.
I guess you can argue that it is safe to keep retrying.
In the case that the Fetch never succeed do we want to transition to the candidate the state? If so, let me add a test that confirms that.
There was a problem hiding this comment.
Hmm.. I would rather not handle this case specially by transitioning to a candidate. It's not a crazy idea, but I think we should attempt that in a separate PR and consider errors more holistically.
I would suggest that we log an error saying that the remote replica seemed to return an invalid response and just keep fetching. Then a user can see the log message and restart the remote replica.
There was a problem hiding this comment.
I would suggest that we log an error saying that the remote replica seemed to return an invalid response and just keep fetching. Then a user can see the log message and restart the remote replica.
Yeah. This is what I implemented and added a test for it. In other words.
- Log an error message
- Tell the raft client that the response was handle successfully but the fetch timer was not reset
In practice this results in the follower continuing to send Fetch requests. After fetchTimeoutMs the follower will transition to candidate as the existing client code does. See https://github.com/apache/kafka/pull/9553/files#diff-86474ad1438150630c21b29a3da2f6dd79d1357e33ac034f00e5fcef0f2e889cR350
Let me know if this is what you were thinking.
| maxSnapshotSize = Integer.MAX_VALUE; | ||
| } | ||
|
|
||
| // TODO: Make sure that we also limit based on the fetch max bytes configuration |
There was a problem hiding this comment.
I created this issue: https://issues.apache.org/jira/browse/KAFKA-10884
We can fix this for Raft Client's implementation of both Fetch and FetchSnapshot at the same time.
|
I manually reran tests since it has been a week since the last update. Here is a link to the results since they have not been automatically linked: https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-9553/14/. |
3 participants
Implements the code necessary for the leader to response to fetch snapshot requests and for the follower to fetch snapshots. It depends on PR #9512 and that PR should be reviewed first. Included are the following changes.
Leader Changes:
Follower Changes:
Raft followers will start fetching snapshot if the leader sends a Fetch response that includes a SnapshotId.
Raft followers send FetchSnapshot requests if there is a pending download. The same timer is used for both Fetch and FetchSnapshot requests.
Raft follower handle FetchSnapshot responses by comping the bytes to the pending SnapshotWriter. This implementation doesn't fix the replicated log after the snapshot has been downloaded. This will be implemented in a future PR.
More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.
Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.
Committer Checklist (excluded from commit message)