Merge pull request #8489 from planetscale/rn-vstreamer-throttler-goro…

…utine-leak

Return from throttler goroutine if context is cancelled to prevent goroutine leaks

go/vt/vttablet/tabletserver/vstreamer/vstreamer.go

@@ -271,16 +271,31 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 		for {
 			// check throttler.
 			if !vs.vse.throttlerClient.ThrottleCheckOKOrWait(ctx) {
+				select {
+				// make sure to leave if context is cancelled
+				case <-ctx.Done():
+					return
+				default:
+					// do nothing special
+				}
 				continue
 			}
-
-			ev, ok := <-events
-			if ok {
-				throttledEvents <- ev
-			} else {
-				close(throttledEvents)
+			select {
+			case ev, ok := <-events:
+				if ok {
+					select {
+					case throttledEvents <- ev:
+					case <-ctx.Done():
+						return
+					}
+				} else {
+					close(throttledEvents)
+					return
+				}
+			case <-ctx.Done():
 				return
 			}
+
 		}
 	}()
 	for {