diff --git a/docs/modules/components/pages/inputs/gcp_spanner_cdc.adoc b/docs/modules/components/pages/inputs/gcp_spanner_cdc.adoc index 13f00c47e4..dd996da1ea 100644 --- a/docs/modules/components/pages/inputs/gcp_spanner_cdc.adoc +++ b/docs/modules/components/pages/inputs/gcp_spanner_cdc.adoc @@ -51,6 +51,7 @@ input: byte_size: 0 period: "" check: "" + checkpoint_limit: 1024 auto_replay_nacks: true ``` @@ -81,6 +82,7 @@ input: period: "" check: "" processors: [] # No default (optional) + checkpoint_limit: 1024 auto_replay_nacks: true ``` @@ -320,6 +322,15 @@ processors: format: json_array ``` +=== `checkpoint_limit` + +The maximum number of messages that can be processed at a given time per partition. Increasing this limit enables parallel processing and batching at the output level. Any given partition watermark will not be committed unless all messages under that offset are delivered in order to preserve at least once delivery guarantees. + + +*Type*: `int` + +*Default*: `1024` + === `auto_replay_nacks` Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation. diff --git a/internal/impl/gcp/enterprise/input_spanner_cdc.go b/internal/impl/gcp/enterprise/input_spanner_cdc.go index c9848aa1fa..d93ceebf6f 100644 --- a/internal/impl/gcp/enterprise/input_spanner_cdc.go +++ b/internal/impl/gcp/enterprise/input_spanner_cdc.go @@ -39,16 +39,21 @@ const ( siFieldMinWatermarkCacheTTL = "min_watermark_cache_ttl" siFieldAllowedModTypes = "allowed_mod_types" siFieldBatchPolicy = "batching" + siFieldCheckpointLimit = "checkpoint_limit" ) // Default values const ( defaultMetadataTableFormat = "cdc_metadata_%s" shutdownTimeout = 5 * time.Second + defaultCheckpointLimit = 1024 ) type spannerCDCInputConfig struct { changestreams.Config + // CheckpointLimit caps in-flight (unacknowledged) messages per partition; + // the ordered watermark tracker blocks once reached, applying backpressure. + CheckpointLimit int } func parseRFC3339Nano(pConf *service.ParsedConfig, key string) (time.Time, error) { @@ -116,6 +121,9 @@ func spannerCDCInputConfigFromParsed(pConf *service.ParsedConfig) (conf spannerC if conf.MinWatermarkCacheTTL, err = pConf.FieldDuration(siFieldMinWatermarkCacheTTL); err != nil { return } + if conf.CheckpointLimit, err = pConf.FieldInt(siFieldCheckpointLimit); err != nil { + return + } return } @@ -155,6 +163,10 @@ https://cloud.google.com/spanner/docs/change-streams Field(service.NewStringListField(siFieldAllowedModTypes).Advanced().Optional().Description("List of modification types to process. If not specified, all modification types are processed. Allowed values: INSERT, UPDATE, DELETE"). ShortDescription("Modification types to process: INSERT, UPDATE, DELETE. All are processed if unset.").Example([]string{"INSERT", "UPDATE", "DELETE"})). Field(service.NewBatchPolicyField(siFieldBatchPolicy)). + Field(service.NewIntField(siFieldCheckpointLimit). + Description("The maximum number of messages that can be processed at a given time per partition. Increasing this limit enables parallel processing and batching at the output level. Any given partition watermark will not be committed unless all messages under that offset are delivered in order to preserve at least once delivery guarantees."). + ShortDescription("The maximum number of in-flight messages per partition."). + Default(defaultCheckpointLimit)). Field(service.NewAutoRetryNacksToggleField()) } @@ -186,6 +198,10 @@ type spannerCDCReader struct { resCh chan asyncMessage subscriber *changestreams.Subscriber stopSig *shutdown.Signaller + + // updateWatermark persists a partition watermark; set to the subscriber's + // UpdatePartitionWatermark in Connect, overridable in tests. + updateWatermark func(ctx context.Context, partitionToken string, ts time.Time) error } var _ service.BatchInput = (*spannerCDCReader)(nil) @@ -216,14 +232,25 @@ func newSpannerCDCReader(conf spannerCDCInputConfig, batching service.BatchPolic log: mgr.Logger(), metrics: changestreams.NewMetrics(mgr.Metrics(), conf.StreamID), batching: batching, - batcher: newSpannerPartitionBatcherFactory(batching, mgr), + batcher: newSpannerPartitionBatcherFactory(batching, mgr, conf.CheckpointLimit), resCh: make(chan asyncMessage), stopSig: shutdown.NewSignaller(), } } +// resetPartitionBatchers discards all cached partition batchers. Called on +// (re)connect: stale batchers hold rows and ack state from the previous +// subscriber session, and the re-read from the persisted watermarks +// re-delivers those rows anyway. The factory is reset in place (never +// swapped) because straggler goroutines from the previous session may still +// hold the pointer. +func (r *spannerCDCReader) resetPartitionBatchers(ctx context.Context) { + r.batcher.Reset(ctx) +} + func (r *spannerCDCReader) emit( ctx context.Context, + batcher *spannerPartitionBatcher, partitionToken string, msg service.MessageBatch, commitTimestamp time.Time, @@ -231,10 +258,31 @@ func (r *spannerCDCReader) emit( if len(msg) == 0 { return nil, nil } + // A zero commitTimestamp means "mid-record, do not advance": substitute + // the last known-safe watermark so an out-of-order resolve can never + // regress or stall behind it. Per-partition callbacks are serialized, so + // this needs no locking. + if commitTimestamp.IsZero() { + commitTimestamp = batcher.lastWatermark + } else { + batcher.lastWatermark = commitTimestamp + } + resolveFn, err := batcher.cp.Track(ctx, commitTimestamp, int64(len(msg))) + if err != nil { + return nil, fmt.Errorf("tracking watermark checkpoint: %w", err) + } ackOnce := ack.NewOnce(func(ctx context.Context) error { + // Only the resolved (contiguous-prefix) watermark is safe to persist: + // a batch acked out of order must not advance the watermark past + // still-unacked earlier batches, or a crash in that window would skip + // their records on restart. + resolved := resolveFn() + if resolved == nil || resolved.IsZero() { + return nil + } // If we processed the message and failed to update the watermark, we // would try to update it on the next message, no need to return an error here. - if err := r.subscriber.UpdatePartitionWatermark(ctx, partitionToken, commitTimestamp); err != nil { + if err := r.updateWatermark(ctx, partitionToken, *resolved); err != nil { r.log.Errorf("%s: failed to update watermark: %v", partitionToken, err) } return nil @@ -268,7 +316,7 @@ func (r *spannerCDCReader) onDataChangeRecord(ctx context.Context, partitionToke if err != nil { return err } - ack, err := r.emit(ctx, partitionToken, msg, ts) + ack, err := r.emit(ctx, batcher, partitionToken, msg, ts) if err != nil { return err } @@ -289,7 +337,7 @@ func (r *spannerCDCReader) onDataChangeRecord(ctx context.Context, partitionToke if err != nil { return err } - ack, err := r.emit(ctx, partitionToken, msg, ts) + ack, err := r.emit(ctx, batcher, partitionToken, msg, ts) if err != nil { return err } @@ -300,7 +348,7 @@ func (r *spannerCDCReader) onDataChangeRecord(ctx context.Context, partitionToke iter := batcher.MaybeFlushWith(dcr) for mb, ts := range iter.Iter(ctx) { - ack, err := r.emit(ctx, partitionToken, mb, ts) + ack, err := r.emit(ctx, batcher, partitionToken, mb, ts) if err != nil { return err } @@ -318,20 +366,29 @@ func (r *spannerCDCReader) Connect(ctx context.Context) error { r.conf.StreamID, r.conf.ProjectID, r.conf.InstanceID, r.conf.DatabaseID) var cb changestreams.CallbackFunc = r.onDataChangeRecord + var flushersDone func() if r.batching.Period != "" { r.log.Infof("Periodic flushing enabled: %s", r.batching.Period) - p := periodicallyFlushingSpannerCDCReader{ + p := &periodicallyFlushingSpannerCDCReader{ spannerCDCReader: r, reqCh: make(map[string]chan callbackRequest), } cb = p.onDataChangeRecord + flushersDone = p.wg.Wait } + // Discard any partition batchers from a previous subscriber session: + // their buffered rows were never acked and will be re-read from the + // persisted watermarks; reusing them would duplicate rows into mixed + // batches and misalign the ack tracker. + r.resetPartitionBatchers(ctx) + var err error r.subscriber, err = changestreams.NewSubscriber(ctx, r.conf.Config, cb, r.log, r.metrics) if err != nil { return fmt.Errorf("create Spanner change stream reader: %w", err) } + r.updateWatermark = r.subscriber.UpdatePartitionWatermark if err := r.subscriber.Setup(ctx); err != nil { return fmt.Errorf("setup Spanner change stream reader: %w", err) @@ -342,11 +399,18 @@ func (r *spannerCDCReader) Connect(ctx context.Context) error { ctx, cancel := r.stopSig.SoftStopCtx(context.Background()) go func() { - defer cancel() if err := r.subscriber.Run(ctx); err != nil { r.log.Errorf("Spanner change stream reader error: %v", err) } r.subscriber.Close() + // Stop the per-partition flusher goroutines and wait for them to exit + // BEFORE signalling stopped: TriggerHasStopped is what lets the + // framework call Connect again, and a straggler flusher racing the + // next session's partition-batcher factory would corrupt its state. + cancel() + if flushersDone != nil { + flushersDone() + } r.stopSig.TriggerHasStopped() }() @@ -401,6 +465,9 @@ type periodicallyFlushingSpannerCDCReader struct { *spannerCDCReader mu sync.RWMutex reqCh map[string]chan callbackRequest + // wg tracks the per-partition flusher goroutines so a reconnect can wait + // for the previous session's flushers to exit before starting. + wg sync.WaitGroup } func (r *periodicallyFlushingSpannerCDCReader) onDataChangeRecord(ctx context.Context, partitionToken string, dcr *changestreams.DataChangeRecord) error { @@ -416,7 +483,7 @@ func (r *periodicallyFlushingSpannerCDCReader) onDataChangeRecord(ctx context.Co r.mu.Unlock() softStopCh := r.stopSig.SoftStopChan() - go func() { + r.wg.Go(func() { r.log.Debugf("%s: starting periodic flusher", partitionToken) defer func() { r.mu.Lock() @@ -444,14 +511,23 @@ func (r *periodicallyFlushingSpannerCDCReader) onDataChangeRecord(ctx context.Co cr.errCh <- r.spannerCDCReader.onDataChangeRecord(ctx, partitionToken, cr.dcr) } } - }() + }) } r.mu.RLock() ch := r.reqCh[partitionToken] r.mu.RUnlock() + if ch == nil { + // The flusher has already exited (shutdown/reconnect in progress). + return fmt.Errorf("no active flusher for partition %s", partitionToken) + } - errCh := make(chan error) + // Buffered so the flusher's reply can never block: the requester abandons + // errCh when the errgroup context is cancelled mid-wait, and an + // unbuffered send would wedge the flusher forever - Run then blocks in + // flushersDone() before TriggerHasStopped, so the input would never + // reconnect. + errCh := make(chan error, 1) select { case <-ctx.Done(): return ctx.Err() diff --git a/internal/impl/gcp/enterprise/input_spanner_cdc_test.go b/internal/impl/gcp/enterprise/input_spanner_cdc_test.go new file mode 100644 index 0000000000..71736b609b --- /dev/null +++ b/internal/impl/gcp/enterprise/input_spanner_cdc_test.go @@ -0,0 +1,172 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package enterprise + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/public/service" +) + +// watermarkRecorder is a test seam for spannerCDCReader.updateWatermark. +type watermarkRecorder struct { + mu sync.Mutex + writes []time.Time +} + +func (w *watermarkRecorder) update(_ context.Context, _ string, ts time.Time) error { + w.mu.Lock() + defer w.mu.Unlock() + w.writes = append(w.writes, ts) + return nil +} + +func (w *watermarkRecorder) recorded() []time.Time { + w.mu.Lock() + defer w.mu.Unlock() + return append([]time.Time(nil), w.writes...) +} + +func newTestSpannerReader(t *testing.T) (*spannerCDCReader, *spannerPartitionBatcher, *watermarkRecorder) { + t.Helper() + + r := newSpannerCDCReader(spannerCDCInputConfig{}, service.BatchPolicy{Count: 1}, service.MockResources()) + rec := &watermarkRecorder{} + r.updateWatermark = rec.update + + // Drain emitted messages so emit's channel send never blocks; acks are + // driven explicitly by the tests via the returned ack.Once. + go func() { + for { + select { + case <-t.Context().Done(): + return + case <-r.resCh: + } + } + }() + + batcher, _, err := r.batcher.forPartition("p1") + require.NoError(t, err) + return r, batcher, rec +} + +func testBatch() service.MessageBatch { + return service.MessageBatch{service.NewMessage([]byte("{}"))} +} + +func TestSpannerEmitOrderedWatermarks(t *testing.T) { + t1 := time.Date(2026, 8, 6, 10, 0, 0, 0, time.UTC) + t2 := t1.Add(time.Second) + t3 := t2.Add(time.Second) + + t.Run("out-of-order acks never advance past unacked batches", func(t *testing.T) { + ctx := t.Context() + r, batcher, rec := newTestSpannerReader(t) + + ack1, err := r.emit(ctx, batcher, "p1", testBatch(), t1) + require.NoError(t, err) + ack2, err := r.emit(ctx, batcher, "p1", testBatch(), t2) + require.NoError(t, err) + + // Acking only the LATER batch must not write anything: its records + // are durable but the earlier batch's are not. (The pre-fix code + // wrote t2 here - the data-loss window.) + require.NoError(t, ack2.Ack(ctx, nil)) + require.Empty(t, rec.recorded(), "watermark must not advance past a still-unacked earlier batch") + + // Acking the earlier batch resolves the full prefix: one write, t2. + require.NoError(t, ack1.Ack(ctx, nil)) + require.Equal(t, []time.Time{t2}, rec.recorded()) + }) + + t.Run("in-order acks advance incrementally", func(t *testing.T) { + ctx := t.Context() + r, batcher, rec := newTestSpannerReader(t) + + ack1, err := r.emit(ctx, batcher, "p1", testBatch(), t1) + require.NoError(t, err) + ack2, err := r.emit(ctx, batcher, "p1", testBatch(), t2) + require.NoError(t, err) + + require.NoError(t, ack1.Ack(ctx, nil)) + require.NoError(t, ack2.Ack(ctx, nil)) + require.Equal(t, []time.Time{t1, t2}, rec.recorded()) + }) + + t.Run("zero-watermark batches carry the last safe watermark forward", func(t *testing.T) { + ctx := t.Context() + r, batcher, rec := newTestSpannerReader(t) + + ack1, err := r.emit(ctx, batcher, "p1", testBatch(), t1) + require.NoError(t, err) + require.NoError(t, ack1.Ack(ctx, nil)) + require.Equal(t, []time.Time{t1}, rec.recorded()) + + // b2 is a mid-record flush (zero watermark), b3 completes a record. + ack2, err := r.emit(ctx, batcher, "p1", testBatch(), time.Time{}) + require.NoError(t, err) + ack3, err := r.emit(ctx, batcher, "p1", testBatch(), t3) + require.NoError(t, err) + + // Acking b3 alone must not advance past t1 (b2 is unacked; a repeat + // write of the current safe watermark t1 is fine and idempotent). + // The pre-fix code wrote t3 here - the data-loss window. + require.NoError(t, ack3.Ack(ctx, nil)) + for _, w := range rec.recorded() { + require.False(t, w.After(t1), "watermark advanced past an unacked batch: %v", w) + } + // Acking b2 resolves the full prefix through b3. + require.NoError(t, ack2.Ack(ctx, nil)) + writes := rec.recorded() + require.Equal(t, t3, writes[len(writes)-1]) + }) + + t.Run("zero-watermark batch acked alone re-writes only the safe watermark", func(t *testing.T) { + ctx := t.Context() + r, batcher, rec := newTestSpannerReader(t) + + ack1, err := r.emit(ctx, batcher, "p1", testBatch(), t1) + require.NoError(t, err) + ack2, err := r.emit(ctx, batcher, "p1", testBatch(), time.Time{}) + require.NoError(t, err) + + require.NoError(t, ack1.Ack(ctx, nil)) + require.NoError(t, ack2.Ack(ctx, nil)) + + // The carried-forward value is t1; writes never regress and never + // mention a timestamp past the last completed record. + writes := rec.recorded() + require.NotEmpty(t, writes) + for _, w := range writes { + require.Equal(t, t1, w) + } + }) +} + +func TestSpannerConnectResetsPartitionBatchers(t *testing.T) { + r, batcher, _ := newTestSpannerReader(t) + _ = batcher + + // Same token returns the cached batcher before reset... + _, existed, err := r.batcher.forPartition("p1") + require.NoError(t, err) + require.True(t, existed) + + // ...and a fresh one after the reset performed on (re)connect. + r.resetPartitionBatchers(t.Context()) + _, existed, err = r.batcher.forPartition("p1") + require.NoError(t, err) + require.False(t, existed, "reconnect must not reuse stale partition batcher state") +} diff --git a/internal/impl/gcp/enterprise/input_spanner_partition_batcher.go b/internal/impl/gcp/enterprise/input_spanner_partition_batcher.go index 2d3c20f674..dbf6467ac5 100644 --- a/internal/impl/gcp/enterprise/input_spanner_partition_batcher.go +++ b/internal/impl/gcp/enterprise/input_spanner_partition_batcher.go @@ -16,6 +16,8 @@ import ( "sync" "time" + "github.com/Jeffail/checkpoint" + "github.com/redpanda-data/benthos/v4/public/service" "github.com/redpanda-data/connect/v4/internal/ack" "github.com/redpanda-data/connect/v4/internal/impl/gcp/enterprise/changestreams" @@ -109,6 +111,15 @@ type spannerPartitionBatcher struct { period *time.Timer acks []*ack.Once rm func() + + // cp orders in-flight batches so the partition watermark only ever + // advances to a commit timestamp once every batch at or below it has been + // acked (see spannerCDCReader.emit). + cp *checkpoint.Capped[time.Time] + // lastWatermark is the most recent non-zero watermark emitted for this + // partition; mid-record (zero watermark) batches carry it forward. Only + // accessed from the partition's serialized callback. + lastWatermark time.Time } func (s *spannerPartitionBatcher) MaybeFlushWith(dcr *changestreams.DataChangeRecord) *spannerPartitionBatchIter { @@ -173,6 +184,9 @@ func (s *spannerPartitionBatcher) Close(ctx context.Context) error { type spannerPartitionBatcherFactory struct { batching service.BatchPolicy res *service.Resources + // checkpointLimit caps in-flight (unacknowledged) messages tracked per + // partition; Track blocks once reached, applying backpressure. + checkpointLimit int mu sync.RWMutex partitions map[string]*spannerPartitionBatcher @@ -181,11 +195,34 @@ type spannerPartitionBatcherFactory struct { func newSpannerPartitionBatcherFactory( batching service.BatchPolicy, res *service.Resources, + checkpointLimit int, ) *spannerPartitionBatcherFactory { + if checkpointLimit <= 0 { + checkpointLimit = defaultCheckpointLimit + } return &spannerPartitionBatcherFactory{ - batching: batching, - res: res, - partitions: make(map[string]*spannerPartitionBatcher), + batching: batching, + res: res, + checkpointLimit: checkpointLimit, + partitions: make(map[string]*spannerPartitionBatcher), + } +} + +// Reset discards every cached partition batcher, closing each one (stopping +// its period timer and closing its service.Batcher). Used on reconnect so no +// stale rows or ack state leak into the new session; the factory pointer is +// never swapped because straggler goroutines from the previous session may +// still hold it. +func (f *spannerPartitionBatcherFactory) Reset(ctx context.Context) { + f.mu.Lock() + old := f.partitions + f.partitions = make(map[string]*spannerPartitionBatcher) + f.mu.Unlock() + + for token, spb := range old { + if err := spb.Close(ctx); err != nil { + f.res.Logger().Debugf("%s: closing discarded partition batcher: %v", token, err) + } } } @@ -200,14 +237,21 @@ func (f *spannerPartitionBatcherFactory) forPartition(partitionToken string) (*s return nil, false, err } - spb = &spannerPartitionBatcher{ + newSpb := &spannerPartitionBatcher{ batcher: b, - rm: func() { - f.mu.Lock() + cp: checkpoint.NewCapped[time.Time](int64(f.checkpointLimit)), + } + // rm removes only this exact instance: after a Reset, a new session's + // batcher may occupy the same token and must not be evicted by the + // discarded one's Close. + newSpb.rm = func() { + f.mu.Lock() + if f.partitions[partitionToken] == newSpb { delete(f.partitions, partitionToken) - f.mu.Unlock() - }, + } + f.mu.Unlock() } + spb = newSpb if d, ok := spb.batcher.UntilNext(); ok { spb.period = time.NewTimer(d) } diff --git a/internal/plugins/cdctest/cdc_conformance_test.go b/internal/plugins/cdctest/cdc_conformance_test.go index cd5eac4de5..8d2efd3613 100644 --- a/internal/plugins/cdctest/cdc_conformance_test.go +++ b/internal/plugins/cdctest/cdc_conformance_test.go @@ -76,7 +76,6 @@ var knownNonConformant = map[string]map[string]string{ }, "gcp_spanner_cdc": { "checkpoint_cache": "not yet exposed under the canonical name", - "checkpoint_limit": "not yet exposed under the canonical name", "snapshot_max_batch_size": "not yet exposed under the canonical name", "max_parallel_snapshot_tables": "not yet exposed under the canonical name", "stream_snapshot": "not yet exposed under the canonical name",