Merge #123833

123833: streamingccl: small replicating split fixups  r=kev-cao a=msbutler

See commits.

Co-authored-by: Michael Butler <butler@cockroachlabs.com>

pkg/ccl/streamingccl/streamingest/stream_ingestion_processor.go

@@ -856,7 +856,6 @@ func (sip *streamIngestionProcessor) handleSplitEvent(key *roachpb.Key) error {
 	if !ingestSplitEvent.Get(&sip.EvalCtx.Settings.SV) {
 		return nil
 	}
-
 	kvDB := sip.FlowCtx.Cfg.DB.KV()
 	rekey, ok, err := sip.rekey(*key)
 	if err != nil {
@@ -865,7 +864,7 @@ func (sip *streamIngestionProcessor) handleSplitEvent(key *roachpb.Key) error {
 	if !ok {
 		return nil
 	}
-	log.Infof(ctx, "replicating split at %s", rekey)
+	log.Infof(ctx, "replicating split at %s", roachpb.Key(rekey).String())
 	expiration := kvDB.Clock().Now().AddDuration(time.Hour)
 	return kvDB.AdminSplit(ctx, rekey, expiration)
 }

pkg/ccl/streamingccl/streamproducer/event_stream.go

@@ -302,15 +302,15 @@ func (s *eventStream) onDeleteRange(ctx context.Context, delRange *kvpb.RangeFee
 	s.setErr(s.maybeFlushBatch(ctx))
 }
 func (s *eventStream) onMetadata(ctx context.Context, metadata *kvpb.RangeFeedMetadata) {
+	if s.addMu != nil {
+		// Split points can be sent concurrently during the initial scan.
+		s.addMu.Lock()
+		defer s.addMu.Unlock()
+	}
 	log.VInfof(ctx, 2, "received metadata event: %s, fromManualSplit: %t, parent start key %s", metadata.Span, metadata.FromManualSplit, metadata.ParentStartKey)
 	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))
 	}