kgo: allow retries on dial timeouts

I saw a dial failing recently because the OS was very overloaded and
then slow -- dial timeouts are fine. Every other dial failure indicates
a bigger problem.

Author: twmb

pkg/kgo/client.go

@@ -1122,7 +1122,7 @@ type failDial struct{ fails int8 }
 // repeatedly fails, we need to forget our cache to force a re-load: the broker
 // may have completely died.
 func (d *failDial) isRepeatedDialFail(err error) bool {
-	if isDialErr(err) {
+	if isAnyDialErr(err) {
 		d.fails++
 		if d.fails == 3 {
 			d.fails = 0

pkg/kgo/consumer.go

@@ -1780,7 +1780,7 @@ func (s *consumerSession) handleListOrEpochResults(loaded loadedOffsets) (reload
 
 		default: // from ErrorCode in a response, or broker request err, or request is canceled as our session is ending
 			reloads.addLoad(load.topic, load.partition, loaded.loadType, load.request)
-			if !kerr.IsRetriable(load.err) && !isRetryableBrokerErr(load.err) && !isDialErr(load.err) && !isContextErr(load.err) { // non-retryable response error; signal such in a response
+			if !kerr.IsRetriable(load.err) && !isRetryableBrokerErr(load.err) && !isDialNonTimeoutErr(load.err) && !isContextErr(load.err) { // non-retryable response error; signal such in a response
 				s.c.addFakeReadyForDraining(load.topic, load.partition, load.err, fmt.Sprintf("notification of non-retryable error from %s request", loaded.loadType))
 			}
 

pkg/kgo/errors.go

@@ -44,7 +44,7 @@ func isRetryableBrokerErr(err error) bool {
 		// If a dial fails, potentially we could retry if the resolver
 		// had a temporary hiccup, but we will err on the side of this
 		// being a slightly less temporary error.
-		return !isDialErr(err)
+		return !isDialNonTimeoutErr(err)
 	}
 	// EOF can be returned if a broker kills a connection unexpectedly, and
 	// we can retry that. Same for ErrClosed.
@@ -88,7 +88,12 @@ func isRetryableBrokerErr(err error) bool {
 	return false
 }
 
-func isDialErr(err error) bool {
+func isDialNonTimeoutErr(err error) bool {
+	var ne *net.OpError
+	return errors.As(err, &ne) && ne.Op == "dial" && !ne.Timeout()
+}
+
+func isAnyDialErr(err error) bool {
 	var ne *net.OpError
 	return errors.As(err, &ne) && ne.Op == "dial"
 }

pkg/kgo/sink.go

@@ -326,7 +326,7 @@ func (s *sink) produce(sem <-chan struct{}) bool {
 		batchesStripped, err := s.doTxnReq(req, txnReq)
 		if err != nil {
 			switch {
-			case isRetryableBrokerErr(err) || isDialErr(err):
+			case isRetryableBrokerErr(err) || isDialNonTimeoutErr(err):
 				s.cl.bumpRepeatedLoadErr(err)
 				s.cl.cfg.logger.Log(LogLevelWarn, "unable to AddPartitionsToTxn due to retryable broker err, bumping client's buffered record load errors by 1 and retrying", "err", err)
 				s.cl.triggerUpdateMetadata(false, "attempting to refresh broker list due to failed AddPartitionsToTxn requests")
@@ -532,7 +532,7 @@ func (s *sink) handleReqClientErr(req *produceRequest, err error) {
 		fallthrough
 
 	case errors.Is(err, errUnknownBroker),
-		isDialErr(err),
+		isDialNonTimeoutErr(err),
 		isRetryableBrokerErr(err):
 		updateMeta := !isRetryableBrokerErr(err)
 		if updateMeta {
@@ -1269,7 +1269,7 @@ func (recBuf *recBuf) bumpRepeatedLoadErr(err error) {
 	var (
 		canFail        = !recBuf.cl.idempotent() || batch0.canFailFromLoadErrs // we can only fail if we are not idempotent or if we have no outstanding requests
 		batch0Fail     = batch0.maybeFailErr(&recBuf.cl.cfg) != nil            // timeout, retries, or aborting
-		netErr         = isRetryableBrokerErr(err) || isDialErr(err)           // we can fail if this is *not* a network error
+		netErr         = isRetryableBrokerErr(err) || isDialNonTimeoutErr(err) // we can fail if this is *not* a network error
 		retryableKerr  = kerr.IsRetriable(err)                                 // we fail if this is not a retryable kerr,
 		isUnknownLimit = recBuf.checkUnknownFailLimit(err)                     // or if it is, but it is UnknownTopicOrPartition and we are at our limit