UnknownTopicRetries: allow -1 to disable the option

twmb · Apr 30, 2022 · 96d647a · 96d647a
1 parent 0fa9fb7
commit 96d647a
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 2 deletions.
diff --git a/pkg/kgo/config.go b/pkg/kgo/config.go
@@ -964,6 +964,8 @@ func RecordRetries(n int) ProducerOpt {
 // exist, and if we repeatedly see that the topic does not exist across
 // multiple metadata queries (which are going to different brokers), then we
 // may as well stop trying and fail the records.
+//
+// If this is -1, the client never fails records with this error.
 func UnknownTopicRetries(n int) ProducerOpt {
 	return producerOpt{func(cfg *cfg) { cfg.maxUnknownFailures = int64(n) }}
 }

diff --git a/pkg/kgo/producer.go b/pkg/kgo/producer.go
@@ -825,7 +825,7 @@ func (cl *Client) waitUnknownTopic(
 			if int64(tries) >= cl.cfg.recordRetries {
 				err = fmt.Errorf("no partitions available after attempting to refresh metadata %d times, last err: %w", tries, retriableErr)
 			}
-			if errors.Is(retriableErr, kerr.UnknownTopicOrPartition) {
+			if cl.cfg.maxUnknownFailures >= 0 && errors.Is(retriableErr, kerr.UnknownTopicOrPartition) {
 				unknownTries++
 				if unknownTries > cl.cfg.maxUnknownFailures {
 					err = retriableErr

diff --git a/pkg/kgo/sink.go b/pkg/kgo/sink.go
@@ -1214,7 +1214,7 @@ func (recBuf *recBuf) checkUnknownFailLimit(err error) bool {
 	} else {
 		recBuf.unknownFailures = 0
 	}
-	return recBuf.unknownFailures > recBuf.cl.cfg.maxUnknownFailures
+	return recBuf.cl.cfg.maxUnknownFailures >= 0 && recBuf.unknownFailures > recBuf.cl.cfg.maxUnknownFailures
 }
 
 // failAllRecords fails all buffered records in this recBuf.