Skip to content

KAFKA-10427: Fetch snapshot #9553

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 11 commits into from
Dec 29, 2020
Merged

KAFKA-10427: Fetch snapshot #9553

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
277e6a5
KAFKA-10427: Basic fetch snapshot implementation
jsancio Aug 12, 2020
e648377
Implement FetchSnapshot response handling
jsancio Nov 3, 2020
4ea3c9b
Include leader epoch in the FetchSnapshot request
jsancio Dec 2, 2020
4609922
Fix FetchSnapsot request and response class
jsancio Dec 8, 2020
a96d16a
Implement better snapshot id validation in fetch response
jsancio Dec 8, 2020
79fed61
Improve documentation
jsancio Dec 8, 2020
33be3a4
Add position out of range error
jsancio Dec 22, 2020
2393cf7
Add the remaining tests
jsancio Dec 22, 2020
61de910
Merge remote-tracking branch 'upstream/trunk' into kafka-10427-fetch-…
jsancio Dec 23, 2020
cad8284
Invalid snapshot should back off instead of reporting success
jsancio Dec 23, 2020
22133d1
Add missing parsing code
jsancio Dec 23, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1055,14 +1055,14 @@ private boolean handleFetchResponse(
partitionResponse.snapshotId().endOffset(),
partitionResponse.snapshotId().epoch()
);
return false;
return true;
Comment thread
jsancio marked this conversation as resolved.
Outdated
} else if (partitionResponse.snapshotId().endOffset() < 0) {
logger.error(
"The leader sent a snapshot id with a valid epoch {} but with an invalid end offset {}",
partitionResponse.snapshotId().epoch(),
partitionResponse.snapshotId().endOffset()
);
return false;
return true;
} else {
OffsetAndEpoch snapshotId = new OffsetAndEpoch(
partitionResponse.snapshotId().endOffset(),
Expand Down Expand Up @@ -1274,7 +1274,12 @@ private boolean handleFetchSnapshotResponse(
/* The leader deleted the snapshot before the follower could download it. Start over by
Comment thread
jsancio marked this conversation as resolved.
* reseting the fetching snapshot state and sending another fetch request.
*/
logger.trace("Leader doesn't know about snapshot id {}, returned error {} and snapshot id {}", state.fetchingSnapshot(), partitionSnapshot.errorCode(), partitionSnapshot.snapshotId());
logger.trace(
"Leader doesn't know about snapshot id {}, returned error {} and snapshot id {}",
state.fetchingSnapshot(),
partitionSnapshot.errorCode(),
partitionSnapshot.snapshotId()
);
state.setFetchingSnapshot(Optional.empty());
state.resetFetchTimeout(currentTimeMs);
return true;
Expand Down
Loading