Return tracked batch errors on nacked split batches

Author: Jeffail

CHANGELOG.md

@@ -26,6 +26,7 @@ All notable changes to this project will be documented in this file.
 ### Changed
 
 - The log events from all inputs and outputs when they first connect have been made more consistent and no longer contain any information regarding the nature of their connections.
+- Splitting message batches with a `split` processor (or custom plugins) no longer results in downstream error handling loops around nacks. This was previously implemented as a feature to ensure unbounded expanded and split batches don't flood downstream services in the event of a minority of errors. However, introducing more clever origin tracking of errored messages has eliminated the need for this undocumented behaviour.
 
 ## 4.26.0 - 2024-03-18
 

internal/impl/pure/processor_split.go

@@ -24,7 +24,8 @@ func init() {
 			Description(`
 This processor is for breaking batches down into smaller ones. In order to break a single message out into multiple messages use the `+"[`unarchive` processor](/docs/components/processors/unarchive)"+`.
 
-If there is a remainder of messages after splitting a batch the remainder is also sent as a single batch. For example, if your target size was 10, and the processor received a batch of 95 message parts, the result would be 9 batches of 10 messages followed by a batch of 5 messages.`).
+If there is a remainder of messages after splitting a batch the remainder is also sent as a single batch. For example, if your target size was 10, and the processor received a batch of 95 message parts, the result would be 9 batches of 10 messages followed by a batch of 5 messages.
+`).
 			Fields(
 				service.NewIntField(splitPFieldSize).
 					Description("The target number of messages.").

internal/pipeline/pool_test.go

@@ -2,7 +2,6 @@ package pipeline_test
 
 import (
 	"context"
-	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -119,32 +118,35 @@ func TestPoolMultiMsgs(t *testing.T) {
 	ctx, done := context.WithTimeout(context.Background(), time.Second*30)
 	defer done()
 
-	mockProc := &mockMultiMsgProcessor{N: 3}
+	mockProc := &mockSplitProcessor{}
 
 	proc, err := pipeline.NewPool(1, log.Noop(), mockProc)
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	tChan, resChan := make(chan message.Transaction), make(chan error)
 	if err := proc.Consume(tChan); err != nil {
 		t.Fatal(err)
 	}
 
 	for j := 0; j < 10; j++ {
-		expMsgs := map[string]struct{}{}
-		for i := 0; i < mockProc.N; i++ {
-			expMsgs[fmt.Sprintf("test%v", i)] = struct{}{}
+		expMsgs := map[string]struct{}{
+			"foo test": {},
+			"bar test": {},
+			"baz test": {},
 		}
 
 		// Send message
 		select {
-		case tChan <- message.NewTransaction(message.QuickBatch(nil), resChan):
+		case tChan <- message.NewTransaction(message.Batch{
+			message.NewPart([]byte(`foo`)),
+			message.NewPart([]byte(`bar`)),
+			message.NewPart([]byte(`baz`)),
+		}, resChan):
 		case <-time.After(time.Second * 5):
 			t.Fatal("Timed out")
 		}
 
-		for i := 0; i < mockProc.N; i++ {
+		for i := 0; i < 3; i++ {
 			// Receive messages
 			var procT message.Transaction
 			var open bool

internal/pipeline/processor.go

@@ -4,10 +4,10 @@ import (
 	"context"
 	"sync"
 
+	"github.com/benthosdev/benthos/v4/internal/batch"
 	"github.com/benthosdev/benthos/v4/internal/component"
 	"github.com/benthosdev/benthos/v4/internal/component/processor"
 	"github.com/benthosdev/benthos/v4/internal/message"
-	"github.com/benthosdev/benthos/v4/internal/old/util/throttle"
 	"github.com/benthosdev/benthos/v4/internal/shutdown"
 )
 
@@ -66,79 +66,66 @@ func (p *Processor) loop() {
 			return
 		}
 
-		resultMsgs, resultRes := processor.ExecuteAll(closeNowCtx, p.msgProcessors, tran.Payload)
-		if len(resultMsgs) == 0 {
-			if err := tran.Ack(closeNowCtx, resultRes); err != nil && closeNowCtx.Err() != nil {
+		sorter, sortBatch := message.NewSortGroup(tran.Payload)
+
+		resultBatches, err := processor.ExecuteAll(closeNowCtx, p.msgProcessors, sortBatch)
+		if len(resultBatches) == 0 || err != nil {
+			if _ = tran.Ack(closeNowCtx, err); closeNowCtx.Err() != nil {
 				return
 			}
 			continue
 		}
 
-		if len(resultMsgs) > 1 {
-			p.dispatchMessages(closeNowCtx, resultMsgs, tran.Ack)
-		} else {
+		if len(resultBatches) == 1 {
 			select {
-			case p.messagesOut <- message.NewTransactionFunc(resultMsgs[0], tran.Ack):
+			case p.messagesOut <- message.NewTransactionFunc(resultBatches[0], tran.Ack):
 			case <-p.shutSig.CloseNowChan():
 				return
 			}
+			continue
 		}
-	}
-}
 
-// dispatchMessages attempts to send a multiple messages results of processors
-// over the shared messages channel. This send is retried until success.
-func (p *Processor) dispatchMessages(ctx context.Context, msgs []message.Batch, ackFn func(context.Context, error) error) {
-	throt := throttle.New(throttle.OptCloseChan(p.shutSig.CloseAtLeisureChan()))
+		var batchErrMut sync.Mutex
+		var batchErr *batch.Error
+		var batchWG sync.WaitGroup
 
-	pending := msgs
-	for len(pending) > 0 {
-		doneChan := make(chan struct{}, len(pending))
+		for _, b := range resultBatches {
+			var wgOnce sync.Once
+			batchWG.Add(1)
+			tmpBatch := b.ShallowCopy()
 
-		var newPending []message.Batch
-		var newPendingMut sync.Mutex
+			select {
+			case p.messagesOut <- message.NewTransactionFunc(tmpBatch, func(ctx context.Context, err error) error {
+				batchErrMut.Lock()
+				defer batchErrMut.Unlock()
 
-		for _, b := range pending {
-			b := b
-			transac := message.NewTransactionFunc(b.ShallowCopy(), func(ctx context.Context, err error) error {
 				if err != nil {
-					newPendingMut.Lock()
-					newPending = append(newPending, b)
-					newPendingMut.Unlock()
-				}
-				select {
-				case doneChan <- struct{}{}:
-				default:
+					if batchErr == nil {
+						batchErr = batch.NewError(sortBatch, err)
+					}
+					for _, m := range tmpBatch {
+						batchErr.Failed(sorter.GetIndex(m), err)
+					}
 				}
-				return nil
-			})
 
-			select {
-			case p.messagesOut <- transac:
-			case <-ctx.Done():
-				select {
-				case doneChan <- struct{}{}:
-				default:
-				}
+				wgOnce.Do(func() {
+					batchWG.Done()
+				})
+				return nil
+			}):
+			case <-p.shutSig.CloseNowChan():
 				return
 			}
 		}
 
-		for i := 0; i < len(pending); i++ {
-			select {
-			case <-doneChan:
-			case <-ctx.Done():
-				return
-			}
-		}
+		batchWG.Wait()
 
-		if pending = newPending; len(pending) > 0 && !throt.Retry() {
-			return
+		if batchErr != nil {
+			_ = tran.Ack(closeNowCtx, batchErr)
+		} else {
+			_ = tran.Ack(closeNowCtx, nil)
 		}
 	}
-
-	throt.Reset()
-	_ = ackFn(ctx, nil)
 }
 
 //------------------------------------------------------------------------------