pkg/kgo: patch AddConsumeTopics

* This previously did not work for the direct consumer
* We now allow this work work for clients that do not consume anything
to begin with
* We now allow a consumer to join a group with no topic interests, so
that the end-user can add interests later

pkg/kgo/config.go

@@ -346,9 +346,6 @@ func (cfg *cfg) validate() error {
  }
 
  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")
  }

pkg/kgo/consumer.go

@@ -290,12 +290,10 @@ func (c *consumer) init(cl *Client) {
  c.sourcesReadyCond = sync.NewCond(&c.sourcesReadyMu)
  c.pollWaitC = sync.NewCond(&c.pollWaitMu)
 
- if len(cl.cfg.topics) == 0 && len(cl.cfg.partitions) == 0 {
- return // not consuming
+ if len(cl.cfg.topics) > 0 || len(cl.cfg.partitions) > 0 {
+ defer cl.triggerUpdateMetadataNow("querying metadata for consumer initialization") // we definitely want to trigger a metadata update
  }
 
- defer cl.triggerUpdateMetadataNow("client initialization") // we definitely want to trigger a metadata update
-
  if len(cl.cfg.group) == 0 {
  c.initDirect()
  } else {

pkg/kgo/consumer_direct.go

@@ -59,9 +59,7 @@ func (d *directConsumer) findNewAssignments() map[string]map[int32]Offset {
 
  toUse := make(map[string]map[int32]Offset, 10)
  for topic, topicPartitions := range topics {
- // If we are using regex topics, we have to check all
- // topic regexes to see if any match on this topic.
- var useTopic bool
+ useTopic := true
  if d.cfg.regex {
  want, seen := d.reSeen[topic]
  if !seen {
@@ -77,8 +75,6 @@ func (d *directConsumer) findNewAssignments() map[string]map[int32]Offset {
  d.reSeen[topic] = want
  }
  useTopic = want
- } else {
- _, useTopic = d.cfg.topics[topic]
  }
 
  // If the above detected that we want to keep this topic, we

pkg/kgo/sink.go

@@ -366,8 +366,8 @@ func (s *sink) produce(sem <-chan struct{}) bool {
 
  batches := req.batches.sliced()
  s.doSequenced(req, func(br *broker, resp kmsg.Response, err error) {
- s.cl.producer.decInflight()
  s.handleReqResp(br, req, resp, err)
+ s.cl.producer.decInflight()
  batches.eachOwnerLocked((*recBatch).decInflight)
  <-sem
  })