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: commit_test.go
    path of file: ./repos/goka/systemtest
    
    the code of the file until where you have to start completion: package systemtest

import (
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/IBM/sarama"
	"github.com/lovoo/goka"
	"github.com/lovoo/goka/codec"
	"github.com/stretchr/testify/require"
)

// TestAutoCommit tests/demonstrates the behavior of disabling the auto-commit functionality.
// The autocommiter sends the offsets of the marked messages to the broker regularily. If the processor shuts down
// (or the group rebalances), the offsets are sent one last time, so just turning it of is not enough.
// To get a processor to not commit any offsets, we're using a fault-injecting proxy
    
func TestUnmarkedMessages(t *testing.T) {
	var (
		group       goka.Group = goka.Group(fmt.Sprintf("%s-%d", "goka-mark-test", time.Now().Unix()))
		inputStream            = goka.Stream(group) + "-input"
		brokers                = initSystemTest(t)
	)

	// make sure we consume everything
	cfg := goka.DefaultConfig()
	cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
	goka.ReplaceGlobalConfig(cfg)

	tmgr, err := goka.DefaultTopicManagerBuilder(brokers)
	require.NoError(t, err)
	require.NoError(t, tmgr.EnsureStreamExists(string(inputStream), 10))
	require.NoError(t, tmgr.Close())

	var values []int64

	// emit exactly one message
	em, err := goka.NewEmitter(brokers, inputStream, new(codec.Int64))
	require.NoError(t, err)
	require.NoError(t, em.EmitSync("key", int64(1)))
	require.NoError(t, em.EmitSync("key", int64(2)))
	require.NoError(t, em.Finish())

	createProc := func() *goka.Processor {
		proc, err := goka.NewProcessor(brokers, goka.DefineGroup(group,
			goka.Input(inputStream, new(codec.Int64), func(ctx goka.Context, msg interface{}) {
				val := msg.(int64)
				values = append(values, val)

				// the only way to not commit a message is to fail the processor.
				// We'll fail after the second message
				if val == 2 {
					ctx.Fail(errors.New("test"))
				}
			}),
		))

		require.NoError(t, err)
		return proc
	}

	// run the first processor
	runProc(createProc())
	pollTimed(t, "all-received1", func() bool {
		return len(values) == 2 && values[0] == 1
	})

	// reset values
	values = nil

	// restart -> we'll only receive the second message
	runProc(createProc())
	pollTimed(t, "all-received2", func() bool {
		return len(values) == 1 && values[0] == 2
	})
}