Repository: numaproj/numaflow (go)

Issue:
Both redis and jetstream in isbsvc causes unexpected behavior
**Describe the bug**

Things can get into a weird state if you accidentally leave multiple isbsvc types in spec.

**To Reproduce**
Steps to reproduce the behavior:

1. The following causes only redis to run:
```yaml
apiVersion: numaflow.numaproj.io/v1alpha1
kind: InterStepBufferService
metadata:
  name: default
spec:
  jetstream:
    version: latest
  redis:
    native:
      version: 6.2.6
```
2. If you then apply the following, both redis and jetstream run:
```yaml
apiVersion: numaflow.numaproj.io/v1alpha1
kind: InterStepBufferService
metadata:
  name: default
spec:
  jetstream:
    version: latest
```

**Expected behavior**
Applied yaml should either error during isbsvc validation or be reflected in cluster.

**Environment (please complete the following information):**

- Kubernetes: 1.25.3
- Numaflow: head of main

**Additional context**
Add any other context about the problem here.

---

<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->

**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.


Pull request description:
fix: validate only one isbsvc implementation type is defined. Fixes #269
Signed-off-by: David Seapy <dseapy@gmail.com>

Fixes #269

Validates only one isbsvc implementation type is defined


Patch:
diff --git a/pkg/reconciler/isbsvc/validate.go b/pkg/reconciler/isbsvc/validate.go
index ef891503..d420e922 100644
--- a/pkg/reconciler/isbsvc/validate.go
+++ b/pkg/reconciler/isbsvc/validate.go
@@ -8,25 +8,28 @@ import (
 
 // ValidateInterStepBufferService accepts an isbs and performs validation against it
 func ValidateInterStepBufferService(isbs *dfv1.InterStepBufferService) error {
+	if isbs.Spec.Redis != nil && isbs.Spec.JetStream != nil {
+		return fmt.Errorf(`invalid spec: "spec.redis" and "spec.jetstream" can not be defined together`)
+	}
 	if isbs.Spec.Redis == nil && isbs.Spec.JetStream == nil {
-		return fmt.Errorf("invalid spec: either \"spec.redis\" or \"spec.jetstream\" needs to be specified")
+		return fmt.Errorf(`invalid spec: either "spec.redis" or "spec.jetstream" needs to be specified`)
 	}
 	if isbs.Spec.Redis != nil {
 		if isbs.Spec.Redis.Native != nil && isbs.Spec.Redis.External != nil {
-			return fmt.Errorf("\"native\" and \"external\" can not be defined together")
+			return fmt.Errorf(`"native" and "external" can not be defined together`)
 		}
 		if isbs.Spec.Redis.Native == nil && isbs.Spec.Redis.External == nil {
-			return fmt.Errorf("either \"native\" or \"external\" must be defined")
+			return fmt.Errorf(`either "native" or "external" must be defined`)
 		}
 		if native := isbs.Spec.Redis.Native; native != nil {
 			if native.Version == "" {
-				return fmt.Errorf("invalid spec: \"spec.redis.native.version\" is not defined")
+				return fmt.Errorf(`invalid spec: "spec.redis.native.version" is not defined`)
 			}
 		}
 	}
 	if x := isbs.Spec.JetStream; x != nil {
 		if x.Version == "" {
-			return fmt.Errorf("invalid spec: \"spec.jetstream.version\" is not defined")
+			return fmt.Errorf(`invalid spec: "spec.jetstream.version" is not defined`)
 		}
 	}
 	return nil


Test patch:
diff --git a/pkg/reconciler/isbsvc/validate_test.go b/pkg/reconciler/isbsvc/validate_test.go
index f54e4adf..97ffda50 100644
--- a/pkg/reconciler/isbsvc/validate_test.go
+++ b/pkg/reconciler/isbsvc/validate_test.go
@@ -42,12 +42,20 @@ func TestValidateInterStepBuffer(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
+	t.Run("test both redis and jetstream configured", func(t *testing.T) {
+		isbs := testJetStreamIsbs.DeepCopy()
+		isbs.Spec.Redis = testRedisIsbs.DeepCopy().Spec.Redis
+		err := ValidateInterStepBufferService(isbs)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), `spec.redis" and "spec.jetstream" can not be defined together`)
+	})
+
 	t.Run("test missing spec.redis", func(t *testing.T) {
 		isbs := testRedisIsbs.DeepCopy()
 		isbs.Spec.Redis = nil
 		err := ValidateInterStepBufferService(isbs)
 		assert.Error(t, err)
-		assert.Contains(t, err.Error(), "either \"spec.redis\" or \"spec.jetstream\" needs to be specified")
+		assert.Contains(t, err.Error(), `either "spec.redis" or "spec.jetstream" needs to be specified`)
 	})
 
 	t.Run("test missing version", func(t *testing.T) {
@@ -55,7 +63,7 @@ func TestValidateInterStepBuffer(t *testing.T) {
 		isbs.Spec.Redis.Native.Version = ""
 		err := ValidateInterStepBufferService(isbs)
 		assert.Error(t, err)
-		assert.Contains(t, err.Error(), "\"spec.redis.native.version\" is not defined")
+		assert.Contains(t, err.Error(), `"spec.redis.native.version" is not defined`)
 	})
 
 	t.Run("test both native and external configured", func(t *testing.T) {