Merge e0fe99d into 0493cfd

go.mod

@@ -4,7 +4,7 @@ go 1.15
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.2.1-0.20210322182015-0a48911c00c6
+	github.com/nats-io/nats-server/v2 v2.2.1-0.20210327180151-03aee09847d0
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0

go.sum

@@ -32,8 +32,8 @@ github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad/go.mod h1
 github.com/nats-io/nats-server/v2 v2.1.8-0.20201129161730-ebe63db3e3ed/go.mod h1:XD0zHR/jTXdZvWaQfS5mQgsXj6x12kMjKLyAk/cOGgY=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20210205154825-f7ab27f7dad4/go.mod h1:kauGd7hB5517KeSqspW2U1Mz/jhPbTrE8eOXzUPk1m0=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20210227190344-51550e242af8/go.mod h1:/QQ/dpqFavkNhVnjvMILSQ3cj5hlmhB66adlgNbjuoA=
-github.com/nats-io/nats-server/v2 v2.2.1-0.20210322182015-0a48911c00c6 h1:qbo0+To1ZYHWPqbaxWKj2Nv5w5aKJoC9AVVxv/Cvg0M=
-github.com/nats-io/nats-server/v2 v2.2.1-0.20210322182015-0a48911c00c6/go.mod h1:eKlAaGmSQHZMFQA6x56AaP5/Bl9N3mWF4awyT2TTpzc=
+github.com/nats-io/nats-server/v2 v2.2.1-0.20210327180151-03aee09847d0 h1:ybeT5VFA73CVQb4rCL+48+up91xWheriSBbJ3M2Pzps=
+github.com/nats-io/nats-server/v2 v2.2.1-0.20210327180151-03aee09847d0/go.mod h1:eKlAaGmSQHZMFQA6x56AaP5/Bl9N3mWF4awyT2TTpzc=
 github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nats.go v1.10.1-0.20200531124210-96f2130e4d55/go.mod h1:ARiFsjW9DVxk48WJbO3OSZ2DG8fjkMi7ecLmXoY/n9I=
 github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a/go.mod h1:8eAIv96Mo9QW6Or40jUHejS7e4VwZ3VRYD6Sf0BTDp4=

test/js_test.go

@@ -581,9 +581,6 @@ func TestJetStreamSubscribe(t *testing.T) {
 }
 
 func TestJetStreamAckPending_Pull(t *testing.T) {
-	// TODO(jaime): Re-enable after API changes.
-	t.SkipNow()
-
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()
 
@@ -3363,13 +3360,16 @@ func withJSServer(t *testing.T, tfn func(t *testing.T, srvs ...*jsServer)) {
 	opts := natsserver.DefaultTestOptions
 	opts.Port = -1
 	opts.JetStream = true
+	opts.LameDuckDuration = 3 * time.Second
+	opts.LameDuckGracePeriod = 2 * time.Second
 	s := &jsServer{Server: RunServerWithOptions(opts), myopts: &opts}
 
 	defer func() {
 		if config := s.JetStreamConfig(); config != nil {
 			os.RemoveAll(config.StoreDir)
 		}
 		s.Shutdown()
+		s.WaitForShutdown()
 	}()
 	tfn(t, s)
 }
@@ -3387,6 +3387,7 @@ func withJSCluster(t *testing.T, clusterName string, size int, tfn func(t *testi
 			}
 			node.restart.Unlock()
 			node.Shutdown()
+			node.WaitForShutdown()
 		}
 	}()
 	tfn(t, nodes...)
@@ -4024,8 +4025,6 @@ func TestJetStream_ClusterReconnect(t *testing.T) {
 func testJetStream_ClusterReconnectDurableQueueSubscriber(t *testing.T, subject string, srvs ...*jsServer) {
 	var (
 		srvA          = srvs[0]
-		srvB          = srvs[1]
-		srvC          = srvs[2]
 		totalMsgs     = 20
 		reconnected   = make(chan struct{})
 		reconnectDone bool
@@ -4063,9 +4062,6 @@ func testJetStream_ClusterReconnectDurableQueueSubscriber(t *testing.T, subject
 	msgs := make(chan *nats.Msg, totalMsgs)
 
 	// Create some queue subscribers.
-	srvAClientURL := srvA.ClientURL()
-	srvBClientURL := srvB.ClientURL()
-	srvCClientURL := srvC.ClientURL()
 	for i := 0; i < 5; i++ {
 		expected := totalMsgs
 		dname := "dur"
@@ -4077,18 +4073,7 @@ func testJetStream_ClusterReconnectDurableQueueSubscriber(t *testing.T, subject
 			case count == 2:
 				// Do not ack and wait for redelivery on reconnect.
 				srvA.Shutdown()
-				return
-			case count == 11:
-				// Do another Shutdown of the server we are connected with.
-				switch nc.ConnectedUrl() {
-				case srvAClientURL:
-					srvA.Shutdown()
-				case srvBClientURL:
-					srvB.Shutdown()
-				case srvCClientURL:
-					srvC.Shutdown()
-				default:
-				}
+				srvA.WaitForShutdown()
 				return
 			case count == expected:
 				done()
@@ -4109,7 +4094,7 @@ func testJetStream_ClusterReconnectDurableQueueSubscriber(t *testing.T, subject
 					}
 				}
 			}
-		}, nats.Durable(dname), nats.ManualAck())
+		}, nats.Durable(dname), nats.AckWait(5*time.Second), nats.ManualAck())
 
 		if err != nil && (err != nats.ErrTimeout && err != context.DeadlineExceeded) {
 			t.Error(err)