streamingccl: use mutex to buffer split point in event stream

Epic: none

Release note: none

pkg/ccl/streamingccl/streamproducer/event_stream.go

@@ -306,7 +306,13 @@ func (s *eventStream) onMetadata(ctx context.Context, metadata *kvpb.RangeFeedMe
 	if metadata.FromManualSplit && !metadata.Span.Key.Equal(metadata.ParentStartKey) {
 		// Only send new manual split keys (i.e. a child rangefeed start key that
 		// differs from the parent start key)
+		if s.addMu != nil {
+			// Split points can be sent concurrently during the initial scan.
+			s.addMu.Lock()
+			defer s.addMu.Unlock()
+		}
 		s.seb.addSplitPoint(metadata.Span.Key)
+		s.setErr(s.maybeFlushBatch(ctx))
 	}
 }