Merge #133894

133894: crosscluster/producer: flush at lower threshold if consumer is ready r=dt a=msbutler This patch allows the event stream to flush at a lower buffer threshold if the consumer is ready for more work. If a consumer is waiting around for more work, the producer should give them more work to keep consumer busy. If the consumer is not ready, the event stream will buffer to a higher threshold, preventing a flush from blocking rangefeeds, inducing catchup scans. Epic: none Release note: none Co-authored-by: Michael Butler <[email protected]>
cockroachdb · Oct 30, 2024 · 548e585 · 548e585
2 parents ef26d15 + 7552cec
commit 548e585
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 2 deletions.
diff --git a/pkg/ccl/crosscluster/logical/logical_replication_writer_processor.go b/pkg/ccl/crosscluster/logical/logical_replication_writer_processor.go
@@ -133,7 +133,7 @@ var batchSizeSetting = settings.RegisterByteSizeSetting(
 	settings.ApplicationLevel,
 	"logical_replication.stream_batch_size",
 	"target batch size for logical replication stream",
-	1<<20,
+	16<<20,
 )
 
 func newLogicalReplicationWriterProcessor(

diff --git a/pkg/ccl/crosscluster/producer/event_stream.go b/pkg/ccl/crosscluster/producer/event_stream.go
@@ -8,6 +8,7 @@ package producer
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/crosscluster"
@@ -68,6 +69,8 @@ type eventStream struct {
 	lastCheckpointLen  int
 
 	debug streampb.DebugProducerStatus
+
+	consumerReady atomic.Bool
 }
 
 var quantize = settings.RegisterDurationSettingWithExplicitUnit(
@@ -233,6 +236,9 @@ func (s *eventStream) Next(ctx context.Context) (bool, error) {
 	s.debug.Flushes.EmitWaitNanos.Add(emitWait)
 	s.debug.LastPolledMicros.Store(timeutil.Now().UnixMicro())
 
+	s.consumerReady.Store(true)
+	defer s.consumerReady.Store(false)
+
 	select {
 	case <-ctx.Done():
 		return false, ctx.Err()
@@ -388,7 +394,13 @@ func (s *eventStream) sendCheckpoint(ctx context.Context, frontier rangefeed.Vis
 }
 
 func (s *eventStream) maybeFlushBatch(ctx context.Context) error {
-	if s.seb.size > int(s.spec.Config.BatchByteSize) {
+	// If the consumer is ready to ingest, flush at a lower threshold. This
+	// ensures the consumer always has work to do.
+	//
+	// If the consumer is not ready, the larger batch delays the flush call and
+	// preventing the slow consumer from blocking rangefeed progress, avoiding
+	// catchup scans.
+	if (s.seb.size > int(s.spec.Config.BatchByteSize)) || s.consumerReady.Load() && s.seb.size > minBatchByteSize {
 		return s.flushBatch(ctx)
 	}
 	return nil
@@ -497,6 +509,7 @@ func (s *eventStream) validateProducerJobAndSpec(ctx context.Context) (roachpb.T
 }
 
 const defaultBatchSize = 1 << 20
+const minBatchByteSize = 1 << 20
 
 func streamPartition(
 	evalCtx *eval.Context, streamID streampb.StreamID, opaqueSpec []byte,