Fix null ref in stage-vendor and Redis configuration

api/pkg/apis/v1alpha1/vendors/stage-vendor.go

@@ -131,6 +131,9 @@ func (s *StageVendor) Init(config vendors.VendorConfig, factories []managers.IMa
 			if err != nil {
 				sLog.Errorf("V (Stage): failed to report error status: %v (%v)", status.ErrorMessage, err)
 			}
+			// If report status succeeded, return an empty err so the subscribe function will not be retried
+			// The actual error will be stored in Activation cr
+			return err
 		}
 		status.Outputs["__namespace"] = triggerData.Namespace
 
@@ -146,6 +149,9 @@ func (s *StageVendor) Init(config vendors.VendorConfig, factories []managers.IMa
 			if err != nil {
 				sLog.Errorf("V (Stage): failed to report error status: %v (%v)", status.ErrorMessage, err)
 			}
+			// If report status succeeded, return an empty err so the subscribe function will not be retried
+			// The actual error will be stored in Activation cr
+			return err
 		}
 		status.Stage = triggerData.Stage
 		status.ActivationGeneration = triggerData.ActivationGeneration

coa/pkg/apis/v1alpha2/providers/pubsub/redis/redis.go

@@ -106,20 +106,21 @@ func RedisPubSubProviderConfigFromMap(properties map[string]string) (RedisPubSub
 	if v, ok := properties["processingTimeout"]; ok {
 		val := v //providers.LoadEnv(v)
 		if val != "" {
-			n, err := utils.UnmarshalDuration(val)
+			n, err := parseDuration(val)
 			if err != nil {
-				return ret, v1alpha2.NewCOAError(err, "invalid int value in the 'processingTimeout' setting of Redis pub-sub provider", v1alpha2.BadConfig)
+				return ret, v1alpha2.NewCOAError(err, "invalid value in the 'processingTimeout' setting of Redis pub-sub provider", v1alpha2.BadConfig)
+			} else {
+				ret.ProcessingTimeout = n
 			}
-			ret.ProcessingTimeout = n
 		}
 	}
 
 	if v, ok := properties["redeliverInterval"]; ok {
 		val := v //providers.LoadEnv(v)
 		if val != "" {
-			n, err := utils.UnmarshalDuration(val)
+			n, err := parseDuration(val)
 			if err != nil {
-				return ret, v1alpha2.NewCOAError(err, "invalid int value in the 'redeliverInterval' setting of Redis pub-sub provider", v1alpha2.BadConfig)
+				return ret, v1alpha2.NewCOAError(err, "invalid value in the 'redeliverInterval' setting of Redis pub-sub provider", v1alpha2.BadConfig)
 			}
 			ret.RedeliverInterval = n
 		}
@@ -379,12 +380,35 @@ func toRedisPubSubProviderConfig(config providers.IProviderConfig) (RedisPubSubP
 	if err != nil {
 		return ret, err
 	}
-	err = json.Unmarshal(data, &ret)
-	//ret.Name = providers.LoadEnv(ret.Name)
-	//ret.Host = providers.LoadEnv(ret.Host)
-	//ret.Password = providers.LoadEnv(ret.Password)
-	if ret.NumberOfWorkers <= 0 {
-		ret.NumberOfWorkers = 1
+	var configs map[string]interface{}
+	err = json.Unmarshal(data, &configs)
+	if err != nil {
+		mLog.Info("  P (Redis PubSub): failed to parse to map[string]interface{} %+v", err)
+		return ret, err
+	}
+	configStrings := map[string]string{}
+	for k, v := range configs {
+		configStrings[k] = utils.FormatAsString(v)
+	}
+
+	ret, err = RedisPubSubProviderConfigFromMap(configStrings)
+	if err != nil {
+		mLog.Info("  P (Redis PubSub): failed to parse to RedisPubSubProviderConfig %+v", err)
+		return ret, err
 	}
 	return ret, err
 }
+
+func parseDuration(val string) (time.Duration, error) {
+	n, err := utils.UnmarshalDuration(val)
+	if err != nil {
+		n, err = time.ParseDuration(val)
+		if err != nil {
+			return time.Duration(1), err
+		} else {
+			return n, nil
+		}
+	} else {
+		return n, nil
+	}
+}

packages/helm/symphony/files/symphony-api.json

@@ -37,7 +37,10 @@
           "host": "{{ include "symphony.redisHost" . }}",
           "requireTLS": false,
           "password": "",
-          "numberOfWorkers": 1
+          "numberOfWorkers": 1,
+          "queueDepth": 10,
+          "processingTimeout": "15s",
+          "redeliverInterval": "1s"
         }
       }
       {{- else }}