Repository: nats-io/nats.go
Pull request #1284 (2023-06-01 14:58:50)

[FIXED] Flaky tests picking same ports more than once
Some clustered tests were failing with "Unable to start NATS Server in Go Routine" panic from server.

This was most probably caused by client choosing random server/cluster ports, where it sometimes chose same port more than once, as observed by adding this log after each test using `withJSCluster()`

```
=== RUN   TestJetStreamConsumerReplicasOption/TestJetStreamConsumerReplicasOption-3

Server 0 , server name:  NODE_0 , server port:  38473 , cluster port:  44391 , routes:  [127.0.0.1:44391 127.0.0.1:44625 127.0.0.1:38473]

Server 1 , server name:  NODE_1 , server port:  39243 , cluster port:  44625 , routes:  [127.0.0.1:44391 127.0.0.1:44625 127.0.0.1:38473]

Server 2 , server name:  NODE_2 , server port:  45755 , cluster port:  38473 , routes:  [127.0.0.1:44391 127.0.0.1:44625 127.0.0.1:38473]

```

In the example above, the test panicked because server port on Server 0 was the same as cluster port on Server 2.


Patch:
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"
 }


Test patch:
diff --git a/jetstream/test/helper_test.go b/jetstream/test/helper_test.go
index 88e4e73..44f6b9e 100644
--- a/jetstream/test/helper_test.go
+++ b/jetstream/test/helper_test.go
@@ -156,7 +156,8 @@ 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)
@@ -166,8 +167,11 @@ func setupJSClusterWithSize(t *testing.T, clusterName string, size int) []*jsSer
 		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 +188,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 +200,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/test/js_test.go b/test/js_test.go
index de3780d..9a765b9 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -5572,18 +5572,22 @@ 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 +5604,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 +5616,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, ","))