Repository: nats-io/nats.go
Pull request #1277 (2023-05-30 12:28:08)

[CHANGED] More verbose endpoint info on INFO requests
Addresses https://github.com/nats-io/nats-architecture-and-design/issues/220


Patch:
diff --git a/bench/bench.go b/bench/bench.go
index 58fc457..5490a19 100644
--- a/bench/bench.go
+++ b/bench/bench.go
@@ -215,6 +215,9 @@ func (sg *SampleGroup) MaxRate() int64 {
 
 // AvgRate returns the average of all the message rates in the SampleGroup
 func (sg *SampleGroup) AvgRate() int64 {
+	if !sg.HasSamples() {
+		return 0
+	}
 	sum := uint64(0)
 	for _, s := range sg.Samples {
 		sum += uint64(s.Rate())
@@ -224,6 +227,9 @@ func (sg *SampleGroup) AvgRate() int64 {
 
 // StdDev returns the standard deviation the message rates in the SampleGroup
 func (sg *SampleGroup) StdDev() float64 {
+	if !sg.HasSamples() {
+		return 0
+	}
 	avg := float64(sg.AvgRate())
 	sum := float64(0)
 	for _, c := range sg.Samples {
diff --git a/jetstream/README.md b/jetstream/README.md
index 9eb3b63..08f4ca8 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -476,7 +476,7 @@ in the buffer.
 
 ```go
 // a maximum of 10 messages or 1024 bytes will be stored in memory (whichever is encountered first)
-iter, _ := cons.Messages(WithMessagesMaxMessages(10), WithMessagesMaxBytes(1024))
+iter, _ := cons.Messages(jetstream.PullMaxMessages(10), jetstream.PullMaxBytes(1024))
 ```
 
 `Messages()` exposes the following options:
diff --git a/jetstream/consumer_config.go b/jetstream/consumer_config.go
index 713daf8..750e83c 100644
--- a/jetstream/consumer_config.go
+++ b/jetstream/consumer_config.go
@@ -190,18 +190,15 @@ func (p DeliverPolicy) String() string {
 }
 
 const (
-	// AckNonePolicy requires no acks for delivered messages.
-	AckNonePolicy AckPolicy = iota
+	// AckExplicitPolicy requires ack or nack for all messages.
+	AckExplicitPolicy AckPolicy = iota
 
 	// AckAllPolicy when acking a sequence number, this implicitly acks all
 	// sequences below this one as well.
 	AckAllPolicy
 
-	// AckExplicitPolicy requires ack or nack for all messages.
-	AckExplicitPolicy
-
-	// For configuration mismatch check
-	ackPolicyNotSet = 99
+	// AckNonePolicy requires no acks for delivered messages.
+	AckNonePolicy
 )
 
 func (p *AckPolicy) UnmarshalJSON(data []byte) error {
@@ -227,7 +224,7 @@ func (p AckPolicy) MarshalJSON() ([]byte, error) {
 	case AckExplicitPolicy:
 		return json.Marshal("explicit")
 	}
-	return nil, fmt.Errorf("nats: unknown acknowlegement policy %v", p)
+	return nil, fmt.Errorf("nats: unknown acknowledgement policy %v", p)
 }
 
 func (p AckPolicy) String() string {
@@ -238,8 +235,6 @@ func (p AckPolicy) String() string {
 		return "AckAll"
 	case AckExplicitPolicy:
 		return "AckExplicit"
-	case ackPolicyNotSet:
-		return "Not Initialized"
 	}
 	return "Unknown AckPolicy"
 }
diff --git a/jetstream/pull.go b/jetstream/pull.go
index acea006..b9cef00 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -185,16 +185,21 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 			return
 		}
 		defer func() {
-			if sub.pending.msgCount < consumeOpts.ThresholdMessages ||
-				(sub.pending.byteCount < consumeOpts.ThresholdBytes && sub.consumeOpts.MaxBytes != 0) &&
+			sub.Lock()
+			pending := sub.pending
+			sub.Unlock()
+
+			if pending.msgCount < consumeOpts.ThresholdMessages ||
+				(pending.byteCount < consumeOpts.ThresholdBytes && sub.consumeOpts.MaxBytes != 0) &&
 					atomic.LoadUint32(&sub.fetchInProgress) == 1 {
 
 				sub.fetchNext <- &pullRequest{
 					Expires:   sub.consumeOpts.Expires,
-					Batch:     sub.consumeOpts.MaxMessages - sub.pending.msgCount,
-					MaxBytes:  sub.consumeOpts.MaxBytes - sub.pending.byteCount,
+					Batch:     sub.consumeOpts.MaxMessages - pending.msgCount,
+					MaxBytes:  sub.consumeOpts.MaxBytes - pending.byteCount,
 					Heartbeat: sub.consumeOpts.Heartbeat,
 				}
+
 				sub.resetPendingMsgs()
 			}
 		}()
@@ -203,7 +208,12 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 			if msgErr == nil {
 				return
 			}
-			if err := sub.handleStatusMsg(msg, msgErr); err != nil {
+
+			sub.Lock()
+			err := sub.handleStatusMsg(msg, msgErr)
+			sub.Unlock()
+
+			if err != nil {
 				if atomic.LoadUint32(&sub.closed) == 1 {
 					return
 				}
@@ -642,7 +652,8 @@ func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
 				}
 				userMsg, err := checkMsg(msg)
 				if err != nil {
-					if !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, ErrNoMessages) && !errors.Is(err, ErrMaxBytesExceeded) {
+					errNotTimeoutOrNoMsgs := !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, ErrNoMessages)
+					if errNotTimeoutOrNoMsgs && !errors.Is(err, ErrMaxBytesExceeded) {
 						res.err = err
 					}
 					res.done = true
diff --git a/js.go b/js.go
index 9ec99d7..76a55d4 100644
--- a/js.go
+++ b/js.go
@@ -679,6 +679,15 @@ func (js *js) newAsyncReply() string {
 	return sb.String()
 }
 
+func (js *js) cleanupReplySub() {
+	js.mu.Lock()
+	if js.rsub != nil {
+		js.rsub.Unsubscribe()
+		js.rsub = nil
+	}
+	js.mu.Unlock()
+}
+
 // registerPAF will register for a PubAckFuture.
 func (js *js) registerPAF(id string, paf *pubAckFuture) (int, int) {
 	js.mu.Lock()
diff --git a/micro/service.go b/micro/service.go
index f4eaeda..445c81f 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -106,15 +106,14 @@ type (
 
 	// EndpointStats contains stats for a specific endpoint.
 	EndpointStats struct {
-		Name                  string            `json:"name"`
-		Subject               string            `json:"subject"`
-		Metadata              map[string]string `json:"metadata"`
-		NumRequests           int               `json:"num_requests"`
-		NumErrors             int               `json:"num_errors"`
-		LastError             string            `json:"last_error"`
-		ProcessingTime        time.Duration     `json:"processing_time"`
-		AverageProcessingTime time.Duration     `json:"average_processing_time"`
-		Data                  json.RawMessage   `json:"data,omitempty"`
+		Name                  string          `json:"name"`
+		Subject               string          `json:"subject"`
+		NumRequests           int             `json:"num_requests"`
+		NumErrors             int             `json:"num_errors"`
+		LastError             string          `json:"last_error"`
+		ProcessingTime        time.Duration   `json:"processing_time"`
+		AverageProcessingTime time.Duration   `json:"average_processing_time"`
+		Data                  json.RawMessage `json:"data,omitempty"`
 	}
 
 	// Ping is the response type for PING monitoring endpoint.
@@ -126,12 +125,12 @@ type (
 	// Info is the basic information about a service type.
 	Info struct {
 		ServiceIdentity
-		Type        string   `json:"type"`
-		Description string   `json:"description"`
-		Subjects    []string `json:"subjects"`
+		Type        string         `json:"type"`
+		Description string         `json:"description"`
+		Endpoints   []EndpointInfo `json:"endpoints"`
 	}
 
-	EndpointSchema struct {
+	EndpointInfo struct {
 		Name     string            `json:"name"`
 		Subject  string            `json:"subject"`
 		Metadata map[string]string `json:"metadata"`
@@ -140,6 +139,8 @@ type (
 	// Endpoint manages a service endpoint.
 	Endpoint struct {
 		EndpointConfig
+		Name string
+
 		service *service
 
 		stats        EndpointStats
@@ -397,6 +398,7 @@ func addEndpoint(s *service, name, subject string, handler Handler, metadata map
 			Handler:  handler,
 			Metadata: metadata,
 		},
+		Name: name,
 	}
 	sub, err := s.nc.QueueSubscribe(
 		subject,
@@ -411,9 +413,8 @@ func addEndpoint(s *service, name, subject string, handler Handler, metadata map
 	endpoint.subscription = sub
 	s.endpoints = append(s.endpoints, endpoint)
 	endpoint.stats = EndpointStats{
-		Name:     name,
-		Subject:  subject,
-		Metadata: endpoint.Metadata,
+		Name:    name,
+		Subject: subject,
 	}
 	return nil
 }
@@ -647,16 +648,20 @@ func (s *service) serviceIdentity() ServiceIdentity {
 
 // Info returns information about the service
 func (s *service) Info() Info {
-	endpoints := make([]string, 0, len(s.endpoints))
+	endpoints := make([]EndpointInfo, 0, len(s.endpoints))
 	for _, e := range s.endpoints {
-		endpoints = append(endpoints, e.Subject)
+		endpoints = append(endpoints, EndpointInfo{
+			Name:     e.Name,
+			Subject:  e.Subject,
+			Metadata: e.Metadata,
+		})
 	}
 
 	return Info{
 		ServiceIdentity: s.serviceIdentity(),
 		Type:            InfoResponseType,
 		Description:     s.Config.Description,
-		Subjects:        endpoints,
+		Endpoints:       endpoints,
 	}
 }
 
@@ -675,7 +680,6 @@ func (s *service) Stats() Stats {
 		endpointStats := &EndpointStats{
 			Name:                  endpoint.stats.Name,
 			Subject:               endpoint.stats.Subject,
-			Metadata:              endpoint.stats.Metadata,
 			NumRequests:           endpoint.stats.NumRequests,
 			NumErrors:             endpoint.stats.NumErrors,
 			LastError:             endpoint.stats.LastError,
@@ -765,9 +769,8 @@ func (e *Endpoint) stop() error {
 
 func (e *Endpoint) reset() {
 	e.stats = EndpointStats{
-		Name:     e.stats.Name,
-		Subject:  e.stats.Subject,
-		Metadata: e.stats.Metadata,
+		Name:    e.stats.Name,
+		Subject: e.stats.Subject,
 	}
 }
 
diff --git a/object.go b/object.go
index e513768..6bce605 100644
--- a/object.go
+++ b/object.go
@@ -370,14 +370,17 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 	}
 
 	// Create our own JS context to handle errors etc.
-	js, err := obs.js.nc.JetStream(PublishAsyncErrHandler(func(js JetStream, _ *Msg, err error) { setErr(err) }))
+	jetStream, err := obs.js.nc.JetStream(PublishAsyncErrHandler(func(js JetStream, _ *Msg, err error) { setErr(err) }))
 	if err != nil {
 		return nil, err
 	}
+
+	defer jetStream.(*js).cleanupReplySub()
+
 	purgePartial := func() {
 		// wait until all pubs are complete or up to default timeout before attempting purge
 		select {
-		case <-js.PublishAsyncComplete():
+		case <-jetStream.PublishAsyncComplete():
 		case <-time.After(obs.js.opts.wait):
 		}
 		obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj})
@@ -423,7 +426,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 			h.Write(m.Data)
 
 			// Send msg itself.
-			if _, err := js.PublishMsgAsync(m); err != nil {
+			if _, err := jetStream.PublishMsgAsync(m); err != nil {
 				purgePartial()
 				return nil, err
 			}
@@ -458,7 +461,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 	}
 
 	// Publish the meta message.
-	_, err = js.PublishMsgAsync(mm)
+	_, err = jetStream.PublishMsgAsync(mm)
 	if err != nil {
 		if r != nil {
 			purgePartial()
@@ -468,7 +471,7 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 
 	// Wait for all to be processed.
 	select {
-	case <-js.PublishAsyncComplete():
+	case <-jetStream.PublishAsyncComplete():
 		if err := getErr(); err != nil {
 			if r != nil {
 				purgePartial()


Test patch:
diff --git a/jetstream/test/helper_test.go b/jetstream/test/helper_test.go
index 88e4e73..c8c4d18 100644
--- a/jetstream/test/helper_test.go
+++ b/jetstream/test/helper_test.go
@@ -156,18 +156,21 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 	nodes := make([]*jsServer, size)
 	opts := make([]*server.Options, 0)
 
-	getAddr := func() (string, string, int) {
+	var activeListeners []net.Listener
+	getAddr := func(t *testing.T) (string, string, int) {
 		l, err := net.Listen("tcp", "127.0.0.1:0")
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
-		defer l.Close()
 
 		addr := l.Addr()
 		host := addr.(*net.TCPAddr).IP.String()
 		port := addr.(*net.TCPAddr).Port
-		l.Close()
 		time.Sleep(100 * time.Millisecond)
+
+		// we cannot close the listener immediately to avoid duplicate port binding
+		// the returned net.Listener has to be closed after all ports are drawn
+		activeListeners = append(activeListeners, l)
 		return addr.String(), host, port
 	}
 
@@ -184,11 +187,11 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 
 		if size > 1 {
 			o.Cluster.Name = clusterName
-			_, host1, port1 := getAddr()
+			_, host1, port1 := getAddr(t)
 			o.Host = host1
 			o.Port = port1
 
-			addr2, host2, port2 := getAddr()
+			addr2, host2, port2 := getAddr(t)
 			o.Cluster.Host = host2
 			o.Cluster.Port = port2
 			o.Tags = []string{o.ServerName}
@@ -196,6 +199,10 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 		}
 		opts = append(opts, &o)
 	}
+	// close all connections used to randomize ports
+	for _, l := range activeListeners {
+		l.Close()
+	}
 
 	if size > 1 {
 		routesStr := server.RoutesFromStr(strings.Join(routes, ","))
diff --git a/jetstream/test/stream_test.go b/jetstream/test/stream_test.go
index e94a941..3543775 100644
--- a/jetstream/test/stream_test.go
+++ b/jetstream/test/stream_test.go
@@ -35,27 +35,27 @@ func TestAddConsumer(t *testing.T) {
 	}{
 		{
 			name:           "create durable pull consumer",
-			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", AckPolicy: jetstream.AckExplicitPolicy},
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
 			shouldCreate:   true,
 		},
 		{
 			name:           "create ephemeral pull consumer",
-			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy},
+			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy},
 			shouldCreate:   true,
 		},
 		{
 			name:           "with filter subject",
-			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy, FilterSubject: "FOO.A"},
+			consumerConfig: jetstream.ConsumerConfig{FilterSubject: "FOO.A"},
 			shouldCreate:   true,
 		},
 		{
 			name:           "with multiple filter subjects",
-			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{"FOO.A", "FOO.B"}},
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", "FOO.B"}},
 			shouldCreate:   true,
 		},
 		{
 			name:           "consumer already exists, update",
-			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", AckPolicy: jetstream.AckExplicitPolicy, Description: "test consumer"},
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", Description: "test consumer"},
 		},
 		{
 			name:           "consumer already exists, illegal update",
@@ -64,7 +64,7 @@ func TestAddConsumer(t *testing.T) {
 		},
 		{
 			name:           "invalid durable name",
-			consumerConfig: jetstream.ConsumerConfig{Durable: "dur.123", AckPolicy: jetstream.AckExplicitPolicy},
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur.123"},
 			withError:      jetstream.ErrInvalidConsumerName,
 		},
 	}
@@ -112,10 +112,13 @@ func TestAddConsumer(t *testing.T) {
 					t.Fatalf("Expected request on %s; got %s", sub.Subject, err)
 				}
 			}
-			_, err = s.Consumer(ctx, c.CachedInfo().Name)
+			ci, err := s.Consumer(ctx, c.CachedInfo().Name)
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
+			if ci.CachedInfo().Config.AckPolicy != test.consumerConfig.AckPolicy {
+				t.Fatalf("Invalid ack policy; want: %s; got: %s", test.consumerConfig.AckPolicy, ci.CachedInfo().Config.AckPolicy)
+			}
 		})
 	}
 }
diff --git a/micro/example_test.go b/micro/example_test.go
index 41434b2..a1fcd5a 100644
--- a/micro/example_test.go
+++ b/micro/example_test.go
@@ -170,7 +170,7 @@ func ExampleService_Info() {
 	fmt.Println(info.Name)
 	fmt.Println(info.Description)
 	fmt.Println(info.Version)
-	fmt.Println(info.Subjects)
+	fmt.Println(info.Endpoints)
 }
 
 func ExampleService_Stats() {
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index d88336a..6129671 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -466,8 +466,8 @@ func TestAddService(t *testing.T) {
 			if test.givenConfig.Endpoint != nil {
 				subjectsNum += 1
 			}
-			if subjectsNum != len(info.Subjects) {
-				t.Fatalf("Invalid number of registered endpoints; want: %d; got: %d", subjectsNum, len(info.Subjects))
+			if subjectsNum != len(info.Endpoints) {
+				t.Fatalf("Invalid number of registered endpoints; want: %d; got: %d", subjectsNum, len(info.Endpoints))
 			}
 			pingSubject, err := micro.ControlSubject(micro.PingVerb, info.Name, info.ID)
 			if err != nil {
@@ -655,39 +655,54 @@ func TestErrHandlerSubjectMatch(t *testing.T) {
 
 func TestGroups(t *testing.T) {
 	tests := []struct {
-		name            string
-		endpointName    string
-		groups          []string
-		expectedSubject string
+		name             string
+		endpointName     string
+		groups           []string
+		expectedEndpoint micro.EndpointInfo
 	}{
 		{
-			name:            "no groups",
-			endpointName:    "foo",
-			expectedSubject: "foo",
+			name:         "no groups",
+			endpointName: "foo",
+			expectedEndpoint: micro.EndpointInfo{
+				Name:    "foo",
+				Subject: "foo",
+			},
 		},
 		{
-			name:            "single group",
-			endpointName:    "foo",
-			groups:          []string{"g1"},
-			expectedSubject: "g1.foo",
+			name:         "single group",
+			endpointName: "foo",
+			groups:       []string{"g1"},
+			expectedEndpoint: micro.EndpointInfo{
+				Name:    "foo",
+				Subject: "g1.foo",
+			},
 		},
 		{
-			name:            "single empty group",
-			endpointName:    "foo",
-			groups:          []string{""},
-			expectedSubject: "foo",
+			name:         "single empty group",
+			endpointName: "foo",
+			groups:       []string{""},
+			expectedEndpoint: micro.EndpointInfo{
+				Name:    "foo",
+				Subject: "foo",
+			},
 		},
 		{
-			name:            "empty groups",
-			endpointName:    "foo",
-			groups:          []string{"", "g1", ""},
-			expectedSubject: "g1.foo",
+			name:         "empty groups",
+			endpointName: "foo",
+			groups:       []string{"", "g1", ""},
+			expectedEndpoint: micro.EndpointInfo{
+				Name:    "foo",
+				Subject: "g1.foo",
+			},
 		},
 		{
-			name:            "multiple groups",
-			endpointName:    "foo",
-			groups:          []string{"g1", "g2", "g3"},
-			expectedSubject: "g1.g2.g3.foo",
+			name:         "multiple groups",
+			endpointName: "foo",
+			groups:       []string{"g1", "g2", "g3"},
+			expectedEndpoint: micro.EndpointInfo{
+				Name:    "foo",
+				Subject: "g1.g2.g3.foo",
+			},
 		},
 	}
 
@@ -728,11 +743,11 @@ func TestGroups(t *testing.T) {
 			}
 
 			info := srv.Info()
-			if len(info.Subjects) != 1 {
-				t.Fatalf("Expected 1 registered endpoint; got: %d", len(info.Subjects))
+			if len(info.Endpoints) != 1 {
+				t.Fatalf("Expected 1 registered endpoint; got: %d", len(info.Endpoints))
 			}
-			if info.Subjects[0] != test.expectedSubject {
-				t.Fatalf("Invalid subject; want: %s, got: %s", test.expectedSubject, info.Subjects[0])
+			if !reflect.DeepEqual(info.Endpoints[0], test.expectedEndpoint) {
+				t.Fatalf("Invalid endpoint; want: %s, got: %s", test.expectedEndpoint, info.Endpoints[0])
 			}
 		})
 	}
@@ -832,7 +847,13 @@ func TestMonitoringHandlers(t *testing.T) {
 					ID:       info.ID,
 					Metadata: map[string]string{},
 				},
-				Subjects: []string{"test.func"},
+				Endpoints: []micro.EndpointInfo{
+					{
+						Name:     "default",
+						Subject:  "test.func",
+						Metadata: map[string]string{"basic": "schema"},
+					},
+				},
 			},
 		},
 		{
@@ -846,7 +867,13 @@ func TestMonitoringHandlers(t *testing.T) {
 					ID:       info.ID,
 					Metadata: map[string]string{},
 				},
-				Subjects: []string{"test.func"},
+				Endpoints: []micro.EndpointInfo{
+					{
+						Name:     "default",
+						Subject:  "test.func",
+						Metadata: map[string]string{"basic": "schema"},
+					},
+				},
 			},
 		},
 		{
@@ -860,7 +887,13 @@ func TestMonitoringHandlers(t *testing.T) {
 					ID:       info.ID,
 					Metadata: map[string]string{},
 				},
-				Subjects: []string{"test.func"},
+				Endpoints: []micro.EndpointInfo{
+					{
+						Name:     "default",
+						Subject:  "test.func",
+						Metadata: map[string]string{"basic": "schema"},
+					},
+				},
 			},
 		},
 		{
@@ -1105,11 +1138,6 @@ func TestServiceStats(t *testing.T) {
 			if stats.Type != micro.StatsResponseType {
 				t.Errorf("Invalid response type; want: %s; got: %s", micro.StatsResponseType, stats.Type)
 			}
-			if test.config.Endpoint != nil && test.config.Endpoint.Metadata != nil {
-				if !reflect.DeepEqual(test.config.Endpoint.Metadata, stats.Endpoints[0].Metadata) {
-					t.Errorf("invalid endpoint metadata: %v", stats.Endpoints[0].Metadata)
-				}
-			}
 
 			if test.expectedStats != nil {
 				var data map[string]interface{}
diff --git a/test/js_test.go b/test/js_test.go
index de3780d..3ce30be 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -5572,18 +5572,21 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 	nodes := make([]*jsServer, size)
 	opts := make([]*server.Options, 0)
 
-	getAddr := func() (string, string, int) {
+	var activeListeners []net.Listener
+	getAddr := func(t *testing.T) (string, string, int) {
 		l, err := net.Listen("tcp", "127.0.0.1:0")
 		if err != nil {
-			panic(err)
+			t.Fatalf("Unexpected error: %v", err)
 		}
-		defer l.Close()
 
 		addr := l.Addr()
 		host := addr.(*net.TCPAddr).IP.String()
 		port := addr.(*net.TCPAddr).Port
-		l.Close()
 		time.Sleep(100 * time.Millisecond)
+
+		// we cannot close the listener immediately to avoid duplicate port binding
+		// the returned net.Listener has to be closed after all ports are drawn
+		activeListeners = append(activeListeners, l)
 		return addr.String(), host, port
 	}
 
@@ -5600,11 +5603,11 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 
 		if size > 1 {
 			o.Cluster.Name = clusterName
-			_, host1, port1 := getAddr()
+			_, host1, port1 := getAddr(t)
 			o.Host = host1
 			o.Port = port1
 
-			addr2, host2, port2 := getAddr()
+			addr2, host2, port2 := getAddr(t)
 			o.Cluster.Host = host2
 			o.Cluster.Port = port2
 			o.Tags = []string{o.ServerName}
@@ -5612,6 +5615,10 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 		}
 		opts = append(opts, &o)
 	}
+	// close all connections used to randomize ports
+	for _, l := range activeListeners {
+		l.Close()
+	}
 
 	if size > 1 {
 		routesStr := server.RoutesFromStr(strings.Join(routes, ","))