Skip to content
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

Jump to bottom

Wait for recovery to complete before sending snapshot on stream scale-up from R1 #5001

Merged
merged 1 commit into from
Jan 26, 2024
Merged

Wait for recovery to complete before sending snapshot on stream scale-up from R1 #5001

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
12 changes: 10 additions & 2 deletions server/jetstream_cluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2313,7 +2313,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
// We can arrive here NOT being the leader, so we send the snapshot only if we are, and in this case
// reset the notion that we need to send the snapshot. If we are not, then the first time the server
// will switch to leader (in the loop below), we will send the snapshot.
if sendSnapshot && isLeader && mset != nil && n != nil {
if sendSnapshot && isLeader && mset != nil && n != nil && !isRecovering {
n.SendSnapshot(mset.stateSnapshot())
sendSnapshot = false
}
Expand All @@ -2337,6 +2337,12 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
if _, b := n.Size(); b > compactSizeMin || n.NeedSnapshot() {
doSnapshot()
}
// If we became leader during this time and we need to send a snapshot to our
// followers, i.e. as a result of a scale-up from R1, do it now.
if sendSnapshot && isLeader && mset != nil && n != nil {
n.SendSnapshot(mset.stateSnapshot())
sendSnapshot = false
}
continue
}
// Apply our entries.
Expand Down Expand Up @@ -2376,7 +2382,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps

case isLeader = <-lch:
if isLeader {
if mset != nil && n != nil && sendSnapshot {
if mset != nil && n != nil && sendSnapshot && !isRecovering {
// If we *are* recovering at the time then this will get done when the apply queue
// handles the nil guard to show the catchup ended.
n.SendSnapshot(mset.stateSnapshot())
sendSnapshot = false
}
Expand Down