kgo.Client: add ConfigValue and ConfigValues

This allows a way to interrogate the client for the value of any option,
whether the option was specified or not.

More type safe and explicit would be to add a getter for every option,
but that'd also bloat the client API unnecessarily.

pkg/kgo/client.go

@@ -16,6 +16,7 @@ import (
 	"math/rand"
 	"net"
 	"reflect"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -153,6 +154,246 @@ func validateCfg(opts ...Opt) (cfg, []hostport, *compressor, error) {
 	return cfg, seeds, compressor, nil
 }
 
+func namefn(fn any) string {
+	v := reflect.ValueOf(fn)
+	if v.Type().Kind() != reflect.Func {
+		return ""
+	}
+	name := runtime.FuncForPC(v.Pointer()).Name()
+	dot := strings.LastIndexByte(name, '.')
+	if dot >= 0 {
+		return name[dot+1:]
+	}
+	return name
+}
+
+// OptValue returns the value for the given configuration option. If the
+// given option does not exist, this returns nil. This function takes either a
+// raw Opt, or an Opt function name.
+//
+// If a configuration option has multiple inputs, this function returns only
+// the first input. If the function is a boolean function (such as
+// BlockRebalanceOnPoll), this function returns the value of the internal bool.
+// Variadic option inputs are returned as a single slice.  Options that are
+// internally stored as a pointer (ClientID, TransactionalID, and InstanceID)
+// are returned as their string input; you can see if the option is internally
+// nil by looking at the second value returned from OptValues.
+//
+//	var (
+//		cl, _ := NewClient(
+//			InstanceID("foo"),
+//			ConsumeTopics("foo", "bar"),
+//		)
+//		iid    = cl.OptValue(InstanceID)           // iid is "foo"
+//		gid    = cl.OptValue(ConsumerGroup)        // gid is "" since groups are not used
+//		topics = cl.OptValue("ConsumeTopics")      // topics is []string{"foo", "bar"}; string lookup for the option works
+//		bpoll  = cl.OptValue(BlockRebalanceOnPoll) // bpoll is false
+//		t      = cl.OptValue(SessionTimeout)       // t is 45s, the internal default
+//		td     = t.(time.Duration)                 // safe conversion since SessionTimeout's input is a time.Duration
+//		unk    = cl.OptValue("Unknown"),           // unk is nil
+//	)
+func (cl *Client) OptValue(opt any) any {
+	vs := cl.OptValues(opt)
+	if len(vs) > 0 {
+		return vs[0]
+	}
+	return nil
+}
+
+// OptValues returns all values for options. This method is useful for
+// options that have multiple inputs (notably, SoftwareNameAndVersion). This is
+// also useful for options that are internally stored as a pointer (ClientID,
+// TransactionalID, and InstanceID) -- this function will return the string
+// value of the option but also whether the option is non-nil. Boolean options
+// are returned as a single-element slice with the bool value. Variadic inputs
+// are returned as a signle slice. If the input option does not exist, this
+// returns nil.
+//
+//	var (
+//		cl, _    = NewClient(
+//			InstanceID("foo"),
+//			ConsumeTopics("foo", "bar"),
+//		)
+//		idValues = cl.OptValues(InstanceID)          // idValues is []any{"foo", true}
+//		tValues  = cl.OptValues(SessionTimeout)      // tValues is []any{45 * time.Second}
+//		topics   = cl.OptValues(ConsumeTopics)       // topics is []any{[]string{"foo", "bar"}
+//		bpoll    = cl.OptValues(BlockRebalanceOnPoll) // bpoll is []any{false}
+//		unknown  = cl.OptValues("Unknown")           // unknown is nil
+//	)
+func (cl *Client) OptValues(opt any) []any {
+	name := namefn(opt)
+	if s, ok := opt.(string); ok {
+		name = s
+	}
+	cfg := &cl.cfg
+
+	switch name {
+	case namefn(ClientID):
+		if cfg.id != nil {
+			return []any{*cfg.id, true}
+		}
+		return []any{"", false}
+	case namefn(SoftwareNameAndVersion):
+		return []any{cfg.softwareName, cfg.softwareVersion}
+	case namefn(WithLogger):
+		if cfg.logger != nil {
+			return []any{cfg.logger.(*wrappedLogger).inner}
+		}
+		return []any{nil}
+	case namefn(RequestTimeoutOverhead):
+		return []any{cfg.requestTimeoutOverhead}
+	case namefn(ConnIdleTimeout):
+		return []any{cfg.connIdleTimeout}
+	case namefn(Dialer):
+		return []any{cfg.dialFn}
+	case namefn(DialTLSConfig):
+		return []any{cfg.dialTLS}
+	case namefn(SeedBrokers):
+		return []any{cfg.seedBrokers}
+	case namefn(MaxVersions):
+		return []any{cfg.maxVersions}
+	case namefn(MinVersions):
+		return []any{cfg.minVersions}
+	case namefn(RetryBackoffFn):
+		return []any{cfg.retryBackoff}
+	case namefn(RequestRetries):
+		return []any{cfg.retries}
+	case namefn(RetryTimeout):
+		return []any{cfg.retryTimeout(0)}
+	case namefn(RetryTimeoutFn):
+		return []any{cfg.retryTimeout}
+	case namefn(AllowAutoTopicCreation):
+		return []any{cfg.allowAutoTopicCreation}
+	case namefn(BrokerMaxWriteBytes):
+		return []any{cfg.maxBrokerWriteBytes}
+	case namefn(BrokerMaxReadBytes):
+		return []any{cfg.maxBrokerReadBytes}
+	case namefn(MetadataMaxAge):
+		return []any{cfg.metadataMaxAge}
+	case namefn(MetadataMinAge):
+		return []any{cfg.metadataMinAge}
+	case namefn(SASL):
+		return []any{cfg.sasls}
+	case namefn(WithHooks):
+		return []any{cfg.hooks}
+	case namefn(ConcurrentTransactionsBackoff):
+		return []any{cfg.txnBackoff}
+
+	case namefn(DefaultProduceTopic):
+		return []any{cfg.defaultProduceTopic}
+	case namefn(RequiredAcks):
+		return []any{cfg.acks}
+	case namefn(DisableIdempotentWrite):
+		return []any{cfg.disableIdempotency}
+	case namefn(MaxProduceRequestsInflightPerBroker):
+		return []any{cfg.maxProduceInflight}
+	case namefn(ProducerBatchCompression):
+		return []any{cfg.compression}
+	case namefn(ProducerBatchMaxBytes):
+		return []any{cfg.maxRecordBatchBytes}
+	case namefn(MaxBufferedRecords):
+		return []any{cfg.maxBufferedRecords}
+	case namefn(RecordPartitioner):
+		return []any{cfg.partitioner}
+	case namefn(ProduceRequestTimeout):
+		return []any{cfg.produceTimeout}
+	case namefn(RecordRetries):
+		return []any{cfg.recordRetries}
+	case namefn(UnknownTopicRetries):
+		return []any{cfg.maxUnknownFailures}
+	case namefn(StopProducerOnDataLossDetected):
+		return []any{cfg.stopOnDataLoss}
+	case namefn(ProducerOnDataLossDetected):
+		return []any{cfg.onDataLoss}
+	case namefn(ProducerLinger):
+		return []any{cfg.linger}
+	case namefn(ManualFlushing):
+		return []any{cfg.manualFlushing}
+	case namefn(RecordDeliveryTimeout):
+		return []any{cfg.recordTimeout}
+	case namefn(TransactionalID):
+		if cfg.txnID != nil {
+			return []any{cfg.txnID, true}
+		}
+		return []any{"", false}
+	case namefn(TransactionTimeout):
+		return []any{cfg.txnTimeout}
+
+	case namefn(ConsumePartitions):
+		return []any{cfg.partitions}
+	case namefn(ConsumePreferringLagFn):
+		return []any{cfg.preferLagFn}
+	case namefn(ConsumeRegex):
+		return []any{cfg.regex}
+	case namefn(ConsumeResetOffset):
+		return []any{cfg.resetOffset}
+	case namefn(ConsumeTopics):
+		return []any{cfg.topics}
+	case namefn(DisableFetchSessions):
+		return []any{cfg.disableFetchSessions}
+	case namefn(FetchIsolationLevel):
+		return []any{cfg.isolationLevel}
+	case namefn(FetchMaxBytes):
+		return []any{int32(cfg.maxBytes)}
+	case namefn(FetchMaxPartitionBytes):
+		return []any{int32(cfg.maxPartBytes)}
+	case namefn(FetchMaxWait):
+		return []any{time.Duration(cfg.maxWait) * time.Millisecond}
+	case namefn(FetchMinBytes):
+		return []any{cfg.minBytes}
+	case namefn(KeepControlRecords):
+		return []any{cfg.keepControl}
+	case namefn(MaxConcurrentFetches):
+		return []any{cfg.maxConcurrentFetches}
+	case namefn(Rack):
+		return []any{cfg.rack}
+
+	case namefn(AdjustFetchOffsetsFn):
+		return []any{cfg.adjustOffsetsBeforeAssign}
+	case namefn(AutoCommitCallback):
+		return []any{cfg.commitCallback}
+	case namefn(AutoCommitInterval):
+		return []any{cfg.autocommitInterval}
+	case namefn(AutoCommitMarks):
+		return []any{cfg.autocommitMarks}
+	case namefn(Balancers):
+		return []any{cfg.balancers}
+	case namefn(BlockRebalanceOnPoll):
+		return []any{cfg.blockRebalanceOnPoll}
+	case namefn(ConsumerGroup):
+		return []any{cfg.group}
+	case namefn(DisableAutoCommit):
+		return []any{cfg.autocommitDisable}
+	case namefn(GreedyAutoCommit):
+		return []any{cfg.autocommitGreedy}
+	case namefn(GroupProtocol):
+		return []any{cfg.protocol}
+	case namefn(HeartbeatInterval):
+		return []any{cfg.heartbeatInterval}
+	case namefn(InstanceID):
+		if cfg.instanceID != nil {
+			return []any{*cfg.instanceID, true}
+		}
+		return []any{"", false}
+	case namefn(OnOffsetsFetched):
+		return []any{cfg.onFetched}
+	case namefn(OnPartitionsAssigned):
+		return []any{cfg.onAssigned}
+	case namefn(OnPartitionsLost):
+		return []any{cfg.onLost}
+	case namefn(OnPartitionsRevoked):
+		return []any{cfg.onRevoked}
+	case namefn(RebalanceTimeout):
+		return []any{cfg.rebalanceTimeout}
+	case namefn(RequireStableFetchOffsets):
+		return []any{cfg.requireStable}
+	case namefn(SessionTimeout):
+		return []any{cfg.sessionTimeout}
+	default:
+		return nil
+	}
+}
+
 // NewClient returns a new Kafka client with the given options or an error if
 // the options are invalid. Connections to brokers are lazily created only when
 // requests are written to them.

pkg/kgo/config.go

@@ -60,8 +60,9 @@ func (consumerOpt) consumerOpt()       {}
 func (groupOpt) groupOpt()             {}
 
 // A cfg can be written to while initializing a client, and after that it is
-// only ever read from. Some areas of initializing may follow options, but all
-// initializing is done before NewClient returns.
+// (mostly) only ever read from. Some areas can continue to be modified --
+// particularly reconfiguring what to consume from -- but most areas are
+// static.
 type cfg struct {
 	/////////////////////
 	// GENERAL SECTION //