Repository: nats-io/nats-server
Pull request #3824 (2023-01-27 20:24:26)

Check if connection name was already set when storing it
Signed-off-by: Waldemar Quevedo <wally@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 1b9c5873..5bd07472 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -619,7 +619,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 		if err != nil {
 			return err
 		}
-		tr, err := NewSubjectTransform(src, d.Subject)
+		tr, err := newSubjectTransform(src, d.Subject)
 		if err != nil {
 			return err
 		}
@@ -650,7 +650,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 				// We need to make the appropriate markers for the wildcards etc.
 				dest = transformTokenize(dest)
 			}
-			tr, err := NewSubjectTransform(src, dest)
+			tr, err := newSubjectTransform(src, dest)
 			if err != nil {
 				return nil, err
 			}
@@ -1828,7 +1828,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		} else {
 			to, _ = transformUntokenize(to)
 			// Create a transform. Do so in reverse such that $ symbols only exist in to
-			if tr, err = NewSubjectTransform(to, transformTokenize(from)); err != nil {
+			if tr, err = newSubjectTransform(to, transformTokenize(from)); err != nil {
 				a.mu.Unlock()
 				return nil, fmt.Errorf("failed to create mapping transform for service import subject %q to %q: %v",
 					from, to, err)
@@ -2375,7 +2375,7 @@ func (a *Account) AddMappedStreamImportWithClaim(account *Account, from, to stri
 			usePub = true
 		} else {
 			// Create a transform
-			if tr, err = NewSubjectTransform(from, transformTokenize(to)); err != nil {
+			if tr, err = newSubjectTransform(from, transformTokenize(to)); err != nil {
 				return fmt.Errorf("failed to create mapping transform for stream import subject %q to %q: %v",
 					from, to, err)
 			}
diff --git a/server/client.go b/server/client.go
index 0c146114..a02a8072 100644
--- a/server/client.go
+++ b/server/client.go
@@ -1883,7 +1883,7 @@ func (c *client) processConnect(arg []byte) error {
 			}
 		}
 		if ncs != _EMPTY_ {
-			c.ncs.Store(fmt.Sprintf("%s - %q", c, ncs))
+			c.ncs.CompareAndSwap(nil, fmt.Sprintf("%s - %q", c, ncs))
 		}
 	}
 
diff --git a/server/consumer.go b/server/consumer.go
index ac84ab2b..0ee07294 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -38,9 +38,6 @@ const (
 	JSPullRequestPendingBytes = "Nats-Pending-Bytes"
 )
 
-// Headers sent when batch size was completed, but there were remaining bytes.
-const JsPullRequestRemainingBytesT = "NATS/1.0 409 Batch Completed\r\n%s: %d\r\n%s: %d\r\n\r\n"
-
 type ConsumerInfo struct {
 	Stream         string          `json:"stream_name"`
 	Name           string          `json:"name"`
@@ -97,9 +94,6 @@ type ConsumerConfig struct {
 
 	// Don't add to general clients.
 	Direct bool `json:"direct,omitempty"`
-
-	// Metadata is additional metadata for the Consumer.
-	Metadata map[string]string `json:"metadata,omitempty"`
 }
 
 // SequenceInfo has both the consumer and the stream sequence and last activity.
@@ -551,14 +545,6 @@ func checkConsumerCfg(
 		}
 	}
 
-	var metadataLen int
-	for k, v := range config.Metadata {
-		metadataLen += len(k) + len(v)
-	}
-	if metadataLen > JSMaxMetadataLen {
-		return NewJSConsumerMetadataLengthError(fmt.Sprintf("%dKB", JSMaxMetadataLen/1024))
-	}
-
 	return nil
 }
 
@@ -3229,7 +3215,6 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			ackReply string
 			delay    time.Duration
 			sz       int
-			wrn, wrb int
 		)
 		o.mu.Lock()
 		// consumer is closed when mset is set to nil.
@@ -3283,7 +3268,6 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		if o.isPushMode() {
 			dsubj = o.dsubj
 		} else if wr := o.nextWaiting(sz); wr != nil {
-			wrn, wrb = wr.n, wr.b
 			dsubj = wr.reply
 			if done := wr.recycleIfDone(); done && o.node != nil {
 				o.removeClusterPendingRequest(dsubj)
@@ -3337,10 +3321,6 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		// Do actual delivery.
 		o.deliverMsg(dsubj, ackReply, pmsg, dc, rp)
 
-		// If given request fulfilled batch size, but there are still pending bytes, send information about it.
-		if wrn <= 0 && wrb > 0 {
-			o.outq.send(newJSPubMsg(dsubj, _EMPTY_, _EMPTY_, []byte(fmt.Sprintf(JsPullRequestRemainingBytesT, JSPullRequestPendingMsgs, wrn, JSPullRequestPendingBytes, wrb)), nil, nil, 0))
-		}
 		// Reset our idle heartbeat timer if set.
 		if hb != nil {
 			hb.Reset(hbd)
diff --git a/server/errors.json b/server/errors.json
index 7b048856..e48eea78 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1328,15 +1328,5 @@
     "help": "",
     "url": "",
     "deprecates": ""
-  },
-  {
-    "constant": "JSConsumerMetadataLengthErrF",
-    "code": 400,
-    "error_code": 10135,
-    "description": "consumer metadata exceeds maximum size of {limit}",
-    "comment": "",
-    "help": "",
-    "url": "",
-    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 6129531c..8a49ecad 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -330,10 +330,6 @@ func generateJSMappingTable(domain string) map[string]string {
 // JSMaxDescription is the maximum description length for streams and consumers.
 const JSMaxDescriptionLen = 4 * 1024
 
-// JSMaxMetadataLen is the maximum length for streams an consumers metadata map.
-// It's calculated by summing length of all keys an values.
-const JSMaxMetadataLen = 128 * 1024
-
 // JSMaxNameLen is the maximum name lengths for streams, consumers and templates.
 // Picked 255 as it seems to be a widely used file name limit
 const JSMaxNameLen = 255
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 0a7a829c..58bdfe30 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -131,9 +131,6 @@ const (
 	// JSConsumerMaxWaitingNegativeErr consumer max waiting needs to be positive
 	JSConsumerMaxWaitingNegativeErr ErrorIdentifier = 10087
 
-	// JSConsumerMetadataLengthErrF consumer metadata exceeds maximum size of {limit}
-	JSConsumerMetadataLengthErrF ErrorIdentifier = 10135
-
 	// JSConsumerNameContainsPathSeparatorsErr Consumer name can not contain path separators
 	JSConsumerNameContainsPathSeparatorsErr ErrorIdentifier = 10127
 
@@ -452,7 +449,6 @@ var (
 		JSConsumerMaxRequestBatchNegativeErr:       {Code: 400, ErrCode: 10114, Description: "consumer max request batch needs to be > 0"},
 		JSConsumerMaxRequestExpiresToSmall:         {Code: 400, ErrCode: 10115, Description: "consumer max request expires needs to be >= 1ms"},
 		JSConsumerMaxWaitingNegativeErr:            {Code: 400, ErrCode: 10087, Description: "consumer max waiting needs to be positive"},
-		JSConsumerMetadataLengthErrF:               {Code: 400, ErrCode: 10135, Description: "consumer metadata exceeds maximum size of {limit}"},
 		JSConsumerNameContainsPathSeparatorsErr:    {Code: 400, ErrCode: 10127, Description: "Consumer name can not contain path separators"},
 		JSConsumerNameExistErr:                     {Code: 400, ErrCode: 10013, Description: "consumer name already in use"},
 		JSConsumerNameTooLongErrF:                  {Code: 400, ErrCode: 10102, Description: "consumer name is too long, maximum allowed is {max}"},
@@ -1031,22 +1027,6 @@ func NewJSConsumerMaxWaitingNegativeError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSConsumerMaxWaitingNegativeErr]
 }
 
-// NewJSConsumerMetadataLengthError creates a new JSConsumerMetadataLengthErrF error: "consumer metadata exceeds maximum size of {limit}"
-func NewJSConsumerMetadataLengthError(limit interface{}, opts ...ErrorOption) *ApiError {
-	eopts := parseOpts(opts)
-	if ae, ok := eopts.err.(*ApiError); ok {
-		return ae
-	}
-
-	e := ApiErrors[JSConsumerMetadataLengthErrF]
-	args := e.toReplacerArgs([]interface{}{"{limit}", limit})
-	return &ApiError{
-		Code:        e.Code,
-		ErrCode:     e.ErrCode,
-		Description: strings.NewReplacer(args...).Replace(e.Description),
-	}
-}
-
 // NewJSConsumerNameContainsPathSeparatorsError creates a new JSConsumerNameContainsPathSeparatorsErr error: "Consumer name can not contain path separators"
 func NewJSConsumerNameContainsPathSeparatorsError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/pse/pse_dragonfly.go b/server/pse/pse_dragonfly.go
deleted file mode 100644
index 55a917ab..00000000
--- a/server/pse/pse_dragonfly.go
+++ /dev/null
@@ -1,36 +0,0 @@
-// Copyright 2015-2023 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-//
-// Copied from pse_openbsd.go
-
-package pse
-
-import (
-	"fmt"
-	"os"
-	"os/exec"
-)
-
-// ProcUsage returns CPU usage
-func ProcUsage(pcpu *float64, rss, vss *int64) error {
-	pidStr := fmt.Sprintf("%d", os.Getpid())
-	out, err := exec.Command("ps", "o", "pcpu=,rss=,vsz=", "-p", pidStr).Output()
-	if err != nil {
-		*rss, *vss = -1, -1
-		return fmt.Errorf("ps call failed:%v", err)
-	}
-	fmt.Sscanf(string(out), "%f %d %d", pcpu, rss, vss)
-	*rss *= 1024 // 1k blocks, want bytes.
-	*vss *= 1024 // 1k blocks, want bytes.
-	return nil
-}
diff --git a/server/stream.go b/server/stream.go
index c3f906cf..5d4226d1 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -83,9 +83,6 @@ type StreamConfig struct {
 	// AllowRollup allows messages to be placed into the system and purge
 	// all older messages using a special msg header.
 	AllowRollup bool `json:"allow_rollup_hdrs"`
-
-	// Metadata is additional metadata for the Stream.
-	Metadata map[string]string `json:"metadata,omitempty"`
 }
 
 // SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received
@@ -430,7 +427,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 			ssi.setIndexName(i)
 			// check the transform, if any, is valid
 			if ssi.SubjectTransformDest != _EMPTY_ {
-				if _, err = NewSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest); err != nil {
+				if _, err = newSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest); err != nil {
 					jsa.mu.Unlock()
 					return nil, fmt.Errorf("subject transform for the source not valid %w", err)
 				}
@@ -483,7 +480,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 
 	// Check for input subject transform
 	if cfg.SubjectTransform != nil {
-		tr, err := NewSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
+		tr, err := newSubjectTransform(cfg.SubjectTransform.Source, cfg.SubjectTransform.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
 			return nil, fmt.Errorf("stream input subject transform not valid %w", err)
@@ -493,7 +490,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 
 	// Check for RePublish.
 	if cfg.RePublish != nil {
-		tr, err := NewSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
+		tr, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
 			return nil, fmt.Errorf("stream republish transform not valid %w", err)
@@ -982,14 +979,6 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream description is too long, maximum allowed is %d", JSMaxDescriptionLen))
 	}
 
-	var metadataLen int
-	for k, v := range config.Metadata {
-		metadataLen += len(k) + len(v)
-	}
-	if metadataLen > JSMaxMetadataLen {
-		return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream metadata exceeds maximum size of %d bytes", JSMaxMetadataLen))
-	}
-
 	cfg := *config
 
 	// Make file the default.
@@ -1291,7 +1280,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		if formsCycle {
 			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish destination forms a cycle"))
 		}
-		if _, err := NewSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
+		if _, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination); err != nil {
 			return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for republish not valid"))
 		}
 	}
@@ -1522,7 +1511,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 					// Check for transform.
 					if s.SubjectTransformDest != _EMPTY_ {
 						var err error
-						if si.tr, err = NewSubjectTransform(s.FilterSubject, s.SubjectTransformDest); err != nil {
+						if si.tr, err = newSubjectTransform(s.FilterSubject, s.SubjectTransformDest); err != nil {
 							return err
 						}
 					}
@@ -1579,7 +1568,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 		if cfg.RePublish.Source == _EMPTY_ {
 			cfg.RePublish.Source = fwcs
 		}
-		tr, err := NewSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
+		tr, err := newSubjectTransform(cfg.RePublish.Source, cfg.RePublish.Destination)
 		if err != nil {
 			jsa.mu.Unlock()
 			return fmt.Errorf("stream configuration for republish not valid")
@@ -2879,7 +2868,7 @@ func (mset *stream) startingSequenceForSources() {
 		si := &sourceInfo{name: ssi.Name, iname: ssi.iname, sf: ssi.FilterSubject}
 		// Check for transform.
 		if ssi.SubjectTransformDest != _EMPTY_ {
-			si.tr, _ = NewSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest) // can not return an error because validated in AddStream
+			si.tr, _ = newSubjectTransform(ssi.FilterSubject, ssi.SubjectTransformDest) // can not return an error because validated in AddStream
 		}
 
 		mset.sources[ssi.iname] = si
diff --git a/server/subject_transform.go b/server/subject_transform.go
index a8c20d1e..2dd553c4 100644
--- a/server/subject_transform.go
+++ b/server/subject_transform.go
@@ -56,17 +56,17 @@ type subjectTransform struct {
 	dtokmfstringargs     []string // destination token mapping function string arguments
 }
 
-// SubjectTransformer transforms subjects using mappings
+// SubjectTransform transforms subjects using mappings
 //
 // This API is not part of the public API and not subject to SemVer protections
-type SubjectTransformer interface {
+type SubjectTransform interface {
 	// TODO(dlc) - We could add in client here to allow for things like foo -> foo.$ACCOUNT
 	Match(string) (string, error)
 	TransformSubject(subject string) string
 	TransformTokenizedSubject(tokens []string) string
 }
 
-func NewSubjectTransform(src, dest string) (*subjectTransform, error) {
+func newSubjectTransform(src, dest string) (*subjectTransform, error) {
 	// No source given is equivalent to the source being ">"
 	if src == _EMPTY_ {
 		src = fwcs
@@ -493,7 +493,7 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string {
 // Reverse a subjectTransform.
 func (tr *subjectTransform) reverse() *subjectTransform {
 	if len(tr.dtokmftokindexesargs) == 0 {
-		rtr, _ := NewSubjectTransform(tr.dest, tr.src)
+		rtr, _ := newSubjectTransform(tr.dest, tr.src)
 		return rtr
 	}
 	// If we are here we need to dynamically get the correct reverse
@@ -513,6 +513,6 @@ func (tr *subjectTransform) reverse() *subjectTransform {
 		}
 	}
 	ndest := strings.Join(nda, tsep)
-	rtr, _ := NewSubjectTransform(nsrc, ndest)
+	rtr, _ := newSubjectTransform(nsrc, ndest)
 	return rtr
 }


Test patch:
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 6c7c5d95..3463a32b 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -16594,39 +16594,25 @@ func TestJetStreamPullMaxBytes(t *testing.T) {
 	req = &JSApiConsumerGetNextRequest{Batch: 1, MaxBytes: 10_000_000, NoWait: true}
 	jreq, _ = json.Marshal(req)
 	nc.PublishRequest(subj, reply, jreq)
-	// we expect two messages, as the second one should be `Batch Completed` status.
-	checkSubsPending(t, sub, 2)
+	checkSubsPending(t, sub, 1)
 
-	// first one is message from the stream.
 	m, err = sub.NextMsg(time.Second)
 	require_NoError(t, err)
 	require_True(t, len(m.Data) == dsz)
 	require_True(t, len(m.Header) == 0)
-	// second one is the status.
-	m, err = sub.NextMsg(time.Second)
-	require_NoError(t, err)
-	if v := m.Header.Get("Description"); v != "Batch Completed" {
-		t.Fatalf("Expected Batch Completed, got: %s", v)
-	}
 	checkSubsPending(t, sub, 0)
 
 	// Same but with batch > 1
 	req = &JSApiConsumerGetNextRequest{Batch: 5, MaxBytes: 10_000_000, NoWait: true}
 	jreq, _ = json.Marshal(req)
 	nc.PublishRequest(subj, reply, jreq)
-	// 6, not 5, as 6th is the status.
-	checkSubsPending(t, sub, 6)
+	checkSubsPending(t, sub, 5)
 	for i := 0; i < 5; i++ {
 		m, err = sub.NextMsg(time.Second)
 		require_NoError(t, err)
 		require_True(t, len(m.Data) == dsz)
 		require_True(t, len(m.Header) == 0)
 	}
-	m, err = sub.NextMsg(time.Second)
-	require_NoError(t, err)
-	if v := m.Header.Get("Description"); v != "Batch Completed" {
-		t.Fatalf("Expected Batch Completed, got: %s", v)
-	}
 	checkSubsPending(t, sub, 0)
 
 	// Now ask for large batch but make sure we are limited by batch size.
@@ -19133,108 +19119,3 @@ func TestJetStreamMsgBlkFailOnKernelFault(t *testing.T) {
 			friendlyBytes(si.Config.MaxBytes), friendlyBytes(int64(si.State.Bytes)))
 	}
 }
-
-func TestJetStreamPullConsumerBatchCompleted(t *testing.T) {
-
-	s := RunBasicJetStreamServer(t)
-	defer s.Shutdown()
-
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"foo"},
-	})
-	require_NoError(t, err)
-
-	msgSize := 128
-	msg := make([]byte, msgSize)
-	rand.Read(msg)
-
-	for i := 0; i < 10; i++ {
-		_, err := js.Publish("foo", msg)
-		require_NoError(t, err)
-	}
-
-	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
-		Durable:   "dur",
-		AckPolicy: nats.AckExplicitPolicy,
-	})
-	require_NoError(t, err)
-
-	req := JSApiConsumerGetNextRequest{Batch: 0, MaxBytes: 1024, Expires: 250 * time.Millisecond}
-
-	reqb, _ := json.Marshal(req)
-	sub := natsSubSync(t, nc, nats.NewInbox())
-	err = nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.dur", sub.Subject, reqb)
-	require_NoError(t, err)
-
-	// Expect first message to arrive normally.
-	_, err = sub.NextMsg(time.Second * 1)
-	require_NoError(t, err)
-
-	// Second message should be info that batch is complete, but there were pending bytes.
-	pullMsg, err := sub.NextMsg(time.Second * 1)
-	require_NoError(t, err)
-
-	if v := pullMsg.Header.Get("Status"); v != "409" {
-		t.Fatalf("Expected 409, got: %s", v)
-	}
-	if v := pullMsg.Header.Get("Description"); v != "Batch Completed" {
-		t.Fatalf("Expected Batch Completed, got: %s", v)
-	}
-
-}
-
-func TestJetStreamConsumerAndStreamMetadata(t *testing.T) {
-	s := RunBasicJetStreamServer(t)
-	defer s.Shutdown()
-
-	metadata := map[string]string{"key": "value", "_nats_created_version": "2.9.11"}
-	acc := s.GlobalAccount()
-
-	// Check stream's first.
-	mset, err := acc.addStream(&StreamConfig{Name: "foo", Metadata: metadata})
-	if err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	if cfg := mset.config(); !reflect.DeepEqual(metadata, cfg.Metadata) {
-		t.Fatalf("Expected a metadata of %q, got %q", metadata, cfg.Metadata)
-	}
-
-	// Now consumer
-	o, err := mset.addConsumer(&ConsumerConfig{
-		Metadata:       metadata,
-		DeliverSubject: "to",
-		AckPolicy:      AckNone})
-	if err != nil {
-		t.Fatalf("Unexpected error adding consumer: %v", err)
-	}
-	if cfg := o.config(); !reflect.DeepEqual(metadata, cfg.Metadata) {
-		t.Fatalf("Expected a metadata of %q, got %q", metadata, cfg.Metadata)
-	}
-
-	// Test max.
-	data := make([]byte, JSMaxMetadataLen/100)
-	rand.Read(data)
-	bigValue := base64.StdEncoding.EncodeToString(data)
-
-	bigMetadata := make(map[string]string, 101)
-	for i := 0; i < 101; i++ {
-		bigMetadata[fmt.Sprintf("key%d", i)] = bigValue
-	}
-
-	_, err = acc.addStream(&StreamConfig{Name: "bar", Metadata: bigMetadata})
-	if err == nil || !strings.Contains(err.Error(), "stream metadata exceeds") {
-		t.Fatalf("Expected an error but got none")
-	}
-
-	_, err = mset.addConsumer(&ConsumerConfig{
-		Metadata:       bigMetadata,
-		DeliverSubject: "to",
-		AckPolicy:      AckNone})
-	if err == nil || !strings.Contains(err.Error(), "consumer metadata exceeds") {
-		t.Fatalf("Expected an error but got none")
-	}
-}
diff --git a/server/subject_transform_test.go b/server/subject_transform_test.go
index 3cb7a0ad..8ac9b6ca 100644
--- a/server/subject_transform_test.go
+++ b/server/subject_transform_test.go
@@ -76,7 +76,7 @@ func TestPlaceHolderIndex(t *testing.T) {
 func TestSubjectTransforms(t *testing.T) {
 	shouldErr := func(src, dest string) {
 		t.Helper()
-		if _, err := NewSubjectTransform(src, dest); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) {
+		if _, err := newSubjectTransform(src, dest); err != ErrBadSubject && !errors.Is(err, ErrInvalidMappingDestination) {
 			t.Fatalf("Did not get an error for src=%q and dest=%q", src, dest)
 		}
 	}
@@ -107,7 +107,7 @@ func TestSubjectTransforms(t *testing.T) {
 
 	shouldBeOK := func(src, dest string) *subjectTransform {
 		t.Helper()
-		tr, err := NewSubjectTransform(src, dest)
+		tr, err := newSubjectTransform(src, dest)
 		if err != nil {
 			t.Fatalf("Got an error %v for src=%q and dest=%q", err, src, dest)
 		}