From 872e5a78a376d8130e77b0ca6f7b322e8d182b6d Mon Sep 17 00:00:00 2001
From: Piotr Tabor <ptab@google.com>
Date: Fri, 11 Sep 2020 12:48:07 +0200
Subject: [PATCH] clientv3/integration: Fix TestLeasingTxtOwnerGet test getting
 stuck in 'cluster_proxy' mode:

Fixes:
  go test --tags cluster_proxy --timeout=30m -run TestLeasingTxnOwnerGet -v ./clientv3/integration/...

The explicit code to close client is needed due to:
  https://github.com/etcd-io/etcd/blob/76e769ce95ca0d4d0e3486712d96956260db04b8/clientv3/watch.go#L72
as just ctx close by LeasingKeyValue store does not interrupts opened Watches.
The only way to interrupt open Watch is to close the 'whole' Watcher / Client.
---
 clientv3/integration/leasing_test.go | 24 +++++++++++++++++++-----
 1 file changed, 19 insertions(+), 5 deletions(-)

diff --git a/clientv3/integration/leasing_test.go b/clientv3/integration/leasing_test.go
index ed1a2df7b..3b641899d 100644
--- a/clientv3/integration/leasing_test.go
+++ b/clientv3/integration/leasing_test.go
@@ -619,16 +619,29 @@ func TestLeasingTxnOwnerGet(t *testing.T) {
 	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
 	defer clus.Terminate(t)
 
-	lkv, closeLKV, err := leasing.NewKV(clus.Client(0), "pfx/")
-	testutil.AssertNil(t, err)
-	defer closeLKV()
+	client := clus.Client(0)
 
+	lkv, closeLKV, err := leasing.NewKV(client, "pfx/")
+	testutil.AssertNil(t, err)
+
+	defer func() {
+		// In '--tags cluster_proxy' mode the client need to be closed before
+		// closeLKV(). This interrupts all outstanding watches. Closing by closeLKV()
+		// is not sufficient as (unfortunately) context close does not interrupts Watches.
+		// See ./clientv3/watch.go:
+		// >> Currently, client contexts are overwritten with "valCtx" that never closes. <<
+		clus.TakeClient(0) // avoid double Close() of the client.
+		client.Close()
+		closeLKV()
+	}()
+
+	// TODO: Randomization in tests is a bad practice (except explicitly exploratory).
 	keyCount := rand.Intn(10) + 1
 	var ops []clientv3.Op
 	presps := make([]*clientv3.PutResponse, keyCount)
 	for i := range presps {
 		k := fmt.Sprintf("k-%d", i)
-		presp, err := clus.Client(0).Put(context.TODO(), k, k+k)
+		presp, err := client.Put(context.TODO(), k, k+k)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -639,6 +652,8 @@ func TestLeasingTxnOwnerGet(t *testing.T) {
 		}
 		ops = append(ops, clientv3.OpGet(k))
 	}
+
+	// TODO: Randomization in unit tests is a bad practice (except explicitly exploratory).
 	ops = ops[:rand.Intn(len(ops))]
 
 	// served through cache
@@ -648,7 +663,6 @@ func TestLeasingTxnOwnerGet(t *testing.T) {
 	cmps, useThen := randCmps("k-", presps)
 
 	if useThen {
-
 		thenOps = ops
 		elseOps = []clientv3.Op{clientv3.OpPut("k", "1")}
 	} else {