Repository: numaproj/numaflow (go)

Issue:
Unable to use Redis ISBSVC
**Describe the bug**
With the `simple-example.yaml` pipeline, vertice pods are entering CrashLoopBackoff with the following error when the ISBSVC is a Redis one:
```
➜  ~ kubectl logs -n poseidon-dev simple-pipeline-in-0-rjvcu 
Defaulted container "numa" out of: numa, init (init)
{"level":"info","ts":"2024-02-19T09:12:01.959378402Z","logger":"numaflow.Source-processor","caller":"commands/processor.go:48","msg":"Starting vertex data processor","version":"Version: v1.1.4, BuildDate: 2024-01-20T22:16:46Z, GitCommit: 7ffb521bcc15612d04fe66de33d199e8c8391a7a, GitTag: v1.1.4, GitTreeState: clean, GoVersion: go1.20.12, Compiler: gc, Platform: linux/amd64"}
Error: failed to create a new NATS client pool: environment variable "NUMAFLOW_ISBSVC_JETSTREAM_URL" not found
Usage:
  numaflow processor [flags]

Flags:
  -h, --help                 help for processor
      --isbsvc-type string   ISB Service type, e.g. jetstream
      --type string          Processor type, 'source', 'sink' or 'udf'

panic: failed to create a new NATS client pool: environment variable "NUMAFLOW_ISBSVC_JETSTREAM_URL" not found

goroutine 1 [running]:
github.com/numaproj/numaflow/cmd/commands.Execute()
	/home/runner/work/numaflow/numaflow/cmd/commands/root.go:33 +0x45
main.main()
	/home/runner/work/numaflow/numaflow/cmd/main.go:24 +0x17

```

Controller logs (note that I was also deleting the pipeline from the default namespace because I forgot to specify the namespace on first yaml applying).
Also, the IP `10.43.213.145` it can't reach is the one of the pipeline daemon:
```
{"level":"info","ts":"2024-02-19T09:09:06.274758783Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:258","msg":"Created vertex successfully","namespace":"poseidon-dev","pipeline":"simple-pipeline","vertex":"simple-pipeline-in"}
{"level":"info","ts":"2024-02-19T09:09:06.277439784Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:202","msg":"Succeeded to create a service","namespace":"poseidon-dev","vertex":"simple-pipeline-in","pipeline":"simple-pipeline","service":"simple-pipeline-in-headless"}
{"level":"info","ts":"2024-02-19T09:09:06.278521597Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:258","msg":"Created vertex successfully","namespace":"poseidon-dev","pipeline":"simple-pipeline","vertex":"simple-pipeline-cat"}
{"level":"info","ts":"2024-02-19T09:09:06.281616283Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:308","msg":"Succeeded to create a pod","namespace":"poseidon-dev","vertex":"simple-pipeline-in","pipeline":"simple-pipeline","pod":"simple-pipeline-in-0-rjvcu"}
{"level":"info","ts":"2024-02-19T09:09:06.281675077Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:321","msg":"Replicas changed","namespace":"poseidon-dev","vertex":"simple-pipeline-in","pipeline":"simple-pipeline","currentReplicas":0,"desiredReplicas":1}
{"level":"info","ts":"2024-02-19T09:09:06.282023655Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:258","msg":"Created vertex successfully","namespace":"poseidon-dev","pipeline":"simple-pipeline","vertex":"simple-pipeline-out"}
{"level":"info","ts":"2024-02-19T09:09:06.285002546Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:302","msg":"Created a job successfully for ISB Svc creating","namespace":"poseidon-dev","pipeline":"simple-pipeline","buffers":["poseidon-dev-simple-pipeline-cat-0","poseidon-dev-simple-pipeline-out-0"],"buckets":["poseidon-dev-simple-pipeline-out_SINK","poseidon-dev-simple-pipeline-in-cat","poseidon-dev-simple-pipeline-cat-out","poseidon-dev-simple-pipeline-in_SOURCE"]}
{"level":"info","ts":"2024-02-19T09:09:06.287896311Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:202","msg":"Succeeded to create a service","namespace":"poseidon-dev","vertex":"simple-pipeline-cat","pipeline":"simple-pipeline","service":"simple-pipeline-cat-headless"}
{"level":"info","ts":"2024-02-19T09:09:06.288450618Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:368","msg":"Succeeded to create a daemon service","namespace":"poseidon-dev","pipeline":"simple-pipeline","service":"simple-pipeline-daemon-svc"}
{"level":"info","ts":"2024-02-19T09:09:06.291215162Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:308","msg":"Succeeded to create a pod","namespace":"poseidon-dev","vertex":"simple-pipeline-cat","pipeline":"simple-pipeline","pod":"simple-pipeline-cat-0-duttw"}
{"level":"info","ts":"2024-02-19T09:09:06.291233257Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:420","msg":"Succeeded to create/recreate a daemon deployment","namespace":"poseidon-dev","pipeline":"simple-pipeline","deployment":"simple-pipeline-daemon"}
{"level":"info","ts":"2024-02-19T09:09:06.291252395Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:321","msg":"Replicas changed","namespace":"poseidon-dev","vertex":"simple-pipeline-cat","pipeline":"simple-pipeline","currentReplicas":0,"desiredReplicas":1}
{"level":"info","ts":"2024-02-19T09:09:06.297616015Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:202","msg":"Succeeded to create a service","namespace":"poseidon-dev","vertex":"simple-pipeline-out","pipeline":"simple-pipeline","service":"simple-pipeline-out-headless"}
{"level":"info","ts":"2024-02-19T09:09:06.300300533Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:308","msg":"Succeeded to create a pod","namespace":"poseidon-dev","vertex":"simple-pipeline-out","pipeline":"simple-pipeline","pod":"simple-pipeline-out-0-fnwlt"}
{"level":"info","ts":"2024-02-19T09:09:06.300335571Z","logger":"numaflow.controller-manager","caller":"vertex/controller.go:321","msg":"Replicas changed","namespace":"poseidon-dev","vertex":"simple-pipeline-out","pipeline":"simple-pipeline","currentReplicas":0,"desiredReplicas":1}
{"level":"info","ts":"2024-02-19T09:09:10.745678837Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:102","msg":"Deleting pipeline","namespace":"default","pipeline":"simple-pipeline"}
{"level":"error","ts":"2024-02-19T09:09:10.747394743Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:108","msg":"Failed to check if it's safe to delete pipeline simple-pipeline: rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp: lookup simple-pipeline-daemon-svc.default.svc on 10.43.0.10:53: no such host\"","namespace":"default","pipeline":"simple-pipeline","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/pipeline.(*pipelineReconciler).reconcile\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/pipeline/controller.go:108\ngithub.com/numaproj/numaflow/pkg/reconciler/pipeline.(*pipelineReconciler).Reconcile\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/pipeline/controller.go:81\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).Reconcile\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:122\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).reconcileHandler\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:323\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).processNextWorkItem\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:274\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).Start.func2.2\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:235"}
{"level":"error","ts":"2024-02-19T09:09:10.747432566Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:83","msg":"Reconcile error","namespace":"default","pipeline":"simple-pipeline","error":"rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp: lookup simple-pipeline-daemon-svc.default.svc on 10.43.0.10:53: no such host\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/pipeline.(*pipelineReconciler).Reconcile\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/pipeline/controller.go:83\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).Reconcile\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:122\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).reconcileHandler\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:323\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).processNextWorkItem\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:274\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).Start.func2.2\n\t/home/runner/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.14.6/pkg/internal/controller/controller.go:235"}
{"level":"info","ts":"2024-02-19T09:10:32.673818221Z","logger":"numaflow.controller-manager","caller":"pipeline/controller.go:102","msg":"Deleting pipeline","namespace":"default","pipeline":"simple-pipeline"}
{"level":"error","ts":"2024-02-19T09:10:57.07908779Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-cat","error":"error while fetching buffer info, failed to get the buffer information of vertex \"simple-pipeline-cat\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
{"level":"error","ts":"2024-02-19T09:10:57.079126104Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-out","error":"error while fetching buffer info, failed to get the buffer information of vertex \"simple-pipeline-out\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
{"level":"error","ts":"2024-02-19T09:10:57.07918074Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-in","error":"failed to get metrics of vertex key \"poseidon-dev/simple-pipeline-in\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
{"level":"error","ts":"2024-02-19T09:11:07.079299352Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-cat","error":"error while fetching buffer info, failed to get the buffer information of vertex \"simple-pipeline-cat\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
{"level":"error","ts":"2024-02-19T09:11:17.080321494Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-out","error":"error while fetching buffer info, failed to get the buffer information of vertex \"simple-pipeline-out\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
{"level":"error","ts":"2024-02-19T09:11:27.0809779Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-in","error":"failed to get metrics of vertex key \"poseidon-dev/simple-pipeline-in\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
{"level":"error","ts":"2024-02-19T09:11:37.081828284Z","logger":"numaflow.autoscaler","caller":"scaling/scaling.go:124","msg":"Failed to scale a vertex","vertex":"poseidon-dev/simple-pipeline-cat","error":"error while fetching buffer info, failed to get the buffer information of vertex \"simple-pipeline-cat\", rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing: dial tcp 10.43.213.145:4327: i/o timeout\"","stacktrace":"github.com/numaproj/numaflow/pkg/reconciler/vertex/scaling.(*Scaler).scale\n\t/home/runner/work/numaflow/numaflow/pkg/reconciler/vertex/scaling/scaling.go:124"}
```

Pipeline daemon logs:
```
➜  ~ kubectl logs --all-containers -n poseidon-dev simple-pipeline-daemon-6cb498bfb-zhf6j
{"level":"info","ts":"2024-02-19T09:09:07.800450562Z","logger":"numaflow.isbsvc-validate","caller":"commands/isbsvc_validate.go:80","msg":"Validate buffers, buckets and side inputs store successfully","pipeline":"simple-pipeline"}
{"level":"info","ts":"2024-02-19T09:30:12.22141854Z","logger":"numaflow.daemon-server","caller":"commands/daemon_server.go:48","msg":"Starting daemon server","pipeline":"simple-pipeline","version":"Version: v1.1.4, BuildDate: 2024-01-20T22:16:46Z, GitCommit: 7ffb521bcc15612d04fe66de33d199e8c8391a7a, GitTag: v1.1.4, GitTreeState: clean, GoVersion: go1.20.12, Compiler: gc, Platform: linux/amd64"}
{"level":"error","ts":"2024-02-19T09:30:12.221635402Z","logger":"numaflow.daemon-server","caller":"server/daemon_server.go:74","msg":"Failed to get a NATS client pool.","pipeline":"simple-pipeline","error":"environment variable \"NUMAFLOW_ISBSVC_JETSTREAM_URL\" not found","stacktrace":"github.com/numaproj/numaflow/pkg/daemon/server.(*daemonServer).Run\n\t/home/runner/work/numaflow/numaflow/pkg/daemon/server/daemon_server.go:74\ngithub.com/numaproj/numaflow/cmd/commands.NewDaemonServerCommand.func1\n\t/home/runner/work/numaflow/numaflow/cmd/commands/daemon_server.go:51\ngithub.com/spf13/cobra.(*Command).execute\n\t/home/runner/go/pkg/mod/github.com/spf13/cobra@v1.6.0/command.go:916\ngithub.com/spf13/cobra.(*Command).ExecuteC\n\t/home/runner/go/pkg/mod/github.com/spf13/cobra@v1.6.0/command.go:1040\ngithub.com/spf13/cobra.(*Command).Execute\n\t/home/runner/go/pkg/mod/github.com/spf13/cobra@v1.6.0/command.go:968\ngithub.com/numaproj/numaflow/cmd/commands.Execute\n\t/home/runner/work/numaflow/numaflow/cmd/commands/root.go:32\nmain.main\n\t/home/runner/work/numaflow/numaflow/cmd/main.go:24\nruntime.main\n\t/opt/hostedtoolcache/go/1.20.12/x64/src/runtime/proc.go:250"}
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x8 pc=0x1bee133]

goroutine 1 [running]:
github.com/numaproj/numaflow/pkg/shared/clients/nats.(*ClientPool).CloseAll(0x363?)
	/home/runner/work/numaflow/numaflow/pkg/shared/clients/nats/client_pool.go:69 +0x33
github.com/numaproj/numaflow/pkg/daemon/server.(*daemonServer).Run(0xc000a49c78, {0x294e2e8, 0xc000dc15f0})
	/home/runner/work/numaflow/numaflow/pkg/daemon/server/daemon_server.go:75 +0x270
github.com/numaproj/numaflow/cmd/commands.NewDaemonServerCommand.func1(0xc000d6a600?, {0x2581745?, 0x1?, 0x1?})
	/home/runner/work/numaflow/numaflow/cmd/commands/daemon_server.go:51 +0x28d
github.com/spf13/cobra.(*Command).execute(0xc000d6a600, {0xc00051c8d0, 0x1, 0x1})
	/home/runner/go/pkg/mod/github.com/spf13/cobra@v1.6.0/command.go:916 +0x862
github.com/spf13/cobra.(*Command).ExecuteC(0x3b90ee0)
	/home/runner/go/pkg/mod/github.com/spf13/cobra@v1.6.0/command.go:1040 +0x3bd
github.com/spf13/cobra.(*Command).Execute(...)
	/home/runner/go/pkg/mod/github.com/spf13/cobra@v1.6.0/command.go:968
github.com/numaproj/numaflow/cmd/commands.Execute()
	/home/runner/work/numaflow/numaflow/cmd/commands/root.go:32 +0x25
main.main()
	/home/runner/work/numaflow/numaflow/cmd/main.go:24 +0x17
```

**To Reproduce**
Steps to reproduce the behavior:

1. (Not sure if related) Remove previous Jetstream ISBSVC.
2. Deploy a redis ISBSVC, (for me external), with the following manifest:
```
➜  ~ kubectl describe -n poseidon-dev isbsvc/default    
Name:         default
Namespace:    poseidon-dev
Labels:       <none>
Annotations:  <none>
API Version:  numaflow.numaproj.io/v1alpha1
Kind:         InterStepBufferService
Metadata:
  Creation Timestamp:  2024-02-19T09:06:10Z
  Generation:          2
  Resource Version:    294759397
  UID:                 e9e4fd63-8e93-40c4-8069-a9cbc2791cf0
Spec:
  Redis:
    External:
      URL:  redis-isbsvc-redis-cluster-0.redis-isbsvc-redis-cluster-headless.poseidon-dev.svc.helios.mycompany.com:6379,redis-isbsvc-redis-cluster-1.redis-isbsvc-redis-cluster-headless.poseidon-dev.svc.helios.mycompany.com:6379,redis-isbsvc-redis-cluster-2.redis-isbsvc-redis-cluster-headless.poseidon-dev.svc.helios.mycompany.com:6379
Status:
  Conditions:
    Last Transition Time:  2024-02-19T09:08:14Z
    Message:               Successful
    Reason:                Successful
    Status:                True
    Type:                  Configured
    Last Transition Time:  2024-02-19T09:08:14Z
    Message:               Successful
    Reason:                Successful
    Status:                True
    Type:                  Deployed
  Config:
    Redis:
      URL:  redis-isbsvc-redis-cluster-0.redis-isbsvc-redis-cluster-headless.poseidon-dev.svc.helios.mycompany.com:6379,redis-isbsvc-redis-cluster-1.redis-isbsvc-redis-cluster-headless.poseidon-dev.svc.helios.mycompany.com:6379,redis-isbsvc-redis-cluster-2.redis-isbsvc-redis-cluster-headless.poseidon-dev.svc.helios.mycompany.com:6379
  Phase:    Running
  Type:     redis
Events:     <none>

```

3. Create a Numaflow pipeline.
```
➜  ~ kubectl describe -n poseidon-dev pipelines.numaflow.numaproj.io simple-pipeline 
Name:         simple-pipeline
Namespace:    poseidon-dev
Labels:       <none>
Annotations:  <none>
API Version:  numaflow.numaproj.io/v1alpha1
Kind:         Pipeline
Metadata:
  Creation Timestamp:  2024-02-19T09:09:06Z
  Finalizers:
    pipeline-controller
  Generation:        2
  Resource Version:  294759750
  UID:               878ebd0f-f5ae-4a27-b445-074818f518cf
Spec:
  Edges:
    From:  in
    To:    cat
    From:  cat
    To:    out
  Lifecycle:
    Delete Grace Period Seconds:  30
    Desired Phase:                Running
    Pause Grace Period Seconds:   30
  Limits:
    Buffer Max Length:   30000
    Buffer Usage Limit:  80
    Read Batch Size:     500
    Read Timeout:        1s
  Vertices:
    Name:  in
    Scale:
    Source:
      Generator:
        Duration:   1s
        Key Count:  5
        Msg Size:   8
        Rpu:        5
        Value:      5
    Name:           cat
    Scale:
    Udf:
      Builtin:
        Name:  cat
    Name:      out
    Scale:
    Sink:
      Log:
  Watermark:
    Disabled:   false
    Max Delay:  0s
Status:
  Conditions:
    Last Transition Time:  2024-02-19T09:09:06Z
    Message:               Successful
    Reason:                Successful
    Status:                True
    Type:                  Configured
    Last Transition Time:  2024-02-19T09:09:06Z
    Message:               Successful
    Reason:                Successful
    Status:                True
    Type:                  Deployed
  Last Updated:            2024-02-19T09:09:06Z
  Phase:                   Running
  Sink Count:              1
  Source Count:            1
  Udf Count:               1
  Vertex Count:            3
Events:
  Type    Reason                     Age    From                 Message
  ----    ------                     ----   ----                 -------
  Normal  CreateVertexSuccess        7m21s  pipeline-controller  Created vertex simple-pipeline-in successfully
  Normal  CreateVertexSuccess        7m21s  pipeline-controller  Created vertex simple-pipeline-cat successfully
  Normal  CreateVertexSuccess        7m21s  pipeline-controller  Created vertex simple-pipeline-out successfully
  Normal  CreateDaemonSvcSuccess     7m21s  pipeline-controller  Succeeded to create daemon service simple-pipeline-daemon-svc
  Normal  CreateDaemonDeploySuccess  7m21s  pipeline-controller  Succeeded to create/recreate daemon deployment simple-pipeline-daemon
```

Note that I also tried to give the isbsvc a specific name different from default (redis), and the issue was still hapenning after configuring my pipeline to use it.

**Expected behavior**
The pipeline use streams from the specific Redis cluster and start properly.

**Environment (please complete the following information):**

- Kubernetes: v1.26.1+k3s1
- Numaflow: v1.1.4

**Additional context**
Jetstream messaging has been working before, it's the first time I've been trying to use the Redis ISBSVC.
I also deleted some `nats-box` container in the pipeline namespace, that I suspect was related to the previous Jetstream ISBSVC. I also tried restarting the containers (like the controller) in `numaflow-system`.

---

>For quick help and support, join our [slack](https://numaproj.slack.com/join/shared_invite/zt-19svuv47m-YKHhsQ~~KK9mBv1E7pNzfg#/shared-invite/email) channel.

<!-- 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:
fix: allow pipeline to start with redis isbsvc (Fixes: #1513)
fixes #1513

Moved all the Jetstream pool initializations inside the Jetstream specific branches of code.
Conditionally avoided certain test suites items on unsupported Redis features that are: reducing, side inputs, or watermarks.

Current tests status:
```
test-e2e: # Passing after disabling TestDropOnFull and TestWatermarkEnabled for redis
test-kafka-e2e: # Untested
test-http-e2e: # Untested
test-nats-e2e: # Untested
test-sdks-e2e: # Untested
test-reduce-one-e2e: # Disabled all individual tests for redis
test-reduce-two-e2e: # Disabled all individual tests for redis
test-api-e2e: # Untested
test-udsource-e2e: # Untested
test-transformer-e2e: # First test of the suite is timing out, are transformers unsported with redis as well ?
test-diamond-e2e: # Passing after disabling TestJoinOnReducePipeline for redis
test-sideinputs-e2e: # Disabled all individual tests for redis
```

Let me know if it's ok to also ignore the transformers tests with the redis ISBSVC.

Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 03346c5e..86b7280a 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -139,7 +139,7 @@ jobs:
       fail-fast: false
       max-parallel: 13
       matrix:
-        driver: [jetstream]
+        driver: [jetstream, redis]
         case: [e2e, diamond-e2e, transformer-e2e, kafka-e2e, http-e2e, nats-e2e, sdks-e2e, reduce-one-e2e, reduce-two-e2e, udsource-e2e, api-e2e, sideinputs-e2e, idle-source-e2e]
     steps:
       - name: Checkout code
diff --git a/pkg/daemon/server/daemon_server.go b/pkg/daemon/server/daemon_server.go
index a958c36d..d4248d74 100644
--- a/pkg/daemon/server/daemon_server.go
+++ b/pkg/daemon/server/daemon_server.go
@@ -69,17 +69,16 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 		natsClientPool *jsclient.ClientPool
 	)
 
-	natsClientPool, err = jsclient.NewClientPool(ctx, jsclient.WithClientPoolSize(1))
-	defer natsClientPool.CloseAll()
-
-	if err != nil {
-		log.Errorw("Failed to get a NATS client pool.", zap.Error(err))
-		return err
-	}
 	switch ds.isbSvcType {
 	case v1alpha1.ISBSvcTypeRedis:
 		isbSvcClient = isbsvc.NewISBRedisSvc(redisclient.NewInClusterRedisClient())
 	case v1alpha1.ISBSvcTypeJetStream:
+		natsClientPool, err = jsclient.NewClientPool(ctx, jsclient.WithClientPoolSize(1))
+		if err != nil {
+			log.Errorw("Failed to get a NATS client pool.", zap.Error(err))
+			return err
+		}
+		defer natsClientPool.CloseAll()
 		isbSvcClient, err = isbsvc.NewISBJetStreamSvc(ds.pipeline.Name, isbsvc.WithJetStreamClient(natsClientPool.NextAvailableClient()))
 		if err != nil {
 			log.Errorw("Failed to get an ISB Service client.", zap.Error(err))
diff --git a/pkg/isb/stores/jetstream/metrics.go b/pkg/isb/stores/jetstream/metrics.go
index d9558ff3..13480c6d 100644
--- a/pkg/isb/stores/jetstream/metrics.go
+++ b/pkg/isb/stores/jetstream/metrics.go
@@ -107,3 +107,10 @@ var isbAckTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Help:      "Processing times of acks for jetstream",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*2, 10),
 }, []string{"buffer"})
+
+// isbDedupCount is used to indicate the number of messages that are duplicate
+var isbDedupCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "isb_jetstream",
+	Name:      "dedup_total",
+	Help:      "Total number of jetstream dedup",
+}, []string{"buffer"})
diff --git a/pkg/isb/stores/jetstream/writer.go b/pkg/isb/stores/jetstream/writer.go
index a680f785..2d10a2b8 100644
--- a/pkg/isb/stores/jetstream/writer.go
+++ b/pkg/isb/stores/jetstream/writer.go
@@ -280,6 +280,9 @@ func (jw *jetStreamWriter) syncWrite(_ context.Context, messages []isb.Message,
 			} else {
 				writeOffsets[idx] = &writeOffset{seq: pubAck.Sequence, partitionIdx: jw.partitionIdx}
 				errs[idx] = nil
+				if pubAck.Duplicate {
+					isbDedupCount.With(metricsLabels).Inc()
+				}
 				jw.log.Debugw("Succeeded to publish a message", zap.String("stream", pubAck.Stream), zap.Any("seq", pubAck.Sequence), zap.Bool("duplicate", pubAck.Duplicate), zap.String("msgID", message.Header.ID), zap.String("domain", pubAck.Domain))
 			}
 		}(msg, index)
diff --git a/pkg/sinks/sink.go b/pkg/sinks/sink.go
index f43fa727..8c7b1490 100644
--- a/pkg/sinks/sink.go
+++ b/pkg/sinks/sink.go
@@ -69,11 +69,6 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	natsClientPool, err = jsclient.NewClientPool(ctx, jsclient.WithClientPoolSize(2))
-	if err != nil {
-		return fmt.Errorf("failed to create a new NATS client pool: %w", err)
-	}
-	defer natsClientPool.CloseAll()
 	// watermark variables no-op initialization
 	// publishWatermark is a map representing a progressor per edge, we are initializing them to a no-op progressor
 	// For sinks, the buffer name is the vertex name
@@ -96,6 +91,13 @@ func (u *SinkProcessor) Start(ctx context.Context) error {
 			readers = append(readers, reader)
 		}
 	case dfv1.ISBSvcTypeJetStream:
+
+		natsClientPool, err = jsclient.NewClientPool(ctx, jsclient.WithClientPoolSize(2))
+		if err != nil {
+			return fmt.Errorf("failed to create a new NATS client pool: %w", err)
+		}
+		defer natsClientPool.CloseAll()
+
 		var readOptions []jetstreamisb.ReadOption
 		if x := u.VertexInstance.Vertex.Spec.Limits; x != nil && x.ReadTimeout != nil {
 			readOptions = append(readOptions, jetstreamisb.WithReadTimeOut(x.ReadTimeout.Duration))
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index 222dfa3d..a24a5320 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -75,13 +75,6 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	// create a new NATS client pool
-	natsClientPool, err := jsclient.NewClientPool(ctx, jsclient.WithClientPoolSize(2))
-	if err != nil {
-		return fmt.Errorf("failed to create a new NATS client pool: %w", err)
-	}
-	defer natsClientPool.CloseAll()
-
 	// watermark variables no-op initialization
 	// create a no op fetcher
 	fetchWatermark, _ := generic.BuildNoOpSourceWatermarkProgressors(sp.VertexInstance.Vertex.GetToBuffers())
@@ -115,6 +108,14 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 			writersMap[e.To] = bufferWriters
 		}
 	case dfv1.ISBSvcTypeJetStream:
+
+		// create a new NATS client pool
+		natsClientPool, err := jsclient.NewClientPool(ctx, jsclient.WithClientPoolSize(2))
+		if err != nil {
+			return fmt.Errorf("failed to create a new NATS client pool: %w", err)
+		}
+		defer natsClientPool.CloseAll()
+
 		for _, e := range sp.VertexInstance.Vertex.Spec.ToEdges {
 			writeOpts := []jetstreamisb.WriteOption{
 				jetstreamisb.WithBufferFullWritingStrategy(e.BufferFullWritingStrategy()),
@@ -212,6 +213,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error {
 		readyCheckers = append(readyCheckers, udsGRPCClient)
 	}
 	maxMessageSize := sharedutil.LookupEnvIntOr(dfv1.EnvGRPCMaxMessageSize, sdkclient.DefaultGRPCMaxMessageSize)
+	var err error
 	if sp.VertexInstance.Vertex.HasUDTransformer() {
 		// Wait for server info to be ready
 		serverInfo, err := sdkserverinfo.SDKServerInfo(sdkserverinfo.WithServerInfoFilePath(sdkserverinfo.SourceTransformerServerInfoFile))
diff --git a/pkg/sources/udsource/grpc_udsource.go b/pkg/sources/udsource/grpc_udsource.go
index a7b348a7..362a9e0e 100644
--- a/pkg/sources/udsource/grpc_udsource.go
+++ b/pkg/sources/udsource/grpc_udsource.go
@@ -28,7 +28,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/sources/udsource/utils"
 )
 
 // GRPCBasedUDSource applies a user-defined source over gRPC
@@ -128,11 +127,13 @@ func (u *GRPCBasedUDSource) ApplyReadFn(ctx context.Context, count int64, timeou
 			}
 			// Convert the datum to ReadMessage and append to the list
 			r := datum.GetResult()
+
+			offset := NewUserDefinedSourceOffset(r.GetOffset())
 			readMessage := &isb.ReadMessage{
 				Message: isb.Message{
 					Header: isb.Header{
 						MessageInfo: isb.MessageInfo{EventTime: r.GetEventTime().AsTime()},
-						ID:          constructMessageID(r),
+						ID:          constructMessageID(offset.String(), r.GetOffset().GetPartitionId()),
 						Keys:        r.GetKeys(),
 						Headers:     r.GetHeaders(),
 					},
@@ -140,7 +141,7 @@ func (u *GRPCBasedUDSource) ApplyReadFn(ctx context.Context, count int64, timeou
 						Payload: r.GetPayload(),
 					},
 				},
-				ReadOffset: utils.ConvertToIsbOffset(r.GetOffset()),
+				ReadOffset: offset,
 			}
 			readMessages = append(readMessages, readMessage)
 		}
@@ -151,7 +152,7 @@ func (u *GRPCBasedUDSource) ApplyReadFn(ctx context.Context, count int64, timeou
 func (u *GRPCBasedUDSource) ApplyAckFn(ctx context.Context, offsets []isb.Offset) error {
 	rOffsets := make([]*sourcepb.Offset, len(offsets))
 	for i, offset := range offsets {
-		rOffsets[i] = utils.ConvertToSourceOffset(offset)
+		rOffsets[i] = ConvertToUserDefinedSourceOffset(offset)
 	}
 	var r = &sourcepb.AckRequest{
 		Request: &sourcepb.AckRequest_Request{
@@ -172,7 +173,7 @@ func (u *GRPCBasedUDSource) ApplyPartitionFn(ctx context.Context) ([]int32, erro
 	return resp.GetResult().GetPartitions(), nil
 }
 
-func constructMessageID(r *sourcepb.ReadResponse_Result) string {
+func constructMessageID(offset string, partitionIdx int32) string {
 	// For a user-defined source, the partition ID plus the offset should be able to uniquely identify a message
-	return fmt.Sprintf("%d-%s", r.GetOffset().GetPartitionId(), string(r.GetOffset().GetOffset()))
+	return fmt.Sprintf("%d-%s", partitionIdx, offset)
 }
diff --git a/pkg/sources/udsource/offset.go b/pkg/sources/udsource/offset.go
new file mode 100644
index 00000000..17937b2e
--- /dev/null
+++ b/pkg/sources/udsource/offset.go
@@ -0,0 +1,70 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package udsource
+
+import (
+	"encoding/base64"
+	"fmt"
+
+	sourcepb "github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1"
+
+	"github.com/numaproj/numaflow/pkg/isb"
+)
+
+// userDefinedSourceOffset is a implementation of isb.Offset from the user-defined source side.
+type userDefinedSourceOffset struct {
+	// NOTE: encodedOffsetStr is base64 encoded string because we use encodedOffsetStr to construct message ID
+	// and message ID is a string.
+	encodedOffsetStr string
+	offset           []byte
+	partitionIdx     int32
+}
+
+func NewUserDefinedSourceOffset(offset *sourcepb.Offset) isb.Offset {
+	return &userDefinedSourceOffset{
+		offset:           offset.GetOffset(),
+		encodedOffsetStr: base64.StdEncoding.EncodeToString(offset.GetOffset()),
+		partitionIdx:     offset.GetPartitionId(),
+	}
+}
+
+func (s *userDefinedSourceOffset) String() string {
+	return fmt.Sprintf("%s-%d", s.encodedOffsetStr, s.partitionIdx)
+}
+
+func (s *userDefinedSourceOffset) PartitionIdx() int32 {
+	return s.partitionIdx
+}
+
+func (s *userDefinedSourceOffset) Sequence() (int64, error) {
+	panic("Sequence is not supported by userDefinedSourceOffset")
+}
+
+func (s *userDefinedSourceOffset) AckIt() error {
+	panic("AckIt is not supported by userDefinedSourceOffset")
+}
+
+func (s *userDefinedSourceOffset) NoAck() error {
+	panic("NoAck is not supported by userDefinedSourceOffset")
+}
+
+func ConvertToUserDefinedSourceOffset(offset isb.Offset) *sourcepb.Offset {
+	return &sourcepb.Offset{
+		PartitionId: offset.PartitionIdx(),
+		Offset:      offset.(*userDefinedSourceOffset).offset,
+	}
+}
diff --git a/pkg/sources/udsource/utils/offset.go b/pkg/sources/udsource/utils/offset.go
deleted file mode 100644
index b7d53029..00000000
--- a/pkg/sources/udsource/utils/offset.go
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
-Copyright 2022 The Numaproj Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package utils
-
-import (
-	sourcepb "github.com/numaproj/numaflow-go/pkg/apis/proto/source/v1"
-
-	"github.com/numaproj/numaflow/pkg/isb"
-)
-
-// simpleSourceOffset is a simple implementation of isb.Offset from the source side.
-type simpleSourceOffset struct {
-	offset       string
-	partitionIdx int32
-}
-
-func NewSimpleSourceOffset(o string, p int32) isb.Offset {
-	return &simpleSourceOffset{
-		offset:       o,
-		partitionIdx: p,
-	}
-}
-
-func (s *simpleSourceOffset) String() string {
-	return s.offset
-}
-
-func (s *simpleSourceOffset) PartitionIdx() int32 {
-	return s.partitionIdx
-}
-
-func (s *simpleSourceOffset) Sequence() (int64, error) {
-	panic("Sequence is not supported by simpleSourceOffset")
-}
-
-func (s *simpleSourceOffset) AckIt() error {
-	panic("AckIt is not supported by simpleSourceOffset")
-}
-
-func (s *simpleSourceOffset) NoAck() error {
-	panic("NoAck is not supported by simpleSourceOffset")
-}
-
-func ConvertToSourceOffset(offset isb.Offset) *sourcepb.Offset {
-	return &sourcepb.Offset{
-		PartitionId: offset.PartitionIdx(),
-		Offset:      []byte(offset.String()),
-	}
-}
-
-func ConvertToIsbOffset(offset *sourcepb.Offset) isb.Offset {
-	return NewSimpleSourceOffset(string(offset.Offset), offset.GetPartitionId())
-}
diff --git a/pkg/udf/map_udf.go b/pkg/udf/map_udf.go
index bf511095..015ff362 100644
--- a/pkg/udf/map_udf.go
+++ b/pkg/udf/map_udf.go
@@ -57,12 +57,6 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	natsClientPool, err := jsclient.NewClientPool(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to create a new NATS client pool: %w", err)
-	}
-	defer natsClientPool.CloseAll()
-
 	fromBuffer := u.VertexInstance.Vertex.OwnedBuffers()
 	log = log.With("protocol", "uds-grpc-map-udf")
 
@@ -80,6 +74,8 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferList(u.VertexInstance.Vertex.GetToBuffers())
 	idleManager = wmb.NewNoOpIdleManager()
 
+	var err error
+
 	// create readers and writers
 	switch u.ISBSvcType {
 	case dfv1.ISBSvcTypeRedis:
@@ -88,6 +84,13 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 			return err
 		}
 	case dfv1.ISBSvcTypeJetStream:
+
+		natsClientPool, err := jsclient.NewClientPool(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to create a new NATS client pool: %w", err)
+		}
+		defer natsClientPool.CloseAll()
+
 		// multiple go routines can share the same set of writers since nats conn is thread safe
 		// https://github.com/nats-io/nats.go/issues/241
 		readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index b11fb455..91239b4c 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -86,12 +86,6 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	natsClientPool, err = jsclient.NewClientPool(ctx)
-	if err != nil {
-		return fmt.Errorf("failed to create a new NATS client pool: %w", err)
-	}
-	defer natsClientPool.CloseAll()
-
 	windowType := u.VertexInstance.Vertex.Spec.UDF.GroupBy.Window
 
 	// based on the window type create the windower, udfApplier and health checker
@@ -198,6 +192,13 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 			return err
 		}
 	case dfv1.ISBSvcTypeJetStream:
+
+		natsClientPool, err = jsclient.NewClientPool(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to create a new NATS client pool: %w", err)
+		}
+		defer natsClientPool.CloseAll()
+
 		readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)
 		if err != nil {
 			return err


Test patch:
diff --git a/pkg/shared/clients/redis/redis_client_test.go b/pkg/shared/clients/redis/redis_client_test.go
index d8e1b686..8c3618d1 100644
--- a/pkg/shared/clients/redis/redis_client_test.go
+++ b/pkg/shared/clients/redis/redis_client_test.go
@@ -25,8 +25,6 @@ import (
 )
 
 func TestNewRedisClient(t *testing.T) {
-	// TODO: fix the test
-	t.SkipNow()
 	ctx := context.TODO()
 	client := NewRedisClient(&redis.UniversalOptions{
 		Addrs: []string{":6379"},
diff --git a/pkg/sources/udsource/grpc_udsource_test.go b/pkg/sources/udsource/grpc_udsource_test.go
index e41521ef..a2afc802 100644
--- a/pkg/sources/udsource/grpc_udsource_test.go
+++ b/pkg/sources/udsource/grpc_udsource_test.go
@@ -37,7 +37,6 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	sourceclient "github.com/numaproj/numaflow/pkg/sdkclient/source"
-	"github.com/numaproj/numaflow/pkg/sources/udsource/utils"
 )
 
 func TestMain(m *testing.M) {
@@ -187,11 +186,13 @@ func Test_gRPCBasedUDSource_ApplyReadWithMockClient(t *testing.T) {
 			},
 		}
 
+		offset := &sourcepb.Offset{Offset: []byte(`test_offset`), PartitionId: 0}
+
 		var TestEventTime = time.Unix(1661169600, 0).UTC()
 		expectedResponse := &sourcepb.ReadResponse{
 			Result: &sourcepb.ReadResponse_Result{
 				Payload:   []byte(`test_payload`),
-				Offset:    &sourcepb.Offset{Offset: []byte(`test_offset`), PartitionId: 0},
+				Offset:    offset,
 				EventTime: timestamppb.New(TestEventTime),
 				Keys:      []string{"test_key"},
 			},
@@ -216,7 +217,7 @@ func Test_gRPCBasedUDSource_ApplyReadWithMockClient(t *testing.T) {
 		assert.Equal(t, 1, len(readMessages))
 		assert.Equal(t, []byte(`test_payload`), readMessages[0].Body.Payload)
 		assert.Equal(t, []string{"test_key"}, readMessages[0].Keys)
-		assert.Equal(t, utils.NewSimpleSourceOffset("test_offset", 0), readMessages[0].ReadOffset)
+		assert.Equal(t, NewUserDefinedSourceOffset(offset), readMessages[0].ReadOffset)
 		assert.Equal(t, TestEventTime, readMessages[0].EventTime)
 	})
 
@@ -268,11 +269,15 @@ func Test_gRPCBasedUDSource_ApplyAckWithMockClient(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
+		offset1 := &sourcepb.Offset{Offset: []byte("test-offset-1"), PartitionId: 0}
+		offset2 := &sourcepb.Offset{Offset: []byte("test-offset-2"), PartitionId: 0}
+
 		mockClient := sourcemock.NewMockSourceClient(ctrl)
 		req := &sourcepb.AckRequest{
 			Request: &sourcepb.AckRequest_Request{
 				Offsets: []*sourcepb.Offset{
-					{Offset: []byte("test-offset-1"), PartitionId: 0}, {Offset: []byte("test-offset-2"), PartitionId: 0},
+					offset1,
+					offset2,
 				},
 			},
 		}
@@ -290,8 +295,8 @@ func Test_gRPCBasedUDSource_ApplyAckWithMockClient(t *testing.T) {
 
 		u := NewMockUDSgRPCBasedUDSource(mockClient)
 		err := u.ApplyAckFn(ctx, []isb.Offset{
-			utils.NewSimpleSourceOffset("test-offset-1", 0),
-			utils.NewSimpleSourceOffset("test-offset-2", 0),
+			NewUserDefinedSourceOffset(offset1),
+			NewUserDefinedSourceOffset(offset2),
 		})
 		assert.NoError(t, err)
 	})
@@ -300,11 +305,15 @@ func Test_gRPCBasedUDSource_ApplyAckWithMockClient(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
+		offset1 := &sourcepb.Offset{Offset: []byte("test-offset-1"), PartitionId: 0}
+		offset2 := &sourcepb.Offset{Offset: []byte("test-offset-2"), PartitionId: 0}
+
 		mockClient := sourcemock.NewMockSourceClient(ctrl)
 		req := &sourcepb.AckRequest{
 			Request: &sourcepb.AckRequest_Request{
 				Offsets: []*sourcepb.Offset{
-					{Offset: []byte("test-offset-1"), PartitionId: 0}, {Offset: []byte("test-offset-2"), PartitionId: 0},
+					offset1,
+					offset2,
 				},
 			},
 		}
@@ -321,8 +330,8 @@ func Test_gRPCBasedUDSource_ApplyAckWithMockClient(t *testing.T) {
 
 		u := NewMockUDSgRPCBasedUDSource(mockClient)
 		err := u.ApplyAckFn(ctx, []isb.Offset{
-			utils.NewSimpleSourceOffset("test-offset-1", 0),
-			utils.NewSimpleSourceOffset("test-offset-2", 0),
+			NewUserDefinedSourceOffset(offset1),
+			NewUserDefinedSourceOffset(offset2),
 		})
 		assert.ErrorIs(t, err, status.New(codes.DeadlineExceeded, "mock test err").Err())
 	})
diff --git a/pkg/sources/udsource/utils/offset_test.go b/pkg/sources/udsource/offset_test.go
similarity index 75%
rename from pkg/sources/udsource/utils/offset_test.go
rename to pkg/sources/udsource/offset_test.go
index 64dae339..bc5c33b8 100644
--- a/pkg/sources/udsource/utils/offset_test.go
+++ b/pkg/sources/udsource/offset_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package utils
+package udsource
 
 import (
 	"testing"
@@ -24,17 +24,18 @@ import (
 )
 
 func TestOffsetConversion(t *testing.T) {
-	testIsbOffset := NewSimpleSourceOffset("test", 0)
-	convertedSrcOffset := ConvertToSourceOffset(testIsbOffset)
-	convertedBackIsbOffset := ConvertToIsbOffset(convertedSrcOffset)
+	offset := &sourcepb.Offset{Offset: []byte("test"), PartitionId: 0}
+	testIsbOffset := NewUserDefinedSourceOffset(offset)
+	ConvertToUserDefinedSourceOffset(testIsbOffset)
+	convertedBackIsbOffset := NewUserDefinedSourceOffset(offset)
 	assert.Equal(t, testIsbOffset.PartitionIdx(), convertedBackIsbOffset.PartitionIdx())
 	assert.Equal(t, testIsbOffset.String(), convertedBackIsbOffset.String())
 	testSrcOffset := &sourcepb.Offset{
 		PartitionId: 0,
 		Offset:      []byte("test"),
 	}
-	convertedIsbOffset := ConvertToIsbOffset(testSrcOffset)
-	convertedBackSrcOffset := ConvertToSourceOffset(convertedIsbOffset)
+	convertedIsbOffset := NewUserDefinedSourceOffset(offset)
+	convertedBackSrcOffset := ConvertToUserDefinedSourceOffset(convertedIsbOffset)
 	assert.Equal(t, testSrcOffset.GetPartitionId(), convertedBackSrcOffset.GetPartitionId())
 	assert.Equal(t, testSrcOffset.GetOffset(), convertedBackSrcOffset.GetOffset())
 }
diff --git a/test/diamond-e2e/diamond_test.go b/test/diamond-e2e/diamond_test.go
index 2e405185..cdb98c77 100644
--- a/test/diamond-e2e/diamond_test.go
+++ b/test/diamond-e2e/diamond_test.go
@@ -21,7 +21,9 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"os"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -35,6 +37,12 @@ type DiamondSuite struct {
 }
 
 func (s *DiamondSuite) TestJoinOnReducePipeline() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := s.Given().Pipeline("@testdata/join-on-reduce-pipeline.yaml").
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index fdb40b3b..c96efdae 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -21,6 +21,7 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -205,6 +206,12 @@ func (s *FunctionalSuite) TestConditionalForwarding() {
 }
 
 func (s *FunctionalSuite) TestDropOnFull() {
+
+	// the drop on full feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	w := s.Given().Pipeline("@testdata/drop-on-full.yaml").
 		When().
 		CreatePipelineAndWait()
@@ -248,6 +255,12 @@ func (s *FunctionalSuite) TestDropOnFull() {
 }
 
 func (s *FunctionalSuite) TestWatermarkEnabled() {
+
+	// the watermark feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	w := s.Given().Pipeline("@testdata/watermark.yaml").
 		When().
 		CreatePipelineAndWait()
diff --git a/test/idle-source-e2e/idle_source_test.go b/test/idle-source-e2e/idle_source_test.go
index d70801bc..5e2ba90e 100644
--- a/test/idle-source-e2e/idle_source_test.go
+++ b/test/idle-source-e2e/idle_source_test.go
@@ -32,7 +32,9 @@ import (
 	"context"
 	"encoding/json"
 	"log"
+	"os"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -46,6 +48,12 @@ type IdleSourceSuite struct {
 }
 
 func (is *IdleSourceSuite) TestIdleKeyedReducePipelineWithHttpSource() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		is.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
 	defer cancel()
 	w := is.Given().Pipeline("@testdata/idle-source-reduce-pipeline.yaml").
@@ -88,6 +96,12 @@ func (is *IdleSourceSuite) TestIdleKeyedReducePipelineWithHttpSource() {
 }
 
 func (is *IdleSourceSuite) TestIdleKeyedReducePipelineWithKafkaSource() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		is.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
 	defer cancel()
 
diff --git a/test/reduce-one-e2e/reduce_one_test.go b/test/reduce-one-e2e/reduce_one_test.go
index 46726aaf..76550e9f 100644
--- a/test/reduce-one-e2e/reduce_one_test.go
+++ b/test/reduce-one-e2e/reduce_one_test.go
@@ -19,7 +19,9 @@ package reduce_one_e2e
 
 import (
 	"context"
+	"os"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -34,6 +36,12 @@ type ReduceSuite struct {
 
 // one reduce vertex (keyed)
 func (r *ReduceSuite) TestSimpleKeyedReducePipeline() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/simple-keyed-reduce-pipeline.yaml").
@@ -74,7 +82,14 @@ func (r *ReduceSuite) TestSimpleKeyedReducePipeline() {
 
 // one reduce vertex(non keyed)
 func (r *ReduceSuite) TestSimpleNonKeyedReducePipeline() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/simple-non-keyed-reduce-pipeline.yaml").
 		When().
@@ -112,7 +127,14 @@ func (r *ReduceSuite) TestSimpleNonKeyedReducePipeline() {
 
 // two reduce vertex(keyed and non keyed)
 func (r *ReduceSuite) TestComplexReducePipelineKeyedNonKeyed() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/complex-reduce-pipeline.yaml").
 		When().
@@ -149,7 +171,14 @@ func (r *ReduceSuite) TestComplexReducePipelineKeyedNonKeyed() {
 }
 
 func (r *ReduceSuite) TestSimpleReducePipelineFailOverUsingWAL() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/simple-reduce-pipeline-wal.yaml").
 		When().
@@ -200,7 +229,14 @@ func (r *ReduceSuite) TestSimpleReducePipelineFailOverUsingWAL() {
 
 // two reduce vertices (keyed and non-keyed) followed by a sliding window vertex
 func (r *ReduceSuite) TestComplexSlidingWindowPipeline() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/complex-sliding-window-pipeline.yaml").
 		When().
diff --git a/test/reduce-two-e2e/reduce_two_test.go b/test/reduce-two-e2e/reduce_two_test.go
index 79f94537..f154e02a 100644
--- a/test/reduce-two-e2e/reduce_two_test.go
+++ b/test/reduce-two-e2e/reduce_two_test.go
@@ -20,7 +20,9 @@ package reduce_two_e2e
 import (
 	"context"
 	"fmt"
+	"os"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -42,6 +44,12 @@ func (r *ReduceSuite) TestReduceStreamJava() {
 }
 
 func (r *ReduceSuite) testReduceStream(lang string) {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := r.Given().Pipeline(fmt.Sprintf("@testdata/reduce-stream/reduce-stream-%s.yaml", lang)).
@@ -82,6 +90,12 @@ func (r *ReduceSuite) testReduceStream(lang string) {
 }
 
 func (r *ReduceSuite) TestSimpleSessionPipeline() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/session-reduce/simple-session-sum-pipeline.yaml").
@@ -131,6 +145,12 @@ func (r *ReduceSuite) TestSimpleSessionKeyedPipelineJava() {
 }
 
 func (r *ReduceSuite) testSimpleSessionKeyedPipeline(lang string) {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := r.Given().Pipeline(fmt.Sprintf("@testdata/session-reduce/simple-session-keyed-counter-pipeline-%s.yaml", lang)).
@@ -179,6 +199,12 @@ func (r *ReduceSuite) testSimpleSessionKeyedPipeline(lang string) {
 }
 
 func (r *ReduceSuite) TestSimpleSessionPipelineFailOverUsingWAL() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		r.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := r.Given().Pipeline("@testdata/session-reduce/simple-session-keyed-counter-pipeline-go.yaml").
diff --git a/test/sdks-e2e/sdks_test.go b/test/sdks-e2e/sdks_test.go
index 9b36250c..569b1940 100644
--- a/test/sdks-e2e/sdks_test.go
+++ b/test/sdks-e2e/sdks_test.go
@@ -21,7 +21,9 @@ package sdks_e2e
 import (
 	"context"
 	"fmt"
+	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -88,6 +90,12 @@ func (s *SDKsSuite) TestMapStreamUDFunctionAndSink() {
 }
 
 func (s *SDKsSuite) TestReduceSDK() {
+
+	// the reduce feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := s.Given().Pipeline("@testdata/simple-keyed-reduce-pipeline.yaml").
@@ -125,6 +133,12 @@ func (s *SDKsSuite) TestReduceSDK() {
 }
 
 func (s *SDKsSuite) TestSourceTransformer() {
+
+	// the transformer feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	var wg sync.WaitGroup
 	wg.Add(3)
 	go func() {
diff --git a/test/sideinputs-e2e/sideinput-e2e_sink_source_test.go b/test/sideinputs-e2e/sideinput-e2e_sink_source_test.go
index 0863d264..dfec5797 100644
--- a/test/sideinputs-e2e/sideinput-e2e_sink_source_test.go
+++ b/test/sideinputs-e2e/sideinput-e2e_sink_source_test.go
@@ -19,6 +19,8 @@ limitations under the License.
 package sideinput_e2e
 
 import (
+	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -38,6 +40,12 @@ func (s *SideInputUDSSuite) setUpTests(pipeLineFile string) *When {
 }
 
 func (s *SideInputUDSSuite) TestSinkWithSideInput() {
+
+	// the side inputs feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	w := s.setUpTests("@testdata/sideinput_sink.yaml")
 	defer w.DeletePipelineAndWait()
 	w.Expect().SinkContains("redis-uds", "e2e-even", SinkCheckWithTimeout(2*time.Minute))
@@ -45,6 +53,12 @@ func (s *SideInputUDSSuite) TestSinkWithSideInput() {
 }
 
 func (s *SideInputUDSSuite) TestSourceWithSideInput() {
+
+	// the side inputs feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	w := s.setUpTests("@testdata/sideinput_source.yaml")
 	defer w.DeletePipelineAndWait()
 	w.Expect().SinkContains("redis-uds", "e2e-even", SinkCheckWithTimeout(2*time.Minute))
diff --git a/test/sideinputs-e2e/sideinput_test.go b/test/sideinputs-e2e/sideinput_test.go
index 98d96d43..22b41da1 100644
--- a/test/sideinputs-e2e/sideinput_test.go
+++ b/test/sideinputs-e2e/sideinput_test.go
@@ -19,7 +19,9 @@ package sideinput_e2e
 
 import (
 	"context"
+	"os"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -33,6 +35,12 @@ type SideInputSuite struct {
 }
 
 func (s *SideInputSuite) TestSimpleMapSideInputPipeline() {
+
+	// the side inputs feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := s.Given().Pipeline("@testdata/map-sideinput-pipeline.yaml").
@@ -69,6 +77,12 @@ func (s *SideInputSuite) TestSimpleMapSideInputPipeline() {
 }
 
 func (s *SideInputSuite) TestSimpleReduceSideInputPipeline() {
+
+	// the side inputs feature is not supported with redis ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	w := s.Given().Pipeline("@testdata/reduce-sideinput-pipeline.yaml").
diff --git a/test/transformer-e2e/transformer_test.go b/test/transformer-e2e/transformer_test.go
index 1b244977..f0f7a69d 100644
--- a/test/transformer-e2e/transformer_test.go
+++ b/test/transformer-e2e/transformer_test.go
@@ -22,6 +22,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
+	"strings"
 	"testing"
 	"time"
 
@@ -86,6 +88,12 @@ func (s *TransformerSuite) TestTimeExtractionFilter() {
 }
 
 func (s *TransformerSuite) TestBuiltinEventTimeExtractor() {
+
+	// this test is skipped for redis as watermark is not supported with this ISBSVC
+	if strings.ToUpper(os.Getenv("ISBSVC")) == "REDIS" {
+		s.T().SkipNow()
+	}
+
 	w := s.Given().Pipeline("@testdata/extract-event-time-from-payload.yaml").
 		When().
 		CreatePipelineAndWait()