integration: defer clus.Terminate in watch tests

Common pattern was defer cancel(), but clus.Terminate() at the end of
the test. This appears to lead to a deadlock that is only released
once the context times out, causing inflated test times.

integration/v3_watch_test.go

@@ -407,6 +407,7 @@ func TestV3WatchCancelUnsynced(t *testing.T) {
 
 func testV3WatchCancel(t *testing.T, startRev int64) {
 	clus := NewClusterV3(t, &ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
@@ -455,8 +456,6 @@ func testV3WatchCancel(t *testing.T, startRev int64) {
 	if !rok {
 		t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
 	}
-
-	clus.Terminate(t)
 }
 
 // TestV3WatchCurrentPutOverlap ensures current watchers receive all events with
@@ -541,7 +540,10 @@ func TestV3WatchCurrentPutOverlap(t *testing.T) {
 
 // TestV3WatchEmptyKey ensures synced watchers see empty key PUTs as PUT events
 func TestV3WatchEmptyKey(t *testing.T) {
+	defer testutil.AfterTest(t)
+
 	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
@@ -581,8 +583,6 @@ func TestV3WatchEmptyKey(t *testing.T) {
 	if !reflect.DeepEqual(resp.Events, wevs) {
 		t.Fatalf("got %v, expected %v", resp.Events, wevs)
 	}
-
-	clus.Terminate(t)
 }
 
 func TestV3WatchMultipleWatchersSynced(t *testing.T) {
@@ -601,6 +601,8 @@ func TestV3WatchMultipleWatchersUnsynced(t *testing.T) {
 // one watcher to test if it receives expected events.
 func testV3WatchMultipleWatchers(t *testing.T, startRev int64) {
 	clus := NewClusterV3(t, &ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
 	kvc := toGRPC(clus.RandClient()).KV
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
@@ -686,8 +688,6 @@ func testV3WatchMultipleWatchers(t *testing.T, startRev int64) {
 	if !rok {
 		t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
 	}
-
-	clus.Terminate(t)
 }
 
 func TestV3WatchMultipleEventsTxnSynced(t *testing.T) {
@@ -703,6 +703,7 @@ func TestV3WatchMultipleEventsTxnUnsynced(t *testing.T) {
 // testV3WatchMultipleEventsTxn tests Watch APIs when it receives multiple events.
 func testV3WatchMultipleEventsTxn(t *testing.T, startRev int64) {
 	clus := NewClusterV3(t, &ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
@@ -772,9 +773,6 @@ func testV3WatchMultipleEventsTxn(t *testing.T, startRev int64) {
 	if !rok {
 		t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
 	}
-
-	// can't defer because tcp ports will be in use
-	clus.Terminate(t)
 }
 
 type eventsSortByKey []*mvccpb.Event
@@ -875,6 +873,8 @@ func TestV3WatchMultipleStreamsUnsynced(t *testing.T) {
 // testV3WatchMultipleStreams tests multiple watchers on the same key on multiple streams.
 func testV3WatchMultipleStreams(t *testing.T, startRev int64) {
 	clus := NewClusterV3(t, &ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
 	wAPI := toGRPC(clus.RandClient()).Watch
 	kvc := toGRPC(clus.RandClient()).KV
 
@@ -939,8 +939,6 @@ func testV3WatchMultipleStreams(t *testing.T, startRev int64) {
 		}(i)
 	}
 	wg.Wait()
-
-	clus.Terminate(t)
 }
 
 // waitResponse waits on the given stream for given duration.