Fix docs and handle phantom messages

Author: Jeffail

internal/impl/pure/processor_split.go

@@ -24,8 +24,7 @@ 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/processor.go

@@ -85,9 +85,12 @@ func (p *Processor) loop() {
 			continue
 		}
 
-		var batchErrMut sync.Mutex
-		var batchErr *batch.Error
-		var batchWG sync.WaitGroup
+		var (
+			errMut     sync.Mutex
+			batchErr   *batch.Error
+			generalErr error
+			batchWG    sync.WaitGroup
+		)
 
 		for _, b := range resultBatches {
 			var wgOnce sync.Once
@@ -96,15 +99,22 @@ func (p *Processor) loop() {
 
 			select {
 			case p.messagesOut <- message.NewTransactionFunc(tmpBatch, func(ctx context.Context, err error) error {
-				batchErrMut.Lock()
-				defer batchErrMut.Unlock()
-
 				if err != nil {
+					errMut.Lock()
+					defer errMut.Unlock()
+
 					if batchErr == nil {
 						batchErr = batch.NewError(sortBatch, err)
 					}
 					for _, m := range tmpBatch {
-						batchErr.Failed(sorter.GetIndex(m), err)
+						if bIndex := sorter.GetIndex(m); bIndex >= 0 {
+							batchErr.Failed(bIndex, err)
+						} else {
+							// We are unable to link this message with an origin
+							// and therefore we must provide a general
+							// batch-wide error instead.
+							generalErr = err
+						}
 					}
 				}
 
@@ -120,7 +130,9 @@ func (p *Processor) loop() {
 
 		batchWG.Wait()
 
-		if batchErr != nil {
+		if generalErr != nil {
+			_ = tran.Ack(closeNowCtx, generalErr)
+		} else if batchErr != nil {
 			_ = tran.Ack(closeNowCtx, batchErr)
 		} else {
 			_ = tran.Ack(closeNowCtx, nil)

internal/pipeline/processor_test.go

@@ -352,3 +352,111 @@ func TestProcessorMultiMsgsBatchError(t *testing.T) {
 		t.Error("Expected mockproc to have waited for close")
 	}
 }
+
+type mockPhantomProcessor struct {
+	hasClosedAsync    bool
+	hasWaitedForClose bool
+	mut               sync.Mutex
+}
+
+func (m *mockPhantomProcessor) ProcessBatch(ctx context.Context, msg message.Batch) ([]message.Batch, error) {
+	var msgs []message.Batch
+	for _, p := range msg {
+		tmpMsg := p.ShallowCopy()
+		tmpMsg.SetBytes(fmt.Appendf(nil, "%s test", p.AsBytes()))
+		msgs = append(msgs, message.Batch{tmpMsg})
+	}
+	msgs = append(msgs, message.Batch{
+		message.NewPart([]byte("phantom message")),
+	})
+	return msgs, nil
+}
+
+func (m *mockPhantomProcessor) Close(ctx context.Context) error {
+	m.mut.Lock()
+	m.hasClosedAsync = true
+	m.hasWaitedForClose = true
+	m.mut.Unlock()
+	return nil
+}
+
+func TestProcessorMultiMsgsBatchUnknownError(t *testing.T) {
+	ctx, done := context.WithTimeout(context.Background(), time.Second*30)
+	defer done()
+
+	mockProc := &mockPhantomProcessor{}
+	proc := pipeline.NewProcessor(mockProc)
+
+	tChan, resChan := make(chan message.Transaction), make(chan error)
+
+	require.NoError(t, proc.Consume(tChan))
+
+	_, inputBatch := message.NewSortGroup(message.Batch{
+		message.NewPart([]byte("foo")),
+		message.NewPart([]byte("bar")),
+		message.NewPart([]byte("baz")),
+	})
+
+	// Send message
+	select {
+	case tChan <- message.NewTransaction(inputBatch, resChan):
+	case <-time.After(time.Second):
+		t.Error("Timed out")
+	}
+
+	expMsgs := map[string]struct{}{
+		"foo test":        {},
+		"bar test":        {},
+		"baz test":        {},
+		"phantom message": {},
+	}
+
+	resFns := []func(context.Context, error) error{}
+
+	// Receive expected messages
+	for i := 0; i < 4; i++ {
+		select {
+		case procT, open := <-proc.TransactionChan():
+			require.True(t, open)
+
+			act := string(procT.Payload.Get(0).AsBytes())
+			if _, exists := expMsgs[act]; !exists {
+				t.Errorf("Unexpected result: %v", act)
+			} else {
+				delete(expMsgs, act)
+			}
+			resFns = append(resFns, procT.Ack)
+		case <-time.After(time.Second):
+			t.Error("Timed out")
+		}
+	}
+
+	assert.Empty(t, expMsgs)
+	require.Len(t, resFns, 4)
+
+	require.NoError(t, resFns[0](ctx, nil))
+	require.NoError(t, resFns[1](ctx, nil))
+	require.NoError(t, resFns[2](ctx, nil))
+	require.NoError(t, resFns[3](ctx, errors.New("oh no")))
+
+	// Receive overall ack
+	select {
+	case err, open := <-resChan:
+		require.True(t, open)
+		require.EqualError(t, err, "oh no")
+
+		var batchErr *batch.Error
+		require.False(t, errors.As(err, &batchErr))
+	case <-time.After(time.Second):
+		t.Error("Timed out")
+	}
+
+	proc.TriggerCloseNow()
+	require.NoError(t, proc.WaitForClose(ctx))
+	if !mockProc.hasClosedAsync {
+		t.Error("Expected mockproc to have closed asynchronously")
+	}
+	if !mockProc.hasWaitedForClose {
+		t.Error("Expected mockproc to have waited for close")
+	}
+}