You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: not_batched_test.go
    path of file: ./repos/benthos/internal/component/output
    
    the code of the file until where you have to start completion: package output

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/benthosdev/benthos/v4/internal/batch"
	"github.com/benthosdev/benthos/v4/internal/component"
	"github.com/benthosdev/benthos/v4/internal/message"
)

type mockNBWriter struct {
	t           *testing.T
	written     []string
	errorOn     []string
	closeCalled bool
	closeChan   chan error
	mut         sync.Mutex
}

func (m *mockNBWriter) Connect(context.Context) error {
	return nil
}

func (m *mockNBWriter) WriteBatch(ctx context.Context, msg message.Batch) error {
	m.mut.Lock()
	defer m.mut.Unlock()

	m.t.Helper()
	assert.Equal(m.t, 1, msg.Len())
	return msg.Iter(func(i int, p *message.Part) error {
		for _, eOn := range m.errorOn {
			if eOn == string(p.AsBytes()) {
				return error
    
func TestNotBatchedBreakOutMessagesErrors(t *testing.T) {
	msg := func(c ...string) message.Batch {
		msg := message.QuickBatch(nil)
		for _, str := range c {
			msg = append(msg, message.NewPart([]byte(str)))
		}
		return msg
	}

	w := &mockNBWriter{t: t, errorOn: []string{"foo1", "foo3"}}
	out, err := NewAsyncWriter("foo", 1, w, component.NoopObservability())
	require.NoError(t, err)

	nbOut := OnlySinglePayloads(out)

	resChan := make(chan error)
	tChan := make(chan message.Transaction)
	require.NoError(t, nbOut.Consume(tChan))

	sourceMessage := msg("foo0", "foo1", "foo2", "foo3", "foo4")
	sortGroup, sourceMessage := message.NewSortGroup(sourceMessage)

	select {
	case tChan <- message.NewTransaction(sourceMessage, resChan):
	case <-time.After(time.Second):
		t.Fatal("timed out")
	}
	select {
	case res := <-resChan:
		err := res
		require.Error(t, err)

		var walkable *batch.Error
		require.ErrorAs(t, err, &walkable, "expected error to be walkable batch error")

		errs := map[int]string{}
		walkable.WalkPartsBySource(sortGroup, sourceMessage, func(i int, _ *message.Part, err error) bool {
			if err != nil {
				errs[i] = err.Error()
			}
			return true
		})
		assert.Equal(t, map[int]string{
			1: "test err",
			3: "test err",
		}, errs)
	case <-time.After(time.Second):
		t.Fatal("timed out")
	}

	ctx, done := context.WithTimeout(context.Background(), time.Second*30)
	defer done()

	nbOut.TriggerCloseNow()
	assert.NoError(t, nbOut.WaitForClose(ctx))
	assert.Equal(t, []string{
		"foo0", "foo2", "foo4",
	}, w.written)
}