func MinVersions(versions *kversion.Versions) Opt {
	return clientOpt{func(cfg *cfg) { cfg.minVersions = versions }}
}
MinVersions sets the minimum Kafka version a request can be downgraded to,
overriding the default of the lowest version.

This option is useful if you are issuing requests that you absolutely do not
want to be downgraded; that is, if you are relying on features in newer
requests, and you are not sure if your brokers can handle those features.
By setting a min version, if the client detects it needs to downgrade past
the version, it will instead avoid issuing the request.

Unlike MaxVersions, if a request is issued that is unknown to the min
versions, the request is allowed. It is assumed that there is no lower bound
for that request.
package kgo
import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"math"
	"math/rand"
	"net"
	"regexp"
	"runtime/debug"
	"sync"
	"time"
	"github.com/twmb/franz-go/pkg/kmsg"
	"github.com/twmb/franz-go/pkg/kversion"
	"github.com/twmb/franz-go/pkg/sasl"
)
type Opt interface {
	apply(*cfg)
}
type ProducerOpt interface {
	Opt
	producerOpt()
}
type ConsumerOpt interface {
	Opt
	consumerOpt()
}
type GroupOpt interface {
	Opt
	groupOpt()
}
type (
	clientOpt   struct{ fn func(*cfg) }
	producerOpt struct{ fn func(*cfg) }
	consumerOpt struct{ fn func(*cfg) }
	groupOpt    struct{ fn func(*cfg) }
)
func (opt clientOpt) apply(cfg *cfg)   { opt.fn(cfg) }
func (opt producerOpt) apply(cfg *cfg) { opt.fn(cfg) }
func (opt consumerOpt) apply(cfg *cfg) { opt.fn(cfg) }
func (opt groupOpt) apply(cfg *cfg)    { opt.fn(cfg) }
func (producerOpt) producerOpt()       {}
func (consumerOpt) consumerOpt()       {}
func (groupOpt) groupOpt()             {}
type cfg struct {
	id                     *string  
	dialFn                 func(context.Context, string, string) (net.Conn, error)
	requestTimeoutOverhead time.Duration
	connIdleTimeout        time.Duration
	softwareName    string  
	softwareVersion string  
	logger Logger
	seedBrokers []string
	maxVersions *kversion.Versions
	minVersions *kversion.Versions
	retryBackoff func(int) time.Duration
	retries      int64
	retryTimeout func(int16) time.Duration
	maxBrokerWriteBytes int32
	maxBrokerReadBytes  int32
	allowAutoTopicCreation bool
	metadataMaxAge time.Duration
	metadataMinAge time.Duration
	sasls []sasl.Mechanism
	hooks hooks
	txnID              *string
	txnTimeout         time.Duration
	acks               Acks
	disableIdempotency bool
	compression        []CompressionCodec  
	defaultProduceTopic string
	maxRecordBatchBytes int32
	maxBufferedRecords  int64
	produceTimeout      time.Duration
	recordRetries       int64
	linger              time.Duration
	recordTimeout       time.Duration
	manualFlushing      bool
	partitioner Partitioner
	stopOnDataLoss bool
	onDataLoss     func(string, int32)
	maxWait        int32
	minBytes       int32
	maxBytes       int32
	maxPartBytes   int32
	resetOffset    Offset
	isolationLevel int8
	keepControl    bool
	rack           string
	maxConcurrentFetches int
	topics     map[string]*regexp.Regexp    
	partitions map[string]map[int32]Offset  
	regex      bool
	group      string           
	instanceID *string          
	balancers  []GroupBalancer  
	protocol   string           
	sessionTimeout    time.Duration
	rebalanceTimeout  time.Duration
	heartbeatInterval time.Duration
	requireStable     bool
	onAssigned func(context.Context, *Client, map[string][]int32)
	onRevoked  func(context.Context, *Client, map[string][]int32)
	onLost     func(context.Context, *Client, map[string][]int32)
	setAssigned       bool
	setRevoked        bool
	setLost           bool
	setCommitCallback bool
	autocommitDisable  bool  
	autocommitGreedy   bool
	autocommitMarks    bool
	autocommitInterval time.Duration
	commitCallback     func(*Client, *kmsg.OffsetCommitRequest, *kmsg.OffsetCommitResponse, error)
}
func (cfg *cfg) cooperative() bool {
	cooperative := true
	for _, balancer := range cfg.balancers {
		cooperative = cooperative && balancer.IsCooperative()
	}
	return cooperative
}
func (cfg *cfg) validate() error {
	if len(cfg.seedBrokers) == 0 {
		return errors.New("config erroneously has no seed brokers")
	}
	if cfg.maxPartBytes > cfg.maxBytes {
		cfg.maxPartBytes = cfg.maxBytes
	}
	if cfg.disableIdempotency && cfg.txnID != nil {
		return errors.New("cannot both disable idempotent writes and use transactional IDs")
	}
	if !cfg.disableIdempotency && cfg.acks.val != -1 {
		return errors.New("idempotency requires acks=all")
	}
	for _, limit := range []struct {
		name    string
		sp      **string  
		s       string
		allowed int
	}{
		{name: "client id", sp: &cfg.id, allowed: 256},
		{name: "software name", s: cfg.softwareName, allowed: 256},
		{name: "software version", s: cfg.softwareVersion, allowed: 256},
		{name: "transactional id", sp: &cfg.txnID, allowed: 16382},
		{name: "rack", s: cfg.rack, allowed: 512},
	} {
		s := limit.s
		if limit.sp != nil && *limit.sp != nil {
			s = **limit.sp
		}
		if len(s) > limit.allowed {
			return fmt.Errorf("%s length %d is larger than max allowed %d", limit.name, len(s), limit.allowed)
		}
	}
	i64lt := func(l, r int64) (bool, string) { return l < r, "less" }
	i64gt := func(l, r int64) (bool, string) { return l > r, "larger" }
	for _, limit := range []struct {
		name    string
		v       int64
		allowed int64
		badcmp  func(int64, int64) (bool, string)
		fmt  string
		durs bool
	}{
		{name: "max broker write bytes", v: int64(cfg.maxBrokerWriteBytes), allowed: 1 << 10, badcmp: i64lt},
		{name: "max broker write bytes", v: int64(cfg.maxBrokerWriteBytes), allowed: 1 << 30, badcmp: i64gt},
		{name: "max broker read bytes", v: int64(cfg.maxBrokerReadBytes), allowed: 1 << 10, badcmp: i64lt},
		{name: "max broker read bytes", v: int64(cfg.maxBrokerReadBytes), allowed: 1 << 30, badcmp: i64gt},
		{name: "max record batch bytes", v: int64(cfg.maxRecordBatchBytes), allowed: 512, badcmp: i64lt},
		{name: "max record batch bytes", v: int64(cfg.maxRecordBatchBytes), allowed: 268435454, badcmp: i64gt},
		{v: int64(cfg.maxBrokerWriteBytes), allowed: int64(cfg.maxRecordBatchBytes), badcmp: i64lt, fmt: "max broker write bytes %v is erroneously less than max record batch bytes %v"},
		{v: int64(cfg.maxBrokerReadBytes), allowed: int64(cfg.maxBytes), badcmp: i64lt, fmt: "max broker read bytes %v is erroneously less than max fetch bytes %v"},
		{name: "max concurrent fetches", v: int64(cfg.maxConcurrentFetches), allowed: 0, badcmp: i64lt},
		{name: "request timeout max overhead", v: int64(cfg.requestTimeoutOverhead), allowed: int64(15 * time.Minute), badcmp: i64gt, durs: true},
		{name: "request timeout min overhead", v: int64(cfg.requestTimeoutOverhead), allowed: int64(time.Second), badcmp: i64lt, durs: true},
		{name: "conn min idle timeout", v: int64(cfg.connIdleTimeout), allowed: int64(time.Second), badcmp: i64lt, durs: true},
		{name: "conn max idle timeout", v: int64(cfg.connIdleTimeout), allowed: int64(15 * time.Minute), badcmp: i64gt, durs: true},
		{name: "metadata max age", v: int64(cfg.metadataMaxAge), allowed: int64(time.Hour), badcmp: i64gt, durs: true},
		{name: "metadata min age", v: int64(cfg.metadataMinAge), allowed: int64(10 * time.Millisecond), badcmp: i64lt, durs: true},
		{v: int64(cfg.metadataMaxAge), allowed: int64(cfg.metadataMinAge), badcmp: i64lt, fmt: "metadata max age %v is erroneously less than metadata min age %v", durs: true},
		{name: "max buffered records", v: int64(cfg.maxBufferedRecords), allowed: 1, badcmp: i64lt},
		{name: "linger", v: int64(cfg.linger), allowed: int64(time.Minute), badcmp: i64gt, durs: true},
		{name: "produce timeout", v: int64(cfg.produceTimeout), allowed: int64(100 * time.Millisecond), badcmp: i64lt, durs: true},
		{name: "record timeout", v: int64(cfg.recordTimeout), allowed: int64(time.Second), badcmp: func(l, r int64) (bool, string) {
			if l == 0 {
				return false, ""  
			}
			return l < r, "less"
		}, durs: true},
		{name: "max fetch wait", v: int64(cfg.maxWait) * int64(time.Millisecond), allowed: int64(10 * time.Millisecond), badcmp: i64lt, durs: true},
		{name: "number of balancers", v: int64(len(cfg.balancers)), allowed: 1, badcmp: i64lt},
		{name: "consumer protocol length", v: int64(len(cfg.protocol)), allowed: 1, badcmp: i64lt},
		{name: "session timeout", v: int64(cfg.sessionTimeout), allowed: int64(100 * time.Millisecond), badcmp: i64lt, durs: true},
		{name: "rebalance timeout", v: int64(cfg.rebalanceTimeout), allowed: int64(100 * time.Millisecond), badcmp: i64lt, durs: true},
		{name: "autocommit interval", v: int64(cfg.autocommitInterval), allowed: int64(100 * time.Millisecond), badcmp: i64lt, durs: true},
		{v: int64(cfg.heartbeatInterval), allowed: int64(cfg.rebalanceTimeout) * int64(time.Millisecond), badcmp: i64gt, durs: true, fmt: "heartbeat interval %v is erroneously larger than the session timeout %v"},
	} {
		bad, cmp := limit.badcmp(limit.v, limit.allowed)
		if bad {
			if limit.fmt != "" {
				if limit.durs {
					return fmt.Errorf(limit.fmt, time.Duration(limit.v), time.Duration(limit.allowed))
				}
				return fmt.Errorf(limit.fmt, limit.v, limit.allowed)
			}
			if limit.durs {
				return fmt.Errorf("%s %v is %s than allowed %v", limit.name, time.Duration(limit.v), cmp, time.Duration(limit.allowed))
			}
			return fmt.Errorf("%s %v is %s than allowed %v", limit.name, limit.v, cmp, limit.allowed)
		}
	}
	if len(cfg.group) > 0 {
		if len(cfg.topics) == 0 {
			return errors.New("unable to consume from a group when no topics are specified")
		}
		if len(cfg.partitions) != 0 {
			return errors.New("invalid direct-partition consuming option when consuming as a group")
		}
	}
	if cfg.regex {
		if len(cfg.partitions) != 0 {
			return errors.New("invalid direct-partition consuming option when consuming as regex")
		}
		for re := range cfg.topics {
			compiled, err := regexp.Compile(re)
			if err != nil {
				return fmt.Errorf("invalid regular expression %q", re)
			}
			cfg.topics[re] = compiled
		}
	}
	if cfg.autocommitDisable && cfg.autocommitGreedy {
		return errors.New("cannot both disable autocommitting and enable greedy autocommitting")
	}
	if cfg.autocommitDisable && cfg.autocommitMarks {
		return errors.New("cannot both disable autocommitting and enable marked autocommitting")
	}
	if cfg.autocommitGreedy && cfg.autocommitMarks {
		return errors.New("cannot enable both greedy autocommitting and marked autocommitting")
	}
	if (cfg.autocommitGreedy || cfg.autocommitDisable || cfg.autocommitMarks || cfg.setCommitCallback) && len(cfg.group) == 0 {
		return errors.New("invalid autocommit options specified when a group was not specified")
	}
	if (cfg.setLost || cfg.setRevoked || cfg.setAssigned) && len(cfg.group) == 0 {
		return errors.New("invalid group partition assigned/revoked/lost functions set when a group was not specified")
	}
	return nil
}
var (
	defaultDialer = &net.Dialer{Timeout: 10 * time.Second}
	reVersion     = regexp.MustCompile(`^[a-zA-Z0-9](?:[a-zA-Z0-9\.-]*[a-zA-Z0-9])?$`)
)
func softwareVersion() string {
	info, ok := debug.ReadBuildInfo()
	if ok {
		for _, dep := range info.Deps {
			if dep.Path == "github.com/twmb/franz-go" {
				if reVersion.MatchString(dep.Version) {
					return dep.Version
				}
			}
		}
	}
	return "unknown"
}
func defaultCfg() cfg {
	defaultID := "kgo"
	return cfg{
		id:     &defaultID,
		dialFn: defaultDialer.DialContext,
		requestTimeoutOverhead: 10 * time.Second,
		connIdleTimeout:        20 * time.Second,
		softwareName:    "kgo",
		softwareVersion: softwareVersion(),
		logger: new(nopLogger),
		seedBrokers: []string{"127.0.0.1"},
		maxVersions: kversion.Stable(),
		retryBackoff: func() func(int) time.Duration {
			var rngMu sync.Mutex
			rng := rand.New(rand.NewSource(time.Now().UnixNano()))
			return func(fails int) time.Duration {
				const (
					min = 250 * time.Millisecond
					max = 5 * time.Second / 2
				)
				if fails <= 0 {
					return min
				}
				if fails > 10 {
					return max
				}
				backoff := min * time.Duration(1<<(fails-1))
				rngMu.Lock()
				jitter := 0.8 + 0.4*rng.Float64()
				rngMu.Unlock()
				backoff = time.Duration(float64(backoff) * jitter)
				if backoff > max {
					return max
				}
				return backoff
			}
		}(),
		retries: 20,
		maxBrokerWriteBytes: 100 << 20,  
		maxBrokerReadBytes:  100 << 20,
		metadataMaxAge: 5 * time.Minute,
		metadataMinAge: 10 * time.Second,
		txnTimeout:          40 * time.Second,
		acks:                AllISRAcks(),
		compression:         []CompressionCodec{SnappyCompression(), NoCompression()},
		maxRecordBatchBytes: 1000000,  
		maxBufferedRecords:  10000,
		produceTimeout:      10 * time.Second,
		recordRetries:       math.MaxInt64,              
		partitioner:         StickyKeyPartitioner(nil),  
		maxWait:        5000,
		minBytes:       1,
		maxBytes:       50 << 20,
		maxPartBytes:   10 << 20,
		resetOffset:    NewOffset().AtStart(),
		isolationLevel: 0,
		maxConcurrentFetches: 0,  
		balancers: []GroupBalancer{
			CooperativeStickyBalancer(),
		},
		protocol: "consumer",
		sessionTimeout:    45000 * time.Millisecond,
		rebalanceTimeout:  60000 * time.Millisecond,
		heartbeatInterval: 3000 * time.Millisecond,
		autocommitInterval: 5 * time.Second,
	}
}
func ClientID(id string) Opt {
	return clientOpt{func(cfg *cfg) { cfg.id = &id }}
}
func SoftwareNameAndVersion(name, version string) Opt {
	return clientOpt{func(cfg *cfg) { cfg.softwareName = name; cfg.softwareVersion = version }}
}
func WithLogger(l Logger) Opt {
	return clientOpt{func(cfg *cfg) { cfg.logger = &wrappedLogger{l} }}
}
func RequestTimeoutOverhead(overhead time.Duration) Opt {
	return clientOpt{func(cfg *cfg) { cfg.requestTimeoutOverhead = overhead }}
}
func ConnIdleTimeout(timeout time.Duration) Opt {
	return clientOpt{func(cfg *cfg) { cfg.connIdleTimeout = timeout }}
}
func Dialer(fn func(ctx context.Context, network, host string) (net.Conn, error)) Opt {
	return clientOpt{func(cfg *cfg) { cfg.dialFn = fn }}
}
func DialTLSConfig(c *tls.Config) Opt {
	return Dialer(func(ctx context.Context, network, host string) (net.Conn, error) {
		c := c.Clone()
		if c.ServerName == "" {
			server, _, err := net.SplitHostPort(host)
			if err != nil {
				return nil, fmt.Errorf("unable to split host:port for dialing: %w", err)
			}
			c.ServerName = server
		}
		return (&tls.Dialer{
			NetDialer: defaultDialer,
			Config:    c,
		}).DialContext(ctx, network, host)
	})
}
func SeedBrokers(seeds ...string) Opt {
	return clientOpt{func(cfg *cfg) { cfg.seedBrokers = append(cfg.seedBrokers[:0], seeds...) }}
}
func MaxVersions(versions *kversion.Versions) Opt {
	return clientOpt{func(cfg *cfg) { cfg.maxVersions = versions }}
}