gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-02 03:38:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into dev

Browse Source
This commit is contained in:
Derek Collison
2023-06-05 14:13:18 -07:00
parent af318be5db 2e2ac33920
commit f342f6a758
11 changed files with 140 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
server/jetstream_cluster.go
View File

@@ -3108,7 +3108,9 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
accStreams = make(map[string]*streamAssignment)
} else if osa := accStreams[stream]; osa != nil && osa != sa {
// Copy over private existing state from former SA.
sa.Group.node = osa.Group.node
if sa.Group != nil {
sa.Group.node = osa.Group.node
}
sa.consumers = osa.consumers
sa.responded = osa.responded
sa.err = osa.err
@@ -3199,7 +3201,9 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
}
// Copy over private existing state from former SA.
sa.Group.node = osa.Group.node
if sa.Group != nil {
sa.Group.node = osa.Group.node
}
sa.consumers = osa.consumers
sa.err = osa.err
@@ -3217,7 +3221,9 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
sa.responded = false
} else {
// Make sure to clean up any old node in case this stream moves back here.
sa.Group.node = nil
if sa.Group != nil {
sa.Group.node = nil
}
}
js.mu.Unlock()
@@ -3790,7 +3796,9 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
} else if oca := sa.consumers[ca.Name]; oca != nil {
wasExisting = true
// Copy over private existing state from former SA.
ca.Group.node = oca.Group.node
if ca.Group != nil {
ca.Group.node = oca.Group.node
}
ca.responded = oca.responded
ca.err = oca.err
}
@@ -3905,8 +3913,12 @@ func (js *jetStream) processConsumerRemoval(ca *consumerAssignment) {
var needDelete bool
if accStreams := cc.streams[ca.Client.serviceAccount()]; accStreams != nil {
if sa := accStreams[ca.Stream]; sa != nil && sa.consumers != nil && sa.consumers[ca.Name] != nil {
needDelete = true
delete(sa.consumers, ca.Name)
oca := sa.consumers[ca.Name]
// Make sure this removal is for what we have, otherwise ignore.
if ca.Group != nil && oca.Group != nil && ca.Group.Name == oca.Group.Name {
needDelete = true
delete(sa.consumers, ca.Name)
}
}
}
js.mu.Unlock()