broker: hide retryable errors *once*

twmb · Nov 29, 2022 · bc6810d · bc6810d
1 parent 20dd08d
commit bc6810d
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/pkg/kgo/broker.go b/pkg/kgo/broker.go
@@ -287,6 +287,13 @@ start:
 	{
 		var err error
 		if cxn, err = b.loadConnection(pr.ctx, req); err != nil {
+			// It is rare, but it is possible that the broker has
+			// an immediate issue on a new connection. We retry
+			// once.
+			if isRetriableBrokerErr(err) && !retriedOnNewConnection {
+				retriedOnNewConnection = true
+				goto start
+			}
 			pr.promise(nil, err)
 			return
 		}
@@ -664,7 +671,7 @@ func (cxn *brokerCxn) init(isProduceCxn bool) error {
 	if !hasVersions {
 		if cxn.b.cl.cfg.maxVersions == nil || cxn.b.cl.cfg.maxVersions.HasKey(18) {
 			if err := cxn.requestAPIVersions(); err != nil {
-				if !errors.Is(err, ErrClientClosed) {
+				if !errors.Is(err, ErrClientClosed) && !isRetriableBrokerErr(err) {
 					cxn.cl.cfg.logger.Log(LogLevelError, "unable to request api versions", "broker", logID(cxn.b.meta.NodeID), "err", err)
 				}
 				return err
@@ -677,7 +684,7 @@ func (cxn *brokerCxn) init(isProduceCxn bool) error {
 	}
 
 	if err := cxn.sasl(); err != nil {
-		if !errors.Is(err, ErrClientClosed) {
+		if !errors.Is(err, ErrClientClosed) && !isRetriableBrokerErr(err) {
 			cxn.cl.cfg.logger.Log(LogLevelError, "unable to initialize sasl", "broker", logID(cxn.b.meta.NodeID), "err", err)
 		}
 		return err