func TestTxnEtl(t *testing.T) {
	t.Parallel()
	topic1, topic1Cleanup := tmpTopic(t)
	defer topic1Cleanup()
	errs := make(chan error)
	body := []byte(randsha())  
	go func() {
		cl, err := NewClient(
			WithLogger(BasicLogger(os.Stderr, testLogLevel, nil)),
			TransactionalID("p"+randsha()),
			TransactionTimeout(2*time.Minute),
		)
		if err != nil {
			panic(err)
		}
		defer cl.Close()
		var offsetsMu sync.Mutex
		offsets := make(map[int32]int64)
		partsUsed := make(map[int32]struct{})
		if err := cl.BeginTransaction(); err != nil {
			errs <- fmt.Errorf("unable to begin transaction: %v", err)
		}
		defer func() {
			if err := cl.Flush(context.Background()); err != nil {
				errs <- fmt.Errorf("unable to flush: %v", err)
			}
			if err := cl.EndTransaction(context.Background(), true); err != nil {
				errs <- fmt.Errorf("unable to end transaction: %v", err)
			}
		}()
		for i := 0; i < testRecordLimit; i++ {
			if i > 0 && i%10000 == 0 {
				if err := cl.Flush(context.Background()); err != nil {
					errs <- fmt.Errorf("unable to flush: %v", err)
				}
				for partition := range partsUsed {
					offsets[partition]++
				}
				if err := cl.EndTransaction(context.Background(), true); err != nil {
					errs <- fmt.Errorf("unable to end transaction: %v", err)
				}
				if err := cl.BeginTransaction(); err != nil {
					errs <- fmt.Errorf("unable to begin transaction: %v", err)
				}
			}
			myKey := []byte(strconv.Itoa(i))
			cl.Produce(
				context.Background(),
				&Record{
					Topic: topic1,
					Key:   myKey,
					Value: body,
				},
				func(r *Record, err error) {
					if err != nil {
						errs <- fmt.Errorf("unexpected produce err: %v", err)
					}
					if !bytes.Equal(r.Key, myKey) {
						errs <- fmt.Errorf("unexpected out of order key; got %s != exp %v", r.Key, myKey)
					}
					offsetsMu.Lock()
					current, ok := offsets[r.Partition]
					if ok && r.Offset != current+1 {
						errs <- fmt.Errorf("partition %d produced offsets out of order, got %d != exp %d", r.Partition, r.Offset, current+1)
					} else if !ok && r.Offset != 0 {
						errs <- fmt.Errorf("expected first produced record to partition to have offset 0, got %d", r.Offset)
					}
					offsets[r.Partition] = r.Offset
					partsUsed[r.Partition] = struct{}{}
					offsetsMu.Unlock()
				},
			)
		}
	}()
	for _, tc := range []struct {
		name     string
		balancer GroupBalancer
	}{
		{"roundrobin", RoundRobinBalancer()},
		{"range", RangeBalancer()},
		{"sticky", StickyBalancer()},
		{"cooperative-sticky", CooperativeStickyBalancer()},
	} {
		t.Run(tc.name, func(t *testing.T) {
			testChainETL(
				t,
				topic1,
				body,
				errs,
				true,
				tc.balancer,
			)
		})
	}
}
This test is identical to TestGroupETL but based around transactions.
package kgo
import (
	"bytes"
	"context"
	"fmt"
	"os"
	"strconv"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)