Repository: nats-io/nats-server
Pull request #6535 (2025-02-20 13:28:55)

[FIXED] (2.11) Idempotent stream/consumer create after server upgrade
If a stream or consumer was created prior to 2.11 there would be no versioning metadata. After the server upgrade sending an identical create request would fail with `stream already exists`. If a stream or consumer already exists without metadata, creates are now idempotent. Which also means that the new versioning metadata will only be set if creating a new stream/consumer, or when updating it.

Also, the signatures of `setStaticStreamMetadata` and `setStaticConsumerMetadata` were updated, since the second parameter is unused. Cleaned that up, along with duplicate test cases in `jetstream_versioning_test.go`.

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>



Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 3e7cd97b..6d035c96 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1593,7 +1593,7 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 	}
 
 	// Initialize asset version metadata.
-	setStaticStreamMetadata(&cfg.StreamConfig, nil)
+	setStaticStreamMetadata(&cfg.StreamConfig)
 
 	streamName := streamNameFromSubject(subject)
 	if streamName != cfg.Name {
@@ -1732,7 +1732,7 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 	}
 
 	// Update asset version metadata.
-	setStaticStreamMetadata(&cfg, &mset.cfg)
+	setStaticStreamMetadata(&cfg)
 
 	if err := mset.updatePedantic(&cfg, ncfg.Pedantic); err != nil {
 		resp.Error = NewJSStreamUpdateError(err, Unless(err))
@@ -4427,16 +4427,14 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 		return
 	}
 
-	var oldCfg *ConsumerConfig
 	if o := stream.lookupConsumer(consumerName); o != nil {
-		oldCfg = &o.cfg
 		// If the consumer already exists then don't allow updating the PauseUntil, just set
 		// it back to whatever the current configured value is.
 		req.Config.PauseUntil = o.cfg.PauseUntil
 	}
 
 	// Initialize/update asset version metadata.
-	setStaticConsumerMetadata(&req.Config, oldCfg)
+	setStaticConsumerMetadata(&req.Config)
 
 	o, err := stream.addConsumerWithAction(&req.Config, req.Action, req.Pedantic)
 
@@ -5010,7 +5008,7 @@ func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account
 
 		// Update asset version metadata due to updating pause/resume.
 		// Only PauseUntil is updated above, so reuse config for both.
-		setStaticConsumerMetadata(nca.Config, nca.Config)
+		setStaticConsumerMetadata(nca.Config)
 
 		eca := encodeAddConsumerAssignment(&nca)
 		cc.meta.Propose(eca)
@@ -5046,8 +5044,7 @@ func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account
 	}
 
 	// Update asset version metadata due to updating pause/resume.
-	// Only PauseUntil is updated above, so reuse config for both.
-	setStaticConsumerMetadata(&ncfg, &ncfg)
+	setStaticConsumerMetadata(&ncfg)
 
 	if err := obs.updateConfig(&ncfg); err != nil {
 		// The only type of error that should be returned here is from o.store,
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index de1dadf4..38eacc5a 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -6120,6 +6120,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 
 	// Capture if we have existing assignment first.
 	if osa := js.streamAssignment(acc.Name, cfg.Name); osa != nil {
+		copyStreamMetadata(cfg, osa.Config)
 		if !reflect.DeepEqual(osa.Config, cfg) {
 			resp.Error = NewJSStreamNameExistError()
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -6267,7 +6268,6 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
 
 	osa := js.streamAssignment(acc.Name, cfg.Name)
-
 	if osa == nil {
 		resp.Error = NewJSStreamNotFoundError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -6275,7 +6275,7 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	}
 
 	// Update asset version metadata.
-	setStaticStreamMetadata(cfg, osa.Config)
+	setStaticStreamMetadata(cfg)
 
 	var newCfg *StreamConfig
 	if jsa := js.accounts[acc.Name]; jsa != nil {
@@ -7398,15 +7398,19 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 				s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 				return
 			}
+		} else {
+			// Initialize/update asset version metadata.
+			// First time creating this consumer, or updating.
+			setStaticConsumerMetadata(cfg)
 		}
 	}
 
 	// Initialize/update asset version metadata.
-	var oldCfg *ConsumerConfig
-	if ca != nil {
-		oldCfg = ca.Config
+	// But only if we're not creating, should only update it the first time
+	// to be idempotent with versions where there's no versioning metadata.
+	if action != ActionCreate {
+		setStaticConsumerMetadata(cfg)
 	}
-	setStaticConsumerMetadata(cfg, oldCfg)
 
 	// If this is new consumer.
 	if ca == nil {
diff --git a/server/jetstream_versioning.go b/server/jetstream_versioning.go
index 82a80980..1192968b 100644
--- a/server/jetstream_versioning.go
+++ b/server/jetstream_versioning.go
@@ -25,12 +25,8 @@ const (
 )
 
 // setStaticStreamMetadata sets JetStream stream metadata, like the server version and API level.
-// Given:
-//   - cfg!=nil, prevCfg==nil		add stream: adds required metadata
-//   - cfg!=nil, prevCfg!=nil		update stream: required metadata is updated
-//
 // Any dynamic metadata is removed, it must not be stored and only be added for responses.
-func setStaticStreamMetadata(cfg *StreamConfig, _ *StreamConfig) {
+func setStaticStreamMetadata(cfg *StreamConfig) {
 	if cfg.Metadata == nil {
 		cfg.Metadata = make(map[string]string)
 	} else {
@@ -64,13 +60,38 @@ func setDynamicStreamMetadata(cfg *StreamConfig) *StreamConfig {
 	return &newCfg
 }
 
-// setStaticConsumerMetadata sets JetStream consumer metadata, like the server version and API level.
-// Given:
-//   - cfg!=nil, prevCfg==nil		add consumer: adds required metadata
-//   - cfg!=nil, prevCfg!=nil		update consumer: required metadata is updated
+// copyConsumerMetadata copies versioning fields from metadata of prevCfg into cfg.
+// Removes versioning fields if no previous metadata, updates if set, and removes fields if it doesn't exist in prevCfg.
+// Any dynamic metadata is removed, it must not be stored and only be added for responses.
 //
+// Note: useful when doing equality checks on cfg and prevCfg, but ignoring any versioning metadata differences.
+func copyStreamMetadata(cfg *StreamConfig, prevCfg *StreamConfig) {
+	if cfg.Metadata != nil {
+		deleteDynamicMetadata(cfg.Metadata)
+	}
+	setOrDeleteInStreamMetadata(cfg, prevCfg, JSRequiredLevelMetadataKey)
+}
+
+// setOrDeleteInConsumerMetadata sets field with key/value in metadata of cfg if set, deletes otherwise.
+func setOrDeleteInStreamMetadata(cfg *StreamConfig, prevCfg *StreamConfig, key string) {
+	if prevCfg != nil && prevCfg.Metadata != nil {
+		if value, ok := prevCfg.Metadata[key]; ok {
+			if cfg.Metadata == nil {
+				cfg.Metadata = make(map[string]string)
+			}
+			cfg.Metadata[key] = value
+			return
+		}
+	}
+	delete(cfg.Metadata, key)
+	if len(cfg.Metadata) == 0 {
+		cfg.Metadata = nil
+	}
+}
+
+// setStaticConsumerMetadata sets JetStream consumer metadata, like the server version and API level.
 // Any dynamic metadata is removed, it must not be stored and only be added for responses.
-func setStaticConsumerMetadata(cfg *ConsumerConfig, _ *ConsumerConfig) {
+func setStaticConsumerMetadata(cfg *ConsumerConfig) {
 	if cfg.Metadata == nil {
 		cfg.Metadata = make(map[string]string)
 	} else {
@@ -127,36 +148,27 @@ func setDynamicConsumerInfoMetadata(info *ConsumerInfo) *ConsumerInfo {
 // Any dynamic metadata is removed, it must not be stored and only be added for responses.
 //
 // Note: useful when doing equality checks on cfg and prevCfg, but ignoring any versioning metadata differences.
-// MUST be followed up with a call to setStaticConsumerMetadata to fix potentially lost metadata.
 func copyConsumerMetadata(cfg *ConsumerConfig, prevCfg *ConsumerConfig) {
 	if cfg.Metadata != nil {
 		deleteDynamicMetadata(cfg.Metadata)
 	}
+	setOrDeleteInConsumerMetadata(cfg, prevCfg, JSRequiredLevelMetadataKey)
+}
 
-	// Remove fields when no previous metadata.
-	if prevCfg == nil || prevCfg.Metadata == nil {
-		if cfg.Metadata != nil {
-			delete(cfg.Metadata, JSRequiredLevelMetadataKey)
-			if len(cfg.Metadata) == 0 {
-				cfg.Metadata = nil
+// setOrDeleteInConsumerMetadata sets field with key/value in metadata of cfg if set, deletes otherwise.
+func setOrDeleteInConsumerMetadata(cfg *ConsumerConfig, prevCfg *ConsumerConfig, key string) {
+	if prevCfg != nil && prevCfg.Metadata != nil {
+		if value, ok := prevCfg.Metadata[key]; ok {
+			if cfg.Metadata == nil {
+				cfg.Metadata = make(map[string]string)
 			}
+			cfg.Metadata[key] = value
+			return
 		}
-		return
 	}
-
-	// Set if exists, delete otherwise.
-	setOrDeleteInMetadata(cfg, prevCfg, JSRequiredLevelMetadataKey)
-}
-
-// setOrDeleteInMetadata sets field with key/value in metadata of cfg if set, deletes otherwise.
-func setOrDeleteInMetadata(cfg *ConsumerConfig, prevCfg *ConsumerConfig, key string) {
-	if value, ok := prevCfg.Metadata[key]; ok {
-		if cfg.Metadata == nil {
-			cfg.Metadata = make(map[string]string)
-		}
-		cfg.Metadata[key] = value
-	} else {
-		delete(cfg.Metadata, key)
+	delete(cfg.Metadata, key)
+	if len(cfg.Metadata) == 0 {
+		cfg.Metadata = nil
 	}
 }
 


Test patch:
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 655d036d..a29465da 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -7801,6 +7801,185 @@ func TestJetStreamClusterRecreateConsumerFromMetaSnapshot(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterUpgradeStreamVersioning(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	ml := c.leader()
+	sjs := ml.getJetStream()
+	rn := sjs.getMetaGroup()
+	acc, err := ml.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+
+	// Create stream config.
+	cfg, apiErr := ml.checkStreamCfg(&StreamConfig{Name: "TEST", Subjects: []string{"foo"}}, acc, false)
+	require_True(t, apiErr == nil)
+
+	// Create and propose stream assignment.
+	ci := &ClientInfo{Cluster: "R3S", Account: globalAccountName}
+	rg, perr := sjs.createGroupForStream(ci, &cfg)
+	require_True(t, perr == nil)
+	sa := &streamAssignment{Group: rg, Sync: syncSubjForStream(), Config: &cfg, Client: ci, Created: time.Now().UTC()}
+	require_NoError(t, rn.Propose(encodeAddStreamAssignment(sa)))
+
+	// Wait for the stream assignment to have gone through.
+	checkFor(t, 2*time.Second, 500*time.Millisecond, func() error {
+		sjs.mu.RLock()
+		defer sjs.mu.RUnlock()
+		if sjs.streamAssignment(globalAccountName, "TEST") == nil {
+			return errors.New("stream assignment does not exist yet")
+		}
+		return nil
+	})
+
+	for _, create := range []bool{true, false} {
+		title := "create"
+		if !create {
+			title = "update"
+		}
+		t.Run(title, func(t *testing.T) {
+			if create {
+				// Create on 2.11+ should be idempotent with previous create on 2.10-.
+				mcfg := &StreamConfig{}
+				setStaticStreamMetadata(mcfg)
+				si, err := js.AddStream(&nats.StreamConfig{
+					Name:     "TEST",
+					Subjects: []string{"foo"},
+					Metadata: setDynamicStreamMetadata(mcfg).Metadata,
+				})
+				require_NoError(t, err)
+				deleteDynamicMetadata(si.Config.Metadata)
+				require_Len(t, len(si.Config.Metadata), 0)
+			} else {
+				// Update populates the versioning metadata.
+				si, err := js.UpdateStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}})
+				require_NoError(t, err)
+				require_Len(t, len(si.Config.Metadata), 3)
+			}
+		})
+	}
+}
+
+func TestJetStreamClusterUpgradeConsumerVersioning(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Subjects:  []string{"foo"},
+		Retention: nats.LimitsPolicy,
+		Replicas:  1,
+	})
+	require_NoError(t, err)
+
+	ml := c.leader()
+	sjs := ml.getJetStream()
+
+	// Wait for the stream assignment to have gone through.
+	checkFor(t, 2*time.Second, 500*time.Millisecond, func() error {
+		sjs.mu.RLock()
+		defer sjs.mu.RUnlock()
+		if sjs.streamAssignment(globalAccountName, "TEST") == nil {
+			return errors.New("stream assignment does not exist yet")
+		}
+		return nil
+	})
+
+	rn := sjs.getMetaGroup()
+	sa := sjs.streamAssignment(globalAccountName, "TEST")
+	require_NotNil(t, sa)
+	acc, err := ml.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+
+	// Create consumer config.
+	cfg := &ConsumerConfig{Durable: "CONSUMER"}
+	streamCfg, ok := sjs.clusterStreamConfig(acc.Name, "TEST")
+	if !ok {
+		require_NoError(t, NewJSStreamNotFoundError())
+	}
+	selectedLimits, _, _, apiErr := acc.selectLimits(cfg.replicas(&streamCfg))
+	if apiErr != nil {
+		require_NoError(t, apiErr)
+	}
+	srvLim := &ml.getOpts().JetStreamLimits
+	apiErr = setConsumerConfigDefaults(cfg, &streamCfg, srvLim, selectedLimits, false)
+	if apiErr != nil {
+		require_NoError(t, apiErr)
+	}
+
+	// Create and propose consumer assignment.
+	ci := &ClientInfo{Cluster: "R3S", Account: globalAccountName}
+	rg := sjs.cluster.createGroupForConsumer(cfg, sa)
+	ca := &consumerAssignment{Group: rg, Stream: "TEST", Name: "CONSUMER", Config: cfg, Client: ci, Created: time.Now().UTC()}
+	require_NoError(t, rn.Propose(encodeAddConsumerAssignment(ca)))
+
+	// Wait for the consumer assignment to have gone through.
+	checkFor(t, 2*time.Second, 500*time.Millisecond, func() error {
+		sjs.mu.RLock()
+		defer sjs.mu.RUnlock()
+		if sjs.consumerAssignment(globalAccountName, "TEST", "CONSUMER") == nil {
+			return errors.New("consumer assignment does not exist yet")
+		}
+		return nil
+	})
+
+	for _, create := range []bool{true, false} {
+		title := "create"
+		if !create {
+			title = "update"
+		}
+		t.Run(title, func(t *testing.T) {
+			createConsumerRequest := func(obsReq CreateConsumerRequest) (*JSApiConsumerInfoResponse, error) {
+				req, err := json.Marshal(obsReq)
+				if err != nil {
+					return nil, err
+				}
+				msg, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", "CONSUMER"), req, time.Second)
+				if err != nil {
+					return nil, err
+				}
+				var resp JSApiConsumerInfoResponse
+				require_NoError(t, json.Unmarshal(msg.Data, &resp))
+				if resp.Error != nil {
+					return nil, resp.Error
+				}
+				return &resp, nil
+			}
+
+			if create {
+				// Create on 2.11+ should be idempotent with previous create on 2.10-.
+				ncfg := &ConsumerConfig{Durable: "CONSUMER"}
+				setStaticConsumerMetadata(ncfg)
+				obsReq := CreateConsumerRequest{
+					Stream: "TEST",
+					Config: *setDynamicConsumerMetadata(ncfg),
+					Action: ActionCreate,
+				}
+				resp, err := createConsumerRequest(obsReq)
+				require_NoError(t, err)
+				deleteDynamicMetadata(resp.Config.Metadata)
+				require_Len(t, len(resp.Config.Metadata), 0)
+			} else {
+				// Update populates the versioning metadata.
+				obsReq := CreateConsumerRequest{
+					Stream: "TEST",
+					Config: ConsumerConfig{Durable: "CONSUMER"},
+					Action: ActionUpdate,
+				}
+				resp, err := createConsumerRequest(obsReq)
+				require_NoError(t, err)
+				require_Len(t, len(resp.Config.Metadata), 3)
+			}
+		})
+	}
+}
+
 //
 // DO NOT ADD NEW TESTS IN THIS FILE (unless to balance test times)
 // Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.
diff --git a/server/jetstream_versioning_test.go b/server/jetstream_versioning_test.go
index a2723c1a..6dcd73b4 100644
--- a/server/jetstream_versioning_test.go
+++ b/server/jetstream_versioning_test.go
@@ -45,48 +45,31 @@ func TestJetStreamSetStaticStreamMetadata(t *testing.T) {
 	for _, test := range []struct {
 		desc             string
 		cfg              *StreamConfig
-		prev             *StreamConfig
 		expectedMetadata map[string]string
 	}{
 		{
-			desc:             "create",
+			desc:             "empty",
 			cfg:              &StreamConfig{},
-			prev:             nil,
 			expectedMetadata: metadataAtLevel("0"),
 		},
 		{
-			desc:             "create/overwrite-user-provided",
+			desc:             "overwrite-user-provided",
 			cfg:              &StreamConfig{Metadata: metadataPrevious()},
-			prev:             nil,
 			expectedMetadata: metadataAtLevel("0"),
 		},
 		{
-			desc:             "update",
-			cfg:              &StreamConfig{},
-			prev:             &StreamConfig{Metadata: metadataPrevious()},
-			expectedMetadata: metadataAtLevel("0"),
-		},
-		{
-			desc:             "update/empty-prev-metadata",
-			cfg:              &StreamConfig{},
-			prev:             &StreamConfig{},
-			expectedMetadata: metadataAtLevel("0"),
-		},
-		{
-			desc:             "update/empty-prev-metadata/delete-user-provided",
+			desc:             "empty-prev-metadata/delete-user-provided",
 			cfg:              &StreamConfig{Metadata: metadataPrevious()},
-			prev:             &StreamConfig{},
 			expectedMetadata: metadataAtLevel("0"),
 		},
 		{
-			desc:             "create/AllowMsgTTL",
+			desc:             "AllowMsgTTL",
 			cfg:              &StreamConfig{AllowMsgTTL: true},
-			prev:             nil,
 			expectedMetadata: metadataAtLevel("1"),
 		},
 	} {
 		t.Run(test.desc, func(t *testing.T) {
-			setStaticStreamMetadata(test.cfg, test.prev)
+			setStaticStreamMetadata(test.cfg)
 			require_Equal(t, test.cfg.Metadata[JSRequiredLevelMetadataKey], test.expectedMetadata[JSRequiredLevelMetadataKey])
 		})
 	}
@@ -101,12 +84,7 @@ func TestJetStreamSetStaticStreamMetadataRemoveDynamicFields(t *testing.T) {
 	}
 
 	cfg := StreamConfig{Metadata: dynamicMetadata()}
-	setStaticStreamMetadata(&cfg, nil)
-	require_True(t, reflect.DeepEqual(cfg.Metadata, metadataAtLevel("0")))
-
-	cfg = StreamConfig{Metadata: dynamicMetadata()}
-	prevCfg := StreamConfig{Metadata: metadataAtLevel("0")}
-	setStaticStreamMetadata(&cfg, &prevCfg)
+	setStaticStreamMetadata(&cfg)
 	require_True(t, reflect.DeepEqual(cfg.Metadata, metadataAtLevel("0")))
 }
 
@@ -122,78 +100,112 @@ func TestJetStreamSetDynamicStreamMetadata(t *testing.T) {
 	require_True(t, reflect.DeepEqual(newCfg.Metadata, metadata))
 }
 
+func TestJetStreamCopyStreamMetadata(t *testing.T) {
+	for _, test := range []struct {
+		desc string
+		cfg  *StreamConfig
+		prev *StreamConfig
+	}{
+		{
+			desc: "no-previous-ignore",
+			cfg:  &StreamConfig{Metadata: metadataAtLevel("-1")},
+			prev: nil,
+		},
+		{
+			desc: "nil-previous-metadata-ignore",
+			cfg:  &StreamConfig{Metadata: metadataAtLevel("-1")},
+			prev: &StreamConfig{Metadata: nil},
+		},
+		{
+			desc: "nil-current-metadata-ignore",
+			cfg:  &StreamConfig{Metadata: nil},
+			prev: &StreamConfig{Metadata: metadataPrevious()},
+		},
+		{
+			desc: "copy-previous",
+			cfg:  &StreamConfig{Metadata: metadataAtLevel("-1")},
+			prev: &StreamConfig{Metadata: metadataPrevious()},
+		},
+		{
+			desc: "delete-missing-fields",
+			cfg:  &StreamConfig{Metadata: metadataAtLevel("-1")},
+			prev: &StreamConfig{Metadata: make(map[string]string)},
+		},
+	} {
+		t.Run(test.desc, func(t *testing.T) {
+			copyStreamMetadata(test.cfg, test.prev)
+
+			var expectedMetadata map[string]string
+			if test.prev != nil {
+				expectedMetadata = test.prev.Metadata
+			}
+
+			value, ok := expectedMetadata[JSRequiredLevelMetadataKey]
+			if ok {
+				require_Equal(t, test.cfg.Metadata[JSRequiredLevelMetadataKey], value)
+			} else {
+				// Key shouldn't exist.
+				_, ok = test.cfg.Metadata[JSRequiredLevelMetadataKey]
+				require_False(t, ok)
+			}
+		})
+	}
+}
+
+func TestJetStreamCopyStreamMetadataRemoveDynamicFields(t *testing.T) {
+	dynamicMetadata := func() map[string]string {
+		return map[string]string{
+			JSServerVersionMetadataKey: "dynamic-version",
+			JSServerLevelMetadataKey:   "dynamic-version",
+		}
+	}
+
+	cfg := StreamConfig{Metadata: dynamicMetadata()}
+	copyStreamMetadata(&cfg, nil)
+	require_Equal(t, len(cfg.Metadata), 0)
+
+	cfg = StreamConfig{Metadata: dynamicMetadata()}
+	prevCfg := StreamConfig{Metadata: metadataAtLevel("0")}
+	copyStreamMetadata(&cfg, &prevCfg)
+	require_True(t, reflect.DeepEqual(cfg.Metadata, metadataAtLevel("0")))
+}
+
 func TestJetStreamSetStaticConsumerMetadata(t *testing.T) {
 	pauseUntil := time.Unix(0, 0)
 	pauseUntilZero := time.Time{}
 	for _, test := range []struct {
 		desc             string
 		cfg              *ConsumerConfig
-		prev             *ConsumerConfig
 		expectedMetadata map[string]string
 	}{
 		{
-			desc:             "create",
+			desc:             "empty",
 			cfg:              &ConsumerConfig{},
-			prev:             nil,
 			expectedMetadata: metadataAtLevel("0"),
 		},
 		{
-			desc:             "create/PauseUntil/zero",
-			cfg:              &ConsumerConfig{PauseUntil: &pauseUntilZero},
-			prev:             nil,
-			expectedMetadata: metadataAtLevel("0"),
-		},
-		{
-			desc:             "create/PauseUntil",
-			cfg:              &ConsumerConfig{PauseUntil: &pauseUntil},
-			prev:             nil,
-			expectedMetadata: metadataAtLevel("1"),
-		},
-		{
-			desc:             "create/overwrite-user-provided",
+			desc:             "overwrite-user-provided",
 			cfg:              &ConsumerConfig{Metadata: metadataPrevious()},
-			prev:             nil,
 			expectedMetadata: metadataAtLevel("0"),
 		},
 		{
-			desc:             "update",
-			cfg:              &ConsumerConfig{},
-			prev:             &ConsumerConfig{Metadata: metadataPrevious()},
-			expectedMetadata: metadataAtLevel("0"),
-		},
-		{
-			desc:             "create/PauseUntil/zero",
+			desc:             "PauseUntil/zero",
 			cfg:              &ConsumerConfig{PauseUntil: &pauseUntilZero},
-			prev:             &ConsumerConfig{Metadata: metadataPrevious()},
 			expectedMetadata: metadataAtLevel("0"),
 		},
 		{
-			desc:             "update/PauseUntil",
+			desc:             "PauseUntil",
 			cfg:              &ConsumerConfig{PauseUntil: &pauseUntil},
-			prev:             &ConsumerConfig{Metadata: metadataPrevious()},
 			expectedMetadata: metadataAtLevel("1"),
 		},
 		{
-			desc:             "create/Pinned",
+			desc:             "Pinned",
 			cfg:              &ConsumerConfig{PriorityPolicy: PriorityPinnedClient, PriorityGroups: []string{"a"}},
-			prev:             &ConsumerConfig{Metadata: metadataPrevious()},
 			expectedMetadata: metadataAtLevel("1"),
 		},
-		{
-			desc:             "update/empty-prev-metadata",
-			cfg:              &ConsumerConfig{},
-			prev:             &ConsumerConfig{},
-			expectedMetadata: metadataAtLevel("0"),
-		},
-		{
-			desc:             "update/empty-prev-metadata/delete-user-provided",
-			cfg:              &ConsumerConfig{Metadata: metadataPrevious()},
-			prev:             &ConsumerConfig{},
-			expectedMetadata: metadataAtLevel("0"),
-		},
 	} {
 		t.Run(test.desc, func(t *testing.T) {
-			setStaticConsumerMetadata(test.cfg, test.prev)
+			setStaticConsumerMetadata(test.cfg)
 			require_Equal(t, test.cfg.Metadata[JSRequiredLevelMetadataKey], test.expectedMetadata[JSRequiredLevelMetadataKey])
 		})
 	}
@@ -208,12 +220,7 @@ func TestJetStreamSetStaticConsumerMetadataRemoveDynamicFields(t *testing.T) {
 	}
 
 	cfg := ConsumerConfig{Metadata: dynamicMetadata()}
-	setStaticConsumerMetadata(&cfg, nil)
-	require_True(t, reflect.DeepEqual(cfg.Metadata, metadataAtLevel("0")))
-
-	cfg = ConsumerConfig{Metadata: dynamicMetadata()}
-	prevCfg := ConsumerConfig{Metadata: metadataAtLevel("0")}
-	setStaticConsumerMetadata(&cfg, &prevCfg)
+	setStaticConsumerMetadata(&cfg)
 	require_True(t, reflect.DeepEqual(cfg.Metadata, metadataAtLevel("0")))
 }