Merge pull request #145347 from asg0451/backport24.3-144497

asg0451 · web-flow · commit a29f43f43189 · 2025-04-28T12:14:29.000-04:00
release-24.3: changefeedccl: fix mock kafka server shutdown
diff --git a/pkg/ccl/changefeedccl/testfeed_test.go b/pkg/ccl/changefeedccl/testfeed_test.go
@@ -1819,11 +1819,15 @@ func (s *fakeKafkaSinkV2) Dial() error {
 			if m.Key != nil {
 				key = sarama.ByteEncoder(m.Key)
 			}
-			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,
+			}:
 			}
 		}
 		return nil

Original file line number	Diff line number	Diff line change
`@@ -1819,11 +1819,15 @@ func (s *fakeKafkaSinkV2) Dial() error {`
`1819`	`1819`	`if m.Key != nil {`
`1820`	`1820`	`key = sarama.ByteEncoder(m.Key)`
`1821`	`1821`	`}`
`1822`		`- s.feedCh <- &sarama.ProducerMessage{`
	`1822`	`+ select {`
	`1823`	`+ case <-ctx.Done():`
	`1824`	`+ return kgo.ProduceResults{kgo.ProduceResult{Err: ctx.Err()}}`
	`1825`	`+ case s.feedCh <- &sarama.ProducerMessage{`
`1823`	`1826`	`Topic: m.Topic,`
`1824`	`1827`	`Key: key,`
`1825`	`1828`	`Value: sarama.ByteEncoder(m.Value),`
`1826`	`1829`	`Partition: m.Partition,`
	`1830`	`+ }:`
`1827`	`1831`	`}`
`1828`	`1832`	`}`
`1829`	`1833`	`return nil`