From 6b7116e154bec0098f2f84bf1a0ed7c45b471c46 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Mon, 17 Aug 2026 17:46:30 +0100 Subject: [PATCH 1/2] api: backfillasyc poc --- internal/component/errors.go | 5 + internal/component/input/async_reader.go | 114 ++++++++- .../input/async_reader_snapshot_test.go | 225 ++++++++++++++++++ internal/component/input/interface.go | 32 +++ public/service/errors.go | 9 + public/service/input.go | 65 ++++- public/service/input_auto_retry_batched.go | 176 +++++++++----- .../service/input_auto_retry_batched_test.go | 67 ++++++ public/service/input_test.go | 71 ++++++ 9 files changed, 703 insertions(+), 61 deletions(-) create mode 100644 internal/component/input/async_reader_snapshot_test.go diff --git a/internal/component/errors.go b/internal/component/errors.go index f90a835cd..35f52afba 100644 --- a/internal/component/errors.go +++ b/internal/component/errors.go @@ -50,6 +50,11 @@ var ( ErrNoAck = errors.New("failed to receive acknowledgement") ErrFailedSend = errors.New("message failed to reach a target destination") + + // ErrBackfillComplete is returned by a BackfillAsync reader's + // BackfillReadBatch method to indicate that its snapshot phase has been + // fully read. It is not itself an error condition. + ErrBackfillComplete = errors.New("snapshot phase complete") ) // ErrBackOff is an error returned that allows for a back off duration to be specified diff --git a/internal/component/input/async_reader.go b/internal/component/input/async_reader.go index 4a56cb2ef..9867f749b 100644 --- a/internal/component/input/async_reader.go +++ b/internal/component/input/async_reader.go @@ -164,6 +164,114 @@ func (r *AsyncReader) loop() { mConn.Incr(1) r.connection.Store(component.ConnectionActive(r.mgr)) + // runBackfillPhase drains a BackfillAsync reader's feed until + // component.ErrBackfillComplete, blocking until every dispatched batch + // is acknowledged (or nacked) before returning. + // + // ok is named because a hard stop also unblocks pendingBackfillAcks.Wait + // without every batch having genuinely settled - if any ack-goroutine + // was cut short rather than receiving a real ack/nack, ok must be false + // so the caller never calls BackfillComplete for unconfirmed rows. + runBackfillPhase := func(sr BackfillAsync) (ok bool) { + var pendingBackfillAcks sync.WaitGroup + var interruptedByHardStop atomic.Bool + defer func() { + r.mgr.Logger().Debug("Waiting for pending backfill acks to resolve.") + pendingBackfillAcks.Wait() + r.mgr.Logger().Debug("Pending backfill acks resolved.") + if interruptedByHardStop.Load() { + ok = false + } + }() + + for { + msg, ackFn, err := sr.BackfillReadBatch(closeAtLeisureCtx) + + if errors.Is(err, component.ErrBackfillComplete) { + return true + } + + if errors.Is(err, component.ErrNotConnected) { + mLostConn.Incr(1) + r.connection.Store(component.ConnectionFailing(r.mgr, component.ErrNotConnected)) + + if !initConnection() { + return false + } + mConn.Incr(1) + r.connection.Store(component.ConnectionActive(r.mgr)) + continue + } + + if r.shutSig.IsSoftStopSignalled() || errors.Is(err, component.ErrTypeClosed) { + return false + } + + if err != nil || len(msg) == 0 { + if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, component.ErrTimeout) && !errors.Is(err, component.ErrNotConnected) { + r.mgr.Logger().Error("Failed to read backfill batch: %v\n", err) + } + + nextBoff := r.readBackoff.NextBackOff() + if nextBoff == backoff.Stop { + r.mgr.Logger().Error("Maximum number of read attempt retries has been met, gracefully terminating input %v", r.typeStr) + return false + } + select { + case <-time.After(nextBoff): + case <-r.shutSig.SoftStopChan(): + return false + } + continue + } + + r.readBackoff.Reset() + mRcvd.Incr(int64(msg.Len())) + r.mgr.Logger().Trace("Consumed %v backfill messages from '%v'.\n", msg.Len(), r.typeStr) + + resChan := make(chan error, 1) + tracing.InitSpans(r.mgr.Tracer(), traceName, msg) + select { + case r.transactions <- message.NewTransaction(msg, resChan): + case <-r.shutSig.SoftStopChan(): + return false + } + + pendingBackfillAcks.Add(1) + go func(m message.Batch, aFn AsyncAckFn, rChan chan error) { + defer pendingBackfillAcks.Done() + + var res error + select { + case res = <-rChan: + case <-r.shutSig.HardStopChan(): + interruptedByHardStop.Store(true) + return + } + + tracing.FinishSpans(m) + if err := aFn(closeNowCtx, res); err != nil { + r.mgr.Logger().Error("Failed to acknowledge backfill message: %v\n", err) + } + }(msg, ackFn, resChan) + } + } + + if sr, ok := r.reader.(BackfillAsync); ok { + if !runBackfillPhase(sr) { + return + } + if sc, ok := r.reader.(BackfillCompleter); ok { + // Every batch handed off during the backfill phase is guaranteed + // resolved at this point, so it's safe for the reader to persist + // a post-backfill resume position now. + if err := sc.BackfillComplete(closeAtLeisureCtx); err != nil { + r.mgr.Logger().Error("Failed to finalize backfill phase for %v: %v", r.typeStr, err) + return + } + } + } + for { msg, ackFn, err := r.reader.ReadBatch(closeAtLeisureCtx) @@ -219,11 +327,7 @@ func (r *AsyncReader) loop() { } pendingAcks.Add(1) - go func( - m message.Batch, - aFn AsyncAckFn, - rChan chan error, - ) { + go func(m message.Batch, aFn AsyncAckFn, rChan chan error) { defer pendingAcks.Done() var res error diff --git a/internal/component/input/async_reader_snapshot_test.go b/internal/component/input/async_reader_snapshot_test.go new file mode 100644 index 000000000..492c3e09b --- /dev/null +++ b/internal/component/input/async_reader_snapshot_test.go @@ -0,0 +1,225 @@ +// Copyright 2025 Redpanda Data, Inc. + +package input_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/redpanda-data/benthos/v4/internal/component" + "github.com/redpanda-data/benthos/v4/internal/component/input" + "github.com/redpanda-data/benthos/v4/internal/manager/mock" + "github.com/redpanda-data/benthos/v4/internal/message" +) + +// mockBackfillAsyncReader implements input.BackfillAsync and +// input.BackfillCompleter, recording event order so tests can assert on the +// ack barrier. +type mockBackfillAsyncReader struct { + mu sync.Mutex + + backfillBatches [][]byte + backfillIdx int + streamBatches [][]byte + streamIdx int + + events []string + + backfillCompleteCalled chan struct{} +} + +func newMockBackfillAsyncReader(snapshotBatches, streamBatches [][]byte) *mockBackfillAsyncReader { + return &mockBackfillAsyncReader{ + backfillBatches: snapshotBatches, + streamBatches: streamBatches, + backfillCompleteCalled: make(chan struct{}), + } +} + +func (r *mockBackfillAsyncReader) record(ev string) { + r.mu.Lock() + r.events = append(r.events, ev) + r.mu.Unlock() +} + +func (r *mockBackfillAsyncReader) Events() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.events...) +} + +func (r *mockBackfillAsyncReader) ConnectionTest(ctx context.Context) component.ConnectionTestResults { + return component.ConnectionTestNotSupported(mock.NewManager()).AsList() +} + +func (r *mockBackfillAsyncReader) Connect(ctx context.Context) error { return nil } + +func (r *mockBackfillAsyncReader) BackfillReadBatch(ctx context.Context) (message.Batch, input.AsyncAckFn, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.backfillIdx >= len(r.backfillBatches) { + return nil, nil, component.ErrBackfillComplete + } + payload := r.backfillBatches[r.backfillIdx] + idx := r.backfillIdx + r.backfillIdx++ + + batch := message.Batch{message.NewPart(payload)} + return batch, func(ctx context.Context, err error) error { + r.record("snapshot-ack-" + string(rune('0'+idx))) + return nil + }, nil +} + +func (r *mockBackfillAsyncReader) BackfillComplete(ctx context.Context) error { + r.record("snapshot-complete") + close(r.backfillCompleteCalled) + return nil +} + +func (r *mockBackfillAsyncReader) ReadBatch(ctx context.Context) (message.Batch, input.AsyncAckFn, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.streamIdx >= len(r.streamBatches) { + // Stall rather than end the stream, so the test controls the + // component's lifecycle explicitly via TriggerStopConsuming. + return nil, nil, component.ErrTimeout + } + payload := r.streamBatches[r.streamIdx] + idx := r.streamIdx + r.streamIdx++ + + batch := message.Batch{message.NewPart(payload)} + return batch, func(ctx context.Context, err error) error { + r.record("stream-ack-" + string(rune('0'+idx))) + return nil + }, nil +} + +func (r *mockBackfillAsyncReader) Close(ctx context.Context) error { return nil } + +// TestAsyncReaderSnapshotPhaseBarrier proves AsyncReader waits for every +// snapshot batch to be acknowledged, even out of order, before calling +// BackfillComplete. +func TestAsyncReaderSnapshotPhaseBarrier(t *testing.T) { + readerImpl := newMockBackfillAsyncReader( + [][]byte{[]byte("snap-0"), []byte("snap-1")}, + [][]byte{[]byte("stream-0")}, + ) + + r, err := input.NewAsyncReader("foo", readerImpl, mock.NewManager()) + require.NoError(t, err) + r.TriggerStartConsuming() + defer func() { + r.TriggerStopConsuming() + require.NoError(t, r.WaitForClose(t.Context())) + }() + + ctx, done := context.WithTimeout(t.Context(), 10*time.Second) + defer done() + + tran0, ok := <-r.TransactionChan() + require.True(t, ok) + + tran1, ok := <-r.TransactionChan() + require.True(t, ok) + require.NoError(t, tran1.Ack(ctx, nil)) + + select { + case <-readerImpl.backfillCompleteCalled: + t.Fatal("BackfillComplete fired while a snapshot batch was still un-acked") + case <-time.After(100 * time.Millisecond): + } + + require.NoError(t, tran0.Ack(ctx, nil)) + + select { + case <-readerImpl.backfillCompleteCalled: + case <-time.After(3 * time.Second): + t.Fatal("BackfillComplete never fired after all snapshot batches were acked") + } + + // Proves streaming only begins once the snapshot phase has resolved. + streamTran, ok := <-r.TransactionChan() + require.True(t, ok) + require.NoError(t, streamTran.Ack(ctx, nil)) + + // ackFn runs in AsyncReader's own goroutine, not synchronously with + // Ack(), so give it a moment to execute before asserting on events. + require.Eventually(t, func() bool { + return len(readerImpl.Events()) == 4 + }, 3*time.Second, 10*time.Millisecond) + + events := readerImpl.Events() + require.Len(t, events, 4) + assert.Equal(t, "snapshot-ack-1", events[0]) + assert.Equal(t, "snapshot-ack-0", events[1]) + assert.Equal(t, "snapshot-complete", events[2]) + assert.Equal(t, "stream-ack-0", events[3]) +} + +// TestAsyncReaderSnapshotPhaseSkippedWhenEmpty proves BackfillComplete still +// fires once, immediately, for a reader with an empty snapshot phase. +func TestAsyncReaderSnapshotPhaseSkippedWhenEmpty(t *testing.T) { + readerImpl := newMockBackfillAsyncReader(nil, [][]byte{[]byte("stream-0")}) + + r, err := input.NewAsyncReader("foo", readerImpl, mock.NewManager()) + require.NoError(t, err) + r.TriggerStartConsuming() + defer func() { + r.TriggerStopConsuming() + require.NoError(t, r.WaitForClose(t.Context())) + }() + + ctx, done := context.WithTimeout(t.Context(), 10*time.Second) + defer done() + + select { + case <-readerImpl.backfillCompleteCalled: + case <-time.After(3 * time.Second): + t.Fatal("BackfillComplete never fired for an empty backfill phase") + } + + streamTran, ok := <-r.TransactionChan() + require.True(t, ok) + require.NoError(t, streamTran.Ack(ctx, nil)) + + require.Eventually(t, func() bool { + return len(readerImpl.Events()) == 2 + }, 3*time.Second, 10*time.Millisecond) + assert.Equal(t, []string{"backfill-complete", "stream-ack-0"}, readerImpl.Events()) +} + +// TestAsyncReaderBackfillPhaseAbortsOnHardStopBeforeAck proves a hard stop +// with an un-acked backfill batch in flight must not let BackfillComplete +// fire, even though the hard stop also unblocks the ack wait. +func TestAsyncReaderBackfillPhaseAbortsOnHardStopBeforeAck(t *testing.T) { + readerImpl := newMockBackfillAsyncReader( + [][]byte{[]byte("snap-0")}, + nil, + ) + + r, err := input.NewAsyncReader("foo", readerImpl, mock.NewManager()) + require.NoError(t, err) + r.TriggerStartConsuming() + + tran, ok := <-r.TransactionChan() + require.True(t, ok) + _ = tran // deliberately never acked - simulates a crash before the ack arrives. + + r.TriggerCloseNow() + require.NoError(t, r.WaitForClose(context.Background())) + + select { + case <-readerImpl.backfillCompleteCalled: + t.Fatal("BackfillComplete must not fire when a backfill batch was cut short by a hard stop instead of being genuinely acknowledged") + default: + } +} diff --git a/internal/component/input/interface.go b/internal/component/input/interface.go index 953ac688e..1946a8534 100644 --- a/internal/component/input/interface.go +++ b/internal/component/input/interface.go @@ -73,3 +73,35 @@ type Async interface { // completion or context cancellation. Close(ctx context.Context) error } + +// BackfillAsync is an optional extension of Async for readers that have a +// distinct initial backfill (backfill) phase before switching to continuous +// streaming, such as CDC-style connectors performing a table backfill ahead +// of log-based replication. +// +// When a reader implements this interface, AsyncReader calls +// BackfillReadBatch repeatedly - exactly like ReadBatch - until it returns +// component.ErrBackfillComplete. AsyncReader tracks every batch dispatched +// this way against a barrier scoped to the backfill phase, and blocks until +// each has been acknowledged (or nacked) downstream before moving on to +// steady-state ReadBatch calls. This removes the need for the reader to +// build its own ack-counting barrier in order to safely persist a +// post-backfill resume position. +type BackfillAsync interface { + Async + + // BackfillReadBatch attempts to read the next batch belonging to the + // backfill phase. Once the backfill has been fully read, including any + // trailing partial batch, this must return component.ErrBackfillComplete. + BackfillReadBatch(ctx context.Context) (message.Batch, AsyncAckFn, error) +} + +// BackfillCompleter is an optional extension for BackfillAsync readers that +// want to be notified once every batch read via BackfillReadBatch has been +// fully acknowledged or nacked downstream, and before AsyncReader begins +// calling ReadBatch. This is the safe point at which to persist a +// post-backfill resume position, since it's guaranteed no backfill batch is +// still in flight. +type BackfillCompleter interface { + BackfillComplete(ctx context.Context) error +} diff --git a/public/service/errors.go b/public/service/errors.go index 2b86d8bce..b632afdc5 100644 --- a/public/service/errors.go +++ b/public/service/errors.go @@ -29,6 +29,12 @@ var ( // ended (as indicated by EndOfInput). This error prompts the upstream // component to gracefully terminate the pipeline. ErrEndOfBuffer = errors.New("end of buffer") + + // ErrBackfillComplete is returned by a SnapshotBatchInput's + // BackfillReadBatch method to indicate that its snapshot phase has been + // fully read, including any trailing partial batch. It is not itself an + // error condition. + ErrBackfillComplete = errors.New("snapshot phase complete") ) // ErrBackOff is an error that plugins can optionally wrap another error with @@ -196,6 +202,9 @@ func publicToInternalErr(err error) error { if errors.Is(err, ErrEndOfBuffer) { return component.ErrTypeClosed } + if errors.Is(err, ErrBackfillComplete) { + return component.ErrBackfillComplete + } if errors.Is(err, ErrNotConnected) { return component.ErrNotConnected } diff --git a/public/service/input.go b/public/service/input.go index 5589e86a1..a526a9546 100644 --- a/public/service/input.go +++ b/public/service/input.go @@ -100,6 +100,31 @@ type BatchInput interface { //------------------------------------------------------------------------------ +// BackfillBatchInput is an optional extension of BatchInput for inputs with +// a distinct initial backfill phase before continuous streaming, such as CDC +// connectors snapshotting a table ahead of log-based replication. +// +// BackfillReadBatch is called repeatedly until it returns +// ErrBackfillComplete. Every batch it returns is guaranteed to be +// acknowledged (or nacked) before steady-state ReadBatch calls begin, so the +// input doesn't need its own ack-counting barrier. +type BackfillBatchInput interface { + BatchInput + + // BackfillReadBatch must return ErrBackfillComplete once the backfill, + // including any trailing partial batch, is fully read. + BackfillReadBatch(context.Context) (MessageBatch, AckFunc, error) +} + +// BackfillCompleter is an optional extension of BackfillBatchInput. +// BackfillComplete is called once every backfill batch has been settled +// downstream and before steady-state ReadBatch calls begin. +type BackfillCompleter interface { + BackfillComplete(context.Context) error +} + +//------------------------------------------------------------------------------ + // Implements input.AsyncReader. type airGapReader struct { o bundle.NewManagement @@ -147,7 +172,11 @@ type airGapBatchReader struct { } func newAirGapBatchReader(o bundle.NewManagement, r BatchInput) input.Async { - return &airGapBatchReader{o: o, r: r} + base := &airGapBatchReader{o: o, r: r} + if sr, ok := r.(BackfillBatchInput); ok { + return &airGapBackfillBatchReader{airGapBatchReader: base, sr: sr} + } + return base } func (a *airGapBatchReader) ConnectionTest(ctx context.Context) component.ConnectionTestResults { @@ -184,6 +213,40 @@ func (a *airGapBatchReader) Close(ctx context.Context) error { //------------------------------------------------------------------------------ +// Implements input.BackfillAsync and input.BackfillCompleter on top of +// airGapBatchReader. Only constructed when the wrapped BatchInput implements +// BackfillBatchInput - see newAirGapBatchReader. +type airGapBackfillBatchReader struct { + *airGapBatchReader + sr BackfillBatchInput +} + +func (a *airGapBackfillBatchReader) BackfillReadBatch(ctx context.Context) (message.Batch, input.AsyncAckFn, error) { + batch, ackFn, err := a.sr.BackfillReadBatch(ctx) + if err != nil { + return nil, nil, publicToInternalErr(err) + } + + mBatch := make(message.Batch, len(batch)) + for i, p := range batch { + mBatch[i] = p.part + } + return mBatch, func(c context.Context, r error) error { + r = toPublicBatchError(r) + return ackFn(c, r) + }, nil +} + +func (a *airGapBackfillBatchReader) BackfillComplete(ctx context.Context) error { + sc, ok := a.sr.(BackfillCompleter) + if !ok { + return nil + } + return publicToInternalErr(sc.BackfillComplete(ctx)) +} + +//------------------------------------------------------------------------------ + // ResourceInput provides access to an input resource. type ResourceInput struct { i input.Streamed diff --git a/public/service/input_auto_retry_batched.go b/public/service/input_auto_retry_batched.go index 8beed8888..37a88e30b 100644 --- a/public/service/input_auto_retry_batched.go +++ b/public/service/input_auto_retry_batched.go @@ -33,66 +33,89 @@ func AutoRetryNacksBatchedToggled(c *ParsedConfig, i BatchInput) (BatchInput, er // // When messages fail to be delivered they will be reattempted with back off // until success or the stream is stopped. +// +// If the wrapped input also implements BackfillBatchInput, the returned +// BatchInput does too, with the same auto-retry behaviour applied to its +// backfill phase via an independent retry list - a nacked backfill batch is +// replayed exactly like a nacked steady-state batch, and does not affect the +// BackfillAsync ack barrier, which only cares that every batch is eventually +// settled one way or another. func AutoRetryNacksBatched(i BatchInput) BatchInput { - return &autoRetryInputBatched{ - retryList: autoretry.NewList( - func(ctx context.Context) (MessageBatch, autoretry.AckFunc, error) { - t, aFn, err := i.ReadBatch(ctx) - - // Make sure we're able to track the position of messages in - // order to reassociate them after a batch-wide error - // downstream. - iParts := make([]*message.Part, len(t)) - for i, p := range t { - iParts[i] = p.part - } - - _, iParts = message.NewSortGroup(iParts) - for i, p := range iParts { - t[i] = NewInternalMessage(p) - } - - return t, autoretry.AckFunc(aFn), err - }, - func(t MessageBatch, err error) MessageBatch { - var bErr *batch.Error - if len(t) == 0 || !errors.As(err, &bErr) || bErr.IndexedErrors() == 0 { - return t - } - - sortGroup := message.TopLevelSortGroup(t[0].part) - if sortGroup == nil { - // We can't associate our source batch with the one that's associated - // with the batch error, therefore we fall back towards treating every - // message as if it was errored the same. - return t - } + base := &autoRetryInputBatched{ + child: i, + retryList: newBatchAutoRetryList(i.ReadBatch), + } + sr, ok := i.(BackfillBatchInput) + if !ok { + return base + } + return &autoRetryInputBatchedBackfill{ + autoRetryInputBatched: base, + backfillRetryList: newBatchAutoRetryList(sr.BackfillReadBatch), + } +} - sortBatch := make(message.Batch, len(t)) - for i, p := range t { - sortBatch[i] = p.part +// newBatchAutoRetryList builds an autoretry.List around any read function +// matching the BatchInput.ReadBatch/BackfillBatchInput.BackfillReadBatch +// shape, so the two phases can be given independent retry lists that share +// identical sort-group and batch-error-splitting behaviour. +func newBatchAutoRetryList(readFn func(context.Context) (MessageBatch, AckFunc, error)) *autoretry.List[MessageBatch] { + return autoretry.NewList( + func(ctx context.Context) (MessageBatch, autoretry.AckFunc, error) { + t, aFn, err := readFn(ctx) + + // Make sure we're able to track the position of messages in + // order to reassociate them after a batch-wide error + // downstream. + iParts := make([]*message.Part, len(t)) + for i, p := range t { + iParts[i] = p.part + } + + _, iParts = message.NewSortGroup(iParts) + for i, p := range iParts { + t[i] = NewInternalMessage(p) + } + + return t, autoretry.AckFunc(aFn), err + }, + func(t MessageBatch, err error) MessageBatch { + var bErr *batch.Error + if len(t) == 0 || !errors.As(err, &bErr) || bErr.IndexedErrors() == 0 { + return t + } + + sortGroup := message.TopLevelSortGroup(t[0].part) + if sortGroup == nil { + // We can't associate our source batch with the one that's associated + // with the batch error, therefore we fall back towards treating every + // message as if it was errored the same. + return t + } + + sortBatch := make(message.Batch, len(t)) + for i, p := range t { + sortBatch[i] = p.part + } + + seenIndexes := map[int]struct{}{} + newBatch := make(MessageBatch, 0, bErr.IndexedErrors()) + bErr.WalkPartsBySource(sortGroup, sortBatch, func(i int, p *message.Part, err error) bool { + if err == nil { + return true } - - seenIndexes := map[int]struct{}{} - newBatch := make(MessageBatch, 0, bErr.IndexedErrors()) - bErr.WalkPartsBySource(sortGroup, sortBatch, func(i int, p *message.Part, err error) bool { - if err == nil { - return true - } - if _, exists := seenIndexes[i]; exists { - return true - } - seenIndexes[i] = struct{}{} - newBatch = append(newBatch, &Message{part: p}) + if _, exists := seenIndexes[i]; exists { return true - }) - if len(newBatch) == 0 { - return t } - return newBatch - }), - child: i, - } + seenIndexes[i] = struct{}{} + newBatch = append(newBatch, &Message{part: p}) + return true + }) + if len(newBatch) == 0 { + return t + } + return newBatch + }) } //------------------------------------------------------------------------------ @@ -146,3 +169,46 @@ func (i *autoRetryInputBatched) Close(ctx context.Context) error { _ = i.retryList.Close(ctx) return i.child.Close(ctx) } + +//------------------------------------------------------------------------------ + +// autoRetryInputBatchedBackfill adds BackfillBatchInput/BackfillCompleter +// support on top of autoRetryInputBatched. It's only constructed by +// AutoRetryNacksBatched when the wrapped BatchInput implements +// BackfillBatchInput, so inputs without a backfill phase are unaffected. +type autoRetryInputBatchedBackfill struct { + *autoRetryInputBatched + + backfillRetryList *autoretry.List[MessageBatch] + backfillClosed int32 +} + +func (i *autoRetryInputBatchedBackfill) BackfillReadBatch(ctx context.Context) (MessageBatch, AckFunc, error) { + batch, rAckFn, err := i.backfillRetryList.Shift(ctx, atomic.LoadInt32(&i.backfillClosed) == 0) + if err != nil { + if errors.Is(err, autoretry.ErrExhausted) { + return nil, nil, ErrBackfillComplete + } + if errors.Is(err, ErrBackfillComplete) { + // Mark the backfill phase as closed and trigger an immediate + // re-read in order to clear any pending retries. + atomic.StoreInt32(&i.backfillClosed, 1) + return i.BackfillReadBatch(ctx) + } + return nil, nil, err + } + return batch.Copy(), AckFunc(rAckFn), nil +} + +func (i *autoRetryInputBatchedBackfill) BackfillComplete(ctx context.Context) error { + sc, ok := i.child.(BackfillCompleter) + if !ok { + return nil + } + return sc.BackfillComplete(ctx) +} + +func (i *autoRetryInputBatchedBackfill) Close(ctx context.Context) error { + _ = i.backfillRetryList.Close(ctx) + return i.autoRetryInputBatched.Close(ctx) +} diff --git a/public/service/input_auto_retry_batched_test.go b/public/service/input_auto_retry_batched_test.go index 05d8df3a6..e72600ee3 100644 --- a/public/service/input_auto_retry_batched_test.go +++ b/public/service/input_auto_retry_batched_test.go @@ -367,3 +367,70 @@ func TestBatchAutoRetryBuffer(t *testing.T) { require.NoError(t, err) assert.Equal(t, exp3, string(b)) } + +type snapshotAsyncBatchInput interface { + BackfillReadBatch(context.Context) (MessageBatch, AckFunc, error) +} + +func TestAutoRetryNacksBatchedWithoutSnapshotSupport(t *testing.T) { + i := &fnBatchInput{ + connect: func() error { return nil }, + read: func() (MessageBatch, AckFunc, error) { return nil, nil, ErrEndOfInput }, + } + + wrapped := AutoRetryNacksBatched(i) + + _, ok := wrapped.(snapshotAsyncBatchInput) + assert.False(t, ok, "AutoRetryNacksBatched must not add BackfillReadBatch when the wrapped input has no snapshot phase") +} + +func TestAutoRetryNacksBatchedRetriesNackedSnapshotBatch(t *testing.T) { + var mu sync.Mutex + delivered := 0 + + i := &fnSnapshotBatchInput{ + fnBatchInput: &fnBatchInput{ + connect: func() error { return nil }, + read: func() (MessageBatch, AckFunc, error) { return nil, nil, ErrEndOfInput }, + }, + snapshotRead: func() (MessageBatch, AckFunc, error) { + mu.Lock() + defer mu.Unlock() + if delivered >= 2 { + return nil, nil, ErrBackfillComplete + } + delivered++ + return MessageBatch{NewMessage([]byte("row"))}, func(ctx context.Context, err error) error { + return nil + }, nil + }, + snapshotComplete: func() error { return nil }, + } + + wrapped := AutoRetryNacksBatched(i) + sa, ok := wrapped.(snapshotAsyncBatchInput) + require.True(t, ok, "AutoRetryNacksBatched must forward BackfillReadBatch when the wrapped input implements SnapshotBatchInput") + + ctx, done := context.WithTimeout(t.Context(), 10*time.Second) + defer done() + + batch, ackFn, err := sa.BackfillReadBatch(ctx) + require.NoError(t, err) + require.Len(t, batch, 1) + + // Nack it - a dropped batch here would silently lose a snapshot row. + require.NoError(t, ackFn(ctx, errors.New("downstream rejected the row"))) + + seen := 0 + for { + _, aFn, err := sa.BackfillReadBatch(ctx) + if errors.Is(err, ErrBackfillComplete) { + break + } + require.NoError(t, err) + seen++ + require.NoError(t, aFn(ctx, nil)) + } + // The nacked batch replayed, plus the one fresh batch still pending = 2. + assert.Equal(t, 2, seen) +} diff --git a/public/service/input_test.go b/public/service/input_test.go index d1f735f73..a2894f509 100644 --- a/public/service/input_test.go +++ b/public/service/input_test.go @@ -12,6 +12,7 @@ import ( "github.com/stretchr/testify/require" "github.com/redpanda-data/benthos/v4/internal/component" + "github.com/redpanda-data/benthos/v4/internal/component/input" "github.com/redpanda-data/benthos/v4/internal/manager/mock" "github.com/redpanda-data/benthos/v4/internal/message" ) @@ -251,3 +252,73 @@ func TestBatchInputAirGapHappy(t *testing.T) { assert.NoError(t, outAckFn(t.Context(), errors.New("foobar"))) assert.EqualError(t, ackErr, "foobar") } + +type fnSnapshotBatchInput struct { + *fnBatchInput + snapshotRead func() (MessageBatch, AckFunc, error) + snapshotComplete func() error +} + +func (f *fnSnapshotBatchInput) BackfillReadBatch(ctx context.Context) (MessageBatch, AckFunc, error) { + return f.snapshotRead() +} + +func (f *fnSnapshotBatchInput) BackfillComplete(ctx context.Context) error { + return f.snapshotComplete() +} + +func TestBatchInputAirGapSnapshotHappy(t *testing.T) { + var snapshotAckErr error + snapshotAckFn := func(ctx context.Context, err error) error { + snapshotAckErr = err + return nil + } + + snapshotCompleteCalled := false + i := &fnSnapshotBatchInput{ + fnBatchInput: &fnBatchInput{ + connect: func() error { return nil }, + }, + snapshotRead: func() (MessageBatch, AckFunc, error) { + return MessageBatch{NewMessage([]byte("snapshot row"))}, snapshotAckFn, nil + }, + snapshotComplete: func() error { + snapshotCompleteCalled = true + return nil + }, + } + + agi := newAirGapBatchReader(mock.NewManager(), i) + + sa, ok := agi.(input.BackfillAsync) + require.True(t, ok, "airGapBatchReader must implement input.BackfillAsync when the wrapped BatchInput implements SnapshotBatchInput") + + outMsg, outAckFn, err := sa.BackfillReadBatch(t.Context()) + require.NoError(t, err) + assert.Equal(t, 1, outMsg.Len()) + assert.Equal(t, "snapshot row", string(outMsg.Get(0).AsBytes())) + + assert.NoError(t, outAckFn(t.Context(), errors.New("snapshot nack"))) + assert.EqualError(t, snapshotAckErr, "snapshot nack") + + i.snapshotRead = func() (MessageBatch, AckFunc, error) { + return nil, nil, ErrBackfillComplete + } + _, _, err = sa.BackfillReadBatch(t.Context()) + assert.Equal(t, component.ErrBackfillComplete, err) + + sc, ok := agi.(input.BackfillCompleter) + require.True(t, ok, "airGapBatchReader must implement input.BackfillCompleter when the wrapped BatchInput implements BackfillCompleter") + require.NoError(t, sc.BackfillComplete(t.Context())) + assert.True(t, snapshotCompleteCalled) +} + +func TestBatchInputAirGapWithoutSnapshot(t *testing.T) { + i := &fnBatchInput{ + connect: func() error { return nil }, + } + agi := newAirGapBatchReader(mock.NewManager(), i) + + _, ok := agi.(input.BackfillAsync) + assert.False(t, ok, "airGapBatchReader must not implement input.BackfillAsync when the wrapped BatchInput has no snapshot phase") +} From 693b8d05e86a2c2f10b0e33793709e094aeb6ea5 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Wed, 19 Aug 2026 12:38:16 +0100 Subject: [PATCH 2/2] add warning if partial flush is stuck waiting --- internal/component/input/async_reader.go | 29 ++++++++++++++----- .../input/async_reader_snapshot_test.go | 10 +++---- 2 files changed, 27 insertions(+), 12 deletions(-) diff --git a/internal/component/input/async_reader.go b/internal/component/input/async_reader.go index 9867f749b..8d79cb182 100644 --- a/internal/component/input/async_reader.go +++ b/internal/component/input/async_reader.go @@ -173,12 +173,27 @@ func (r *AsyncReader) loop() { // was cut short rather than receiving a real ack/nack, ok must be false // so the caller never calls BackfillComplete for unconfirmed rows. runBackfillPhase := func(sr BackfillAsync) (ok bool) { - var pendingBackfillAcks sync.WaitGroup - var interruptedByHardStop atomic.Bool + var ( + pendingAcks sync.WaitGroup + interruptedByHardStop atomic.Bool + ) defer func() { - r.mgr.Logger().Debug("Waiting for pending backfill acks to resolve.") - pendingBackfillAcks.Wait() - r.mgr.Logger().Debug("Pending backfill acks resolved.") + done := make(chan struct{}) + go func() { + pendingAcks.Wait() + close(done) + }() + + // A downstream batching policy with no period configured only + // flushes on count/byte_size/check, so trailing backfill acks + // can stall here indefinitely if nothing else fills the batch. + select { + case <-done: + case <-time.After(60 * time.Second): + r.mgr.Logger().Warn("Waiting on pending backfill acks for input %v; verify downstream batching policy has a period to flush partial batches.", r.typeStr) + <-done + } + if interruptedByHardStop.Load() { ok = false } @@ -237,9 +252,9 @@ func (r *AsyncReader) loop() { return false } - pendingBackfillAcks.Add(1) + pendingAcks.Add(1) go func(m message.Batch, aFn AsyncAckFn, rChan chan error) { - defer pendingBackfillAcks.Done() + defer pendingAcks.Done() var res error select { diff --git a/internal/component/input/async_reader_snapshot_test.go b/internal/component/input/async_reader_snapshot_test.go index 492c3e09b..aa3b4ce25 100644 --- a/internal/component/input/async_reader_snapshot_test.go +++ b/internal/component/input/async_reader_snapshot_test.go @@ -72,13 +72,13 @@ func (r *mockBackfillAsyncReader) BackfillReadBatch(ctx context.Context) (messag batch := message.Batch{message.NewPart(payload)} return batch, func(ctx context.Context, err error) error { - r.record("snapshot-ack-" + string(rune('0'+idx))) + r.record("backfill-ack-" + string(rune('0'+idx))) return nil }, nil } func (r *mockBackfillAsyncReader) BackfillComplete(ctx context.Context) error { - r.record("snapshot-complete") + r.record("backfill-complete") close(r.backfillCompleteCalled) return nil } @@ -159,9 +159,9 @@ func TestAsyncReaderSnapshotPhaseBarrier(t *testing.T) { events := readerImpl.Events() require.Len(t, events, 4) - assert.Equal(t, "snapshot-ack-1", events[0]) - assert.Equal(t, "snapshot-ack-0", events[1]) - assert.Equal(t, "snapshot-complete", events[2]) + assert.Equal(t, "backfill-ack-1", events[0]) + assert.Equal(t, "backfill-ack-0", events[1]) + assert.Equal(t, "backfill-complete", events[2]) assert.Equal(t, "stream-ack-0", events[3]) }