changefeedccl: make mock pulsar sink synchronous

Although the comments in mock pulsar sink say that emitting messages and
flushing are synchronous, the use of a buffered channel to transmit
messages from the mock producer to the test feed makes it asynchronous.
This leads to problems where the consumer (the test) may not have stored
the data in a "durable" way before the changefeed job completes. As a
result, tests can be flaky as it does not always look like all rows have
been emitted.

This PR makes the mock pulsar sink synchronous by using an unbuffered
channel.

Fixes: #119289
Informs: #118899

Epic: CRDB-9180

Release note: None

pkg/ccl/changefeedccl/changefeed_test.go

@@ -7538,8 +7538,7 @@ func TestChangefeedOnlyInitialScanCSV(t *testing.T) {
 		}
 	}
 
-	// TODO(#119289): re-enable pulsar
-	cdcTest(t, testFn, feedTestEnterpriseSinks, feedTestOmitSinks("pulsar"))
+	cdcTest(t, testFn, feedTestEnterpriseSinks)
 }
 
 func TestChangefeedOnlyInitialScanCSVSinkless(t *testing.T) {

pkg/ccl/changefeedccl/testfeed_test.go

@@ -2584,7 +2584,8 @@ type mockPulsarServer struct {
 
 func makeMockPulsarServer() *mockPulsarServer {
 	return &mockPulsarServer{
-		msgCh: make(chan *pulsar.ProducerMessage, 2048),
+		// TODO(#118899): Make msgCh a buffered channel for async message sending.
+		msgCh: make(chan *pulsar.ProducerMessage),
 	}
 }