[NUC-233] streamconsumergroup - add retries on network errors

pkg/common/helper.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"reflect"
 	"runtime"
+	"strings"
 	"time"
 
 	"github.com/nuclio/errors"
@@ -170,3 +171,20 @@ func StringSlicesEqual(slice1 []string, slice2 []string) bool {
 
 	return true
 }
+
+func EngineErrorIsNonFatal(err error) bool {
+	var nonFatalEngineErrorsPartialMatch = []string{
+		"dialing to the given TCP address timed out",
+		"timeout",
+		"refused",
+	}
+	if err != nil && len(err.Error()) > 0 {
+		for _, nonFatalError := range nonFatalEngineErrorsPartialMatch {
+			if strings.Contains(err.Error(), nonFatalError) || strings.Contains(errors.Cause(err).Error(), nonFatalError) {
+				return true
+			}
+		}
+		return false
+	}
+	return true
+}

pkg/dataplane/streamconsumergroup/claim.go

@@ -129,7 +129,9 @@ func (c *claim) fetchRecordBatches(stopChannel chan struct{}, fetchInterval time
 		&c.getShardLocationBackoff, func(attempt int) (bool, error) {
 			c.currentShardLocation, err = c.getCurrentShardLocation(c.shardID)
 			if err != nil {
-
+				if common.EngineErrorIsNonFatal(err) {
+					return true, errors.Wrap(err, "Failed to get shard location due to a network error")
+				}
 				// requested for an immediate stop
 				if err == v3ioerrors.ErrStopped {
 					return false, nil

pkg/dataplane/streamconsumergroup/streamconsumergroup.go

@@ -118,6 +118,9 @@ func (scg *streamConsumerGroup) setState(modifier stateModifier,
 		if err != nil && !errors.Is(err, v3ioerrors.ErrNotFound) {
 			return true, errors.Wrap(err, "Failed getting current state from persistency")
 		}
+		if common.EngineErrorIsNonFatal(err) {
+			return true, errors.Wrap(err, "Failed getting current state from persistency due to a network error")
+		}
 
 		if state == nil {
 			state, err = newState()