From 53cafd82e55bbd6da5681107187f6b66fdee74f3 Mon Sep 17 00:00:00 2001 From: Neil Twigg Date: Tue, 3 Dec 2024 09:15:33 +0000 Subject: [PATCH] Re-add `"stream"` to consumer assignments This was changed in #6189 but we need it to be in the snapshot after all in order to be able to safely roll back to earlier versions. Signed-off-by: Neil Twigg --- server/jetstream_cluster.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go index 2cebbc5aa8..c5d67078a2 100644 --- a/server/jetstream_cluster.go +++ b/server/jetstream_cluster.go @@ -150,7 +150,7 @@ type consumerAssignment struct { Client *ClientInfo `json:"client,omitempty"` Created time.Time `json:"created"` Name string `json:"name"` - Stream string `json:"stream,omitempty"` // Not actually optional but empty in meta snapshots. + Stream string `json:"stream"` Config *ConsumerConfig `json:"consumer"` Group *raftGroup `json:"group"` Subject string `json:"subject,omitempty"` @@ -1556,7 +1556,7 @@ func (js *jetStream) metaSnapshot() []byte { continue } cca := *ca - cca.Stream = _EMPTY_ // Will be rehydrated from parent stream assignment on decoding. + cca.Stream = wsa.Config.Name // Needed for safe roll-backs. cca.Client = cca.Client.forAssignmentSnap() cca.Subject, cca.Reply = _EMPTY_, _EMPTY_ wsa.Consumers = append(wsa.Consumers, &cca) @@ -1615,7 +1615,9 @@ func (js *jetStream) applyMetaSnapshot(buf []byte, ru *recoveryUpdates, isRecove if len(wsa.Consumers) > 0 { sa.consumers = make(map[string]*consumerAssignment) for _, ca := range wsa.Consumers { - ca.Stream = sa.Config.Name // Rehydrate from the stream name. + if ca.Stream == _EMPTY_ { + ca.Stream = sa.Config.Name // Rehydrate from the stream name. + } sa.consumers[ca.Name] = ca } }