crosscluster/producer: add debug setting to drop all data

For debugging purposes, it can be useful to run the replication flow
but drop rather than emit all data changes. As this loses all data, it
is gated behind the unsafe setting interlock.

Release note: none.
Epic: none.

Author: dt

pkg/crosscluster/producer/event_stream.go

@@ -366,15 +366,27 @@ func (s *eventStream) maybeFlushBatch(ctx context.Context) error {
 	return nil
 }
 
+var debugSettingDropData = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"physical_replication.producer.unsafe_debug.discard_all_data.enabled",
+	"discard all row data during cluster replication (for experimental debugging purposes only)",
+	false,
+	settings.WithUnsafe,
+)
+
 func (s *eventStream) flushBatch(ctx context.Context, reason streampb.FlushReason) error {
 	if s.seb.size == 0 {
 		return nil
 	}
+	defer s.seb.reset()
+
+	if debugSettingDropData.Get(s.execCfg.SV()) {
+		return nil
+	}
+
 	s.seqNum++
 	s.debug.Flushed(int64(s.seb.size), reason, s.seqNum)
 
-	defer s.seb.reset()
-
 	return s.sendFlush(ctx, &streampb.StreamEvent{StreamSeq: s.seqNum, Batch: &s.seb.batch})
 }
 func (s *eventStream) sendFlush(ctx context.Context, event *streampb.StreamEvent) error {