Merge #152961

152961: kvserver/rangefeed: deflake TestUnbufferedRegOnCatchUpSwitchOver/disconnect_after_runOutputLoop r=wenyihu6 a=stevendanna

This test was using the capacity of the catchUpBuf to decide how many post-catch-up events need to be delivered. This is a problem because the the output loop might have already finished the catchup scan, flushed its empty catch up buffer, set the buffer to nil (0 cap), and exited before we ever even observe the cap of the catchup buffer.

One note is that this test is a bit odd because depending on the speed of the catchup up scan, the events might come to us via the catchup buffer publish or directly.

Fixes #152960
Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/rangefeed/unbuffered_registration_test.go

@@ -359,19 +359,17 @@ func TestUnbufferedRegOnCatchUpSwitchOver(t *testing.T) {
 			defer wg.Done()
 			r.runOutputLoop(ctx, 0)
 		}()
-		capOfBuf := cap(r.mu.catchUpBuf)
-		r.publish(ctx, ev1, &SharedBudgetAllocation{refCount: 1})
-		r.publish(ctx, ev2, &SharedBudgetAllocation{refCount: 1})
-		r.publish(ctx, ev3, &SharedBudgetAllocation{refCount: 1})
-		r.publish(ctx, ev4, &SharedBudgetAllocation{refCount: 1})
-		r.publish(ctx, ev5, &SharedBudgetAllocation{refCount: 1})
+		postCatchUpEvents := []*kvpb.RangeFeedEvent{ev1, ev2, ev3, ev4, ev5}
+		for _, ev := range postCatchUpEvents {
+			r.publish(ctx, ev, &SharedBudgetAllocation{refCount: 1})
+		}
 		catchUpEvents := expEvents(false)
 		wg.Wait()
 
 		require.False(t, r.getOverflowed())
 		require.Nil(t, r.getBuf())
-		s.waitForEventCount(t, capOfBuf+len(catchUpEvents))
+		s.waitForEventCount(t, len(postCatchUpEvents)+len(catchUpEvents))
 		require.Equal(t,
-			[]*kvpb.RangeFeedEvent{ev1, ev2, ev3, ev4, ev5}, s.mu.events[len(catchUpEvents):])
+			postCatchUpEvents, s.mu.events[len(catchUpEvents):])
 	})
 }