kgo: fix / improve handling deleted topics while regex consuming

twmb · twmb · commit 627d39a0d8ad · 2023-07-27T11:47:52.000-06:00
* The topic name was not ever actually saved on an internal struct, so it was impossible to stop consuming a regex consumed topic * If all topics were deleted, the conditional (len(latest)) never fired and could not purge the deleted topics internally * We now avoid spam reloading metadata if consuming runs into UnknownTopicOrPartition -- we still reload, just not with an immediate trigger Closes #523.
diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go
@@ -281,6 +281,8 @@ func (cl *Client) OptValues(opt any) []any {
 		return []any{cfg.hooks}
 	case namefn(ConcurrentTransactionsBackoff):
 		return []any{cfg.txnBackoff}
+	case namefn(considerMissingTopicDeletedAfter):
+		return []any{cfg.missingTopicDelete}
 
 	case namefn(DefaultProduceTopic):
 		return []any{cfg.defaultProduceTopic}
diff --git a/pkg/kgo/config.go b/pkg/kgo/config.go
@@ -121,6 +121,7 @@ type cfg struct {
 	recordTimeout       time.Duration
 	manualFlushing      bool
 	txnBackoff          time.Duration
+	missingTopicDelete  time.Duration
 
 	partitioner Partitioner
 
@@ -480,8 +481,9 @@ func defaultCfg() cfg {
 		maxBrokerWriteBytes: 100 << 20, // Kafka socket.request.max.bytes default is 100<<20
 		maxBrokerReadBytes:  100 << 20,
 
-		metadataMaxAge: 5 * time.Minute,
-		metadataMinAge: 5 * time.Second / 2,
+		metadataMaxAge:     5 * time.Minute,
+		metadataMinAge:     5 * time.Second,
+		missingTopicDelete: 15 * time.Second,
 
 		//////////////
 		// producer //
@@ -787,7 +789,7 @@ func MetadataMaxAge(age time.Duration) Opt {
 }
 
 // MetadataMinAge sets the minimum time between metadata queries, overriding
-// the default 2.5s. You may want to raise or lower this to reduce the number of
+// the default 5s. You may want to raise or lower this to reduce the number of
 // metadata queries the client will make. Notably, if metadata detects an error
 // in any topic or partition, it triggers itself to update as soon as allowed.
 func MetadataMinAge(age time.Duration) Opt {
@@ -831,6 +833,13 @@ func ConcurrentTransactionsBackoff(backoff time.Duration) Opt {
 	return clientOpt{func(cfg *cfg) { cfg.txnBackoff = backoff }}
 }
 
+// considerMissingTopicDeletedAfter sets the amount of time a topic can be
+// missing from metadata responses _after_ loading it at least once before it
+// is considered deleted.
+func considerMissingTopicDeletedAfter(t time.Duration) Opt {
+	return clientOpt{func(cfg *cfg) { cfg.missingTopicDelete = t }}
+}
+
 ////////////////////////////
 // PRODUCER CONFIGURATION //
 ////////////////////////////
diff --git a/pkg/kgo/consumer_direct_test.go b/pkg/kgo/consumer_direct_test.go
@@ -2,6 +2,7 @@ package kgo
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sort"
 	"sync/atomic"
@@ -319,3 +320,48 @@ func TestPauseIssue489(t *testing.T) {
 		cl.ResumeFetchPartitions(map[string][]int32{t1: {0}})
 	}
 }
+
+func TestIssue523(t *testing.T) {
+	t.Parallel()
+
+	t1, cleanup := tmpTopicPartitions(t, 1)
+	defer cleanup()
+	g1, gcleanup := tmpGroup(t)
+	defer gcleanup()
+
+	cl, _ := NewClient(
+		getSeedBrokers(),
+		DefaultProduceTopic(t1),
+		ConsumeTopics(".*"+t1+".*"),
+		ConsumeRegex(),
+		ConsumerGroup(g1),
+		MetadataMinAge(100*time.Millisecond),
+		FetchMaxWait(time.Second),
+		KeepRetryableFetchErrors(),
+	)
+	defer cl.Close()
+
+	if err := cl.ProduceSync(context.Background(), StringRecord("foo")).FirstErr(); err != nil {
+		t.Fatal(err)
+	}
+
+	cl.PollFetches(context.Background())
+
+	cleanup() // delete the topic
+
+	start := time.Now()
+	for {
+		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		fs := cl.PollFetches(ctx)
+		cancel()
+		if errors.Is(fs.Err0(), context.DeadlineExceeded) {
+			break
+		}
+		if time.Since(start) > 40*time.Second { // missing topic delete is 15s by default
+			t.Fatalf("still repeatedly requesting metadata after 20s")
+		}
+		if fs.Err0() != nil {
+			time.Sleep(time.Second)
+		}
+	}
+}
diff --git a/pkg/kgo/metadata.go b/pkg/kgo/metadata.go
@@ -341,13 +341,11 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 	}
 	groupExternal.updateLatest(latest)
 
-	const maxMissTime = 15 * time.Second
-
 	// If we are consuming with regex and fetched all topics, the metadata
 	// may have returned topics the consumer is not yet tracking. We ensure
 	// that we will store the topics at the end of our metadata update.
 	tpsConsumerLoad := tpsConsumer.load()
-	if all && len(latest) > 0 {
+	if all {
 		allTopics := make([]string, 0, len(latest))
 		for topic := range latest {
 			allTopics = append(allTopics, topic)
@@ -356,16 +354,16 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 		defer tpsConsumer.storeData(tpsConsumerLoad)
 
 		// For regex consuming, if a topic is not returned in the
-		// response and for at least maxMissTime from when we first
-		// discovered it, we assume the topic has been deleted and
-		// purge it. We allow for maxMissTime because (in testing
-		// locally) Kafka can originally broadcast a newly created
-		// topic exists and then fail to broadcast that info again for
-		// a while.
+		// response and for at least missingTopicDelete from when we
+		// first discovered it, we assume the topic has been deleted
+		// and purge it. We allow for missingTopicDelete because (in
+		// testing locally) Kafka can originally broadcast a newly
+		// created topic exists and then fail to broadcast that info
+		// again for a while.
 		var purgeTopics []string
 		for topic, tps := range tpsConsumerLoad {
 			if _, ok := latest[topic]; !ok {
-				if td := tps.load(); td.when != 0 && time.Since(time.Unix(td.when, 0)) > maxMissTime {
+				if td := tps.load(); td.when != 0 && time.Since(time.Unix(td.when, 0)) > cl.cfg.missingTopicDelete {
 					purgeTopics = append(purgeTopics, td.topic)
 				} else {
 					retryWhy.add(topic, -1, errMissingTopic)
@@ -445,7 +443,7 @@ func (cl *Client) updateMetadata() (retryWhy multiUpdateWhy, err error) {
 		var bumpFail []string
 		for _, tps := range missingProduceTopics {
 			if all {
-				if td := tps.load(); td.when != 0 && time.Since(time.Unix(td.when, 0)) > maxMissTime {
+				if td := tps.load(); td.when != 0 && time.Since(time.Unix(td.when, 0)) > cl.cfg.missingTopicDelete {
 					bumpFail = append(bumpFail, td.topic)
 				} else {
 					retryWhy.add(td.topic, -1, errMissingTopic)
@@ -683,6 +681,7 @@ func (cl *Client) mergeTopicPartitions(
 
 	lv.loadErr = r.loadErr
 	lv.isInternal = r.isInternal
+	lv.topic = r.topic
 	if lv.when == 0 {
 		lv.when = r.when
 	}
@@ -876,6 +875,18 @@ type kerrOrString struct {
 	s string
 }
 
+func (m *multiUpdateWhy) isOnly(err error) bool {
+	if m == nil {
+		return false
+	}
+	for e := range *m {
+		if !errors.Is(err, e.k) {
+			return false
+		}
+	}
+	return true
+}
+
 func (m *multiUpdateWhy) add(t string, p int32, err error) {
 	if err == nil {
 		return
diff --git a/pkg/kgo/source.go b/pkg/kgo/source.go
@@ -680,8 +680,7 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 		reloadOffsets   listOrEpochLoads
 		preferreds      cursorPreferreds
 		allErrsStripped bool
-		updateMeta      bool
-		updateWhy       string
+		updateWhy       multiUpdateWhy
 		handled         = make(chan struct{})
 	)
 
@@ -692,7 +691,7 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 	// Processing the response only needs the source's nodeID and client.
 	go func() {
 		defer close(handled)
-		fetch, reloadOffsets, preferreds, allErrsStripped, updateMeta, updateWhy = s.handleReqResp(br, req, resp)
+		fetch, reloadOffsets, preferreds, allErrsStripped, updateWhy = s.handleReqResp(br, req, resp)
 	}()
 
 	select {
@@ -772,8 +771,21 @@ func (s *source) fetch(consumerSession *consumerSession, doneFetch chan<- struct
 		s.session.bumpEpoch(resp.SessionID)
 	}
 
-	if updateMeta && !reloadOffsets.loadWithSessionNow(consumerSession, updateWhy) {
-		s.cl.triggerUpdateMetadataNow(updateWhy)
+	// If we have a reason to update (per-partition fetch errors), and the
+	// reason is not just unknown topic or partition, then we immediately
+	// update metadata. We avoid updating for unknown because it _likely_
+	// means the topic does not exist and reloading is wasteful. We only
+	// trigger a metadata update if we have no reload offsets. Having
+	// reload offsets *always* triggers a metadata update.
+	if updateWhy != nil {
+		why := updateWhy.reason("fetch had inner topic errors")
+		if !reloadOffsets.loadWithSessionNow(consumerSession, why) {
+			if updateWhy.isOnly(kerr.UnknownTopicOrPartition) || updateWhy.isOnly(kerr.UnknownTopicID) {
+				s.cl.triggerUpdateMetadata(false, why)
+			} else {
+				s.cl.triggerUpdateMetadataNow(why)
+			}
+		}
 	}
 
 	if fetch.hasErrorsOrRecords() {
@@ -808,12 +820,10 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 	reloadOffsets listOrEpochLoads,
 	preferreds cursorPreferreds,
 	allErrsStripped bool,
-	updateMeta bool,
-	why string,
+	updateWhy multiUpdateWhy,
 ) {
 	f = Fetch{Topics: make([]FetchTopic, 0, len(resp.Topics))}
 	var (
-		updateWhy        multiUpdateWhy
 		debugWhyStripped multiUpdateWhy
 		numErrsStripped  int
 		kip320           = s.cl.supportsOffsetForLeaderEpoch()
@@ -878,7 +888,6 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 
 			fp := partOffset.processRespPartition(br, rp, s.cl.decompressor, s.cl.cfg.hooks)
 			if fp.Err != nil {
-				updateMeta = true
 				updateWhy.add(topic, partition, fp.Err)
 			}
 
@@ -1024,7 +1033,7 @@ func (s *source) handleReqResp(br *broker, req *fetchRequest, resp *kmsg.FetchRe
 		s.cl.cfg.logger.Log(LogLevelDebug, "fetch stripped partitions", "why", debugWhyStripped.reason(""))
 	}
 
-	return f, reloadOffsets, preferreds, req.numOffsets == numErrsStripped, updateMeta, updateWhy.reason("fetch had inner topic errors")
+	return f, reloadOffsets, preferreds, req.numOffsets == numErrsStripped, updateWhy
 }
 
 // processRespPartition processes all records in all potentially compressed
