changefeedccl: fix mock kafka server shutdown

Fix a test timeout due to a batching sink worker
being blocked on Flush after the context was
cancelled.

Fixes: #144213

Release note: None

Author: asg0451

pkg/ccl/changefeedccl/testfeed_test.go

@@ -1876,12 +1876,16 @@ func (s *fakeKafkaSinkV2) Dial() error {
 				})
 			}
 
-			s.feedCh <- &sarama.ProducerMessage{
+			select {
+			case <-ctx.Done():
+				return kgo.ProduceResults{kgo.ProduceResult{Err: ctx.Err()}}
+			case s.feedCh <- &sarama.ProducerMessage{
 				Topic:     m.Topic,
 				Key:       key,
 				Value:     sarama.ByteEncoder(m.Value),
 				Partition: m.Partition,
 				Headers:   headers,
+			}:
 			}
 		}
 		return nil