func (s *GroupTransactSession) PollFetches(ctx context.Context) Fetches {
	return s.cl.PollFetches(ctx)
}
PollFetches is a wrapper around Client.PollFetches, with the exact same
semantics. Please refer to that function's documentation.

It is invalid to call PollFetches concurrently with Begin or End.
package kgo
import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
type TransactionEndTry bool
const (
	TryAbort TransactionEndTry = false
	TryCommit TransactionEndTry = true
)
type GroupTransactSession struct {
	cl *Client
	cooperative bool
	failMu sync.Mutex
	revoked   bool
	revokedCh chan struct{}  
	lost      bool
	lostCh    chan struct{}  
}
func NewGroupTransactSession(opts ...Opt) (*GroupTransactSession, error) {
	s := &GroupTransactSession{
		revokedCh: make(chan struct{}),
		lostCh:    make(chan struct{}),
	}
	var noGroup error
	opts = append(opts, groupOpt{func(cfg *cfg) {
		if cfg.group == "" {
			cfg.seedBrokers = nil  
			noGroup = errors.New("missing required group")
			return
		}
		s.cooperative = cfg.cooperative()
		userRevoked := cfg.onRevoked
		cfg.onRevoked = func(ctx context.Context, cl *Client, rev map[string][]int32) {
			s.failMu.Lock()
			defer s.failMu.Unlock()
			if s.revoked {
				return
			}
			if s.cooperative && len(rev) == 0 && !s.revoked {
				cl.cfg.logger.Log(LogLevelInfo, "transact session in on_revoke with nothing to revoke; allowing next commit")
			} else {
				cl.cfg.logger.Log(LogLevelInfo, "transact session in on_revoke; aborting next commit if we are currently in a transaction")
				s.revoked = true
				close(s.revokedCh)
			}
			if userRevoked != nil {
				userRevoked(ctx, cl, rev)
			}
		}
		userLost := cfg.onLost
		cfg.onLost = func(ctx context.Context, cl *Client, lost map[string][]int32) {
			s.failMu.Lock()
			defer s.failMu.Unlock()
			if s.lost {
				return
			}
			cl.cfg.logger.Log(LogLevelInfo, "transact session in on_lost; aborting next commit if we are currently in a transaction")
			s.lost = true
			close(s.lostCh)
			if userLost != nil {
				userLost(ctx, cl, lost)
			} else if userRevoked != nil {
				userRevoked(ctx, cl, lost)
			}
		}
	}})
	cl, err := NewClient(opts...)
	if err != nil {
		if noGroup != nil {
			err = noGroup
		}
		return nil, err
	}
	s.cl = cl
	return s, nil
}
func (s *GroupTransactSession) Client() *Client {
	return s.cl
}
func (s *GroupTransactSession) Close() {
	s.cl.Close()
}