Merge #73436

73436: kvserver: add timeouts for rangefeed test cases r=tbg a=erikgrinaker This splits out some test tweaks from #72724, since we're exploring alternative approaches for that PR and may abandon it. Release note: None Co-authored-by: Erik Grinaker <[email protected]>
cockroachdb · Dec 7, 2021 · bad623e · bad623e
2 parents 39923c0 + c3836b0
commit bad623e
Showing 1 changed file with 18 additions and 16 deletions.
diff --git a/pkg/kv/kvserver/replica_rangefeed_test.go b/pkg/kv/kvserver/replica_rangefeed_test.go
@@ -155,8 +155,9 @@ func TestReplicaRangefeed(t *testing.T) {
 				Span:     rangefeedSpan,
 				WithDiff: true,
 			}
-			pErr := store.RangeFeed(&req, stream)
-			streamErrC <- pErr
+			timer := time.AfterFunc(10*time.Second, stream.Cancel)
+			defer timer.Stop()
+			streamErrC <- store.RangeFeed(&req, stream)
 		}(i)
 	}
 
@@ -501,8 +502,9 @@ func TestReplicaRangefeedRetryErrors(t *testing.T) {
 				},
 				Span: rangefeedSpan,
 			}
-			pErr := store.RangeFeed(&req, stream)
-			streamErrC <- pErr
+			timer := time.AfterFunc(10*time.Second, stream.Cancel)
+			defer timer.Stop()
+			streamErrC <- store.RangeFeed(&req, stream)
 		}()
 
 		// Wait for the first checkpoint event.
@@ -535,6 +537,8 @@ func TestReplicaRangefeedRetryErrors(t *testing.T) {
 				},
 				Span: rangefeedSpan,
 			}
+			timer := time.AfterFunc(10*time.Second, stream.Cancel)
+			defer timer.Stop()
 			streamErrC <- store.RangeFeed(&req, stream)
 		}()
 
@@ -577,9 +581,9 @@ func TestReplicaRangefeedRetryErrors(t *testing.T) {
 				},
 				Span: rangefeedLeftSpan,
 			}
-
-			pErr := store.RangeFeed(&req, streamLeft)
-			streamLeftErrC <- pErr
+			timer := time.AfterFunc(10*time.Second, streamLeft.Cancel)
+			defer timer.Stop()
+			streamLeftErrC <- store.RangeFeed(&req, streamLeft)
 		}()
 
 		// Establish a rangefeed on the right replica.
@@ -593,9 +597,9 @@ func TestReplicaRangefeedRetryErrors(t *testing.T) {
 				},
 				Span: rangefeedRightSpan,
 			}
-
-			pErr := store.RangeFeed(&req, streamRight)
-			streamRightErrC <- pErr
+			timer := time.AfterFunc(10*time.Second, streamRight.Cancel)
+			defer timer.Stop()
+			streamRightErrC <- store.RangeFeed(&req, streamRight)
 		}()
 
 		// Wait for the first checkpoint event on each stream.
@@ -647,12 +651,9 @@ func TestReplicaRangefeedRetryErrors(t *testing.T) {
 				},
 				Span: rangefeedSpan,
 			}
-
 			timer := time.AfterFunc(10*time.Second, stream.Cancel)
 			defer timer.Stop()
-
-			pErr := partitionStore.RangeFeed(&req, stream)
-			streamErrC <- pErr
+			streamErrC <- partitionStore.RangeFeed(&req, stream)
 		}()
 
 		// Wait for the first checkpoint event.
@@ -757,8 +758,9 @@ func TestReplicaRangefeedRetryErrors(t *testing.T) {
 				Span: rangefeedSpan,
 			}
 			kvserver.RangefeedEnabled.Override(ctx, &store.ClusterSettings().SV, true)
-			pErr := store.RangeFeed(&req, stream)
-			streamErrC <- pErr
+			timer := time.AfterFunc(10*time.Second, stream.Cancel)
+			defer timer.Stop()
+			streamErrC <- store.RangeFeed(&req, stream)
 		}()
 
 		// Wait for the first checkpoint event.