Repository: numaproj/numaflow (go)

Issue:
Numaflow UI Search Bar Improvement
# Summary

What change needs making?
The current search bar needs the exact namespace name to find the result. Could we add, for example, incremental search?
Also it would be great if we can have a preview of all available namespaces.

# Use Cases

When would you use this?
I can enter partial name, for example, "test", to get all the available namespaces that have the word "test" in the name.
When I open the UI I can see a list of available namespaces and click the one I want from the list to enter the pipeline view.



Pull request description:
feat: incremental search and namespace preview in search bar
Closes #606 


enhanced the namespace search bar to provide incremental search
provided a view of all available cluster namespaces on the search dropdown

![image](https://user-images.githubusercontent.com/49195734/229053152-fbf462b5-1a0d-4f8e-8ee6-186147aac5b7.png)
![image](https://user-images.githubusercontent.com/49195734/229053209-8f9f1d3d-bc8d-424d-a392-e093346e87d5.png)


Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 60fbb5d7..61b17aa5 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -40,9 +40,9 @@ jobs:
           path: ~/.cache/go-build
           key: ${{ runner.os }}-go-build-v1-${{ hashFiles('**/go.mod') }}
       - name: Setup Golang
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
       - name: Add bins to PATH
         run: |
           echo /home/runner/go/bin >> $GITHUB_PATH
@@ -94,9 +94,9 @@ jobs:
           --health-retries 5
     steps:
       - name: Set up Go 1.x
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
         id: go
       - name: Check out code into the Go module directory
         uses: actions/checkout@v3
@@ -119,9 +119,9 @@ jobs:
     steps:
       - uses: actions/checkout@v3
       - name: Setup Golang
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
       - name: Restore Go build cache
         uses: actions/cache@v3
         with:
@@ -139,7 +139,7 @@ jobs:
       max-parallel: 7
       matrix:
         driver: [jetstream]
-        case: [e2e-suite-1, e2e-suite-2, kafka-e2e, http-e2e, nats-e2e, redis-source-e2e, sdks-e2e, reduce-e2e]
+        case: [e2e-suite-1, e2e-suite-2, kafka-e2e, http-e2e, nats-e2e, redis-streams-e2e, sdks-e2e, reduce-e2e]
     steps:
       - name: Checkout code
         uses: actions/checkout@v3
@@ -149,9 +149,9 @@ jobs:
           path: ~/.cache/go-build
           key: ${{ runner.os }}-go-build-v2-${{ hashFiles('**/go.mod') }}
       - name: Setup Golang
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
       - name: Add bins to PATH
         run: |
           echo /home/runner/go/bin >> $GITHUB_PATH
diff --git a/.github/workflows/nightly-build.yml b/.github/workflows/nightly-build.yml
index 4ee6f61d..2a9b7b89 100644
--- a/.github/workflows/nightly-build.yml
+++ b/.github/workflows/nightly-build.yml
@@ -24,9 +24,9 @@ jobs:
         uses: actions/checkout@v3
 
       - name: Setup Go
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: 1.19
+          go-version: '1.19'
 
       - name: Build binaries
         run: |
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 68be289a..e4b77d49 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -22,9 +22,9 @@ jobs:
         uses: actions/checkout@v3
 
       - name: Setup Go
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: 1.19
+          go-version: '1.19'
 
       - name: Build binaries
         run: |
@@ -99,9 +99,9 @@ jobs:
           else
             echo "VERSION=${GITHUB_REF##*/}" >> $GITHUB_ENV
           fi
-      - uses: actions/setup-go@v3.1.0
+      - uses: actions/setup-go@v4.0.0
         with:
-          go-version: 1.19
+          go-version: '1.19'
       - uses: actions/checkout@v3
       - run: go install sigs.k8s.io/bom/cmd/bom@v0.2.0
       - run: go install github.com/spdx/spdx-sbom-generator/cmd/generator@v0.0.13
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 49137775..702dcc93 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,42 @@
 # Changelog
 
+## v0.7.3 (2023-04-02)
+
+ * [68a1479](https://github.com/numaproj/numaflow/commit/68a14793ff698f68d45e404e0269de32fd3c8ed3) Update manifests to v0.7.3
+ * [a17a41d](https://github.com/numaproj/numaflow/commit/a17a41df7324c7f72b4dcd09592593143c57fc76) feat: integrate serde WAL (#650)
+ * [096c6ac](https://github.com/numaproj/numaflow/commit/096c6acf6b82cf540e387c731b96ddad8e9861a2) fix: unit test (#653)
+ * [cce50ff](https://github.com/numaproj/numaflow/commit/cce50ffd16a39931fe031603e53256ffe593ccae) feat: handle idle watermark for reduce vertex (#627)
+ * [968cc5f](https://github.com/numaproj/numaflow/commit/968cc5f50acec697a482c2bcdd93935c71d202c2) feat: Redis streams source (#628)
+ * [38baae8](https://github.com/numaproj/numaflow/commit/38baae8aa1b4a7152b2aa9d537b6934fc1cb2f11) feat: Adds SASL (plain and gssapi) support for kafka source (#643)
+ * [563b85b](https://github.com/numaproj/numaflow/commit/563b85b19a5ec9f3c58ad86d6cdee1bab4849d12) feat: namespace scope api and disable namespace search on UI (#638)
+ * [c819469](https://github.com/numaproj/numaflow/commit/c8194690f383624fd0e27c75b87069266fb6590d) fix: GetHeadWatermark Logic (#636)
+ * [d37f4db](https://github.com/numaproj/numaflow/commit/d37f4db5c29155bf084ca8b84a642020041c92af) feat: enable edge-level kill switch to drop messages when buffer is full, for the non-reduce forwarder (#634)
+ * [a0dce69](https://github.com/numaproj/numaflow/commit/a0dce69a35c11a54a1c399e77a91bacb1a646882) fix: IdleWatermark unit test (#640)
+ * [f840e1c](https://github.com/numaproj/numaflow/commit/f840e1c30a94d8d0b0644304b2bec71bbb565869) fix: desired replicas should not be greater than pending (#639)
+ * [0ca5630](https://github.com/numaproj/numaflow/commit/0ca5630e6695fe9d526b8a1a000e44211a7610a2) fix: add timeout to the test (#618)
+ * [b049f0b](https://github.com/numaproj/numaflow/commit/b049f0b4e37fd9ff37bfffb18ab39baf5acf0d09) feat: kustomize integration (#637)
+ * [01305ea](https://github.com/numaproj/numaflow/commit/01305ea3ec8e34fdd92a239bcf4ca2f63cc00032) fix: exclude ack pending messages (#631)
+ * [4ec4b3d](https://github.com/numaproj/numaflow/commit/4ec4b3d7f450ed8f7d5da5838223f560750dde03) feat: UI error component (#613)
+ * [90ca505](https://github.com/numaproj/numaflow/commit/90ca505bb027ec9d1ae79c728d6af2a86110edae) fix: do not update status.replicas until pod operation succeeds (#620)
+ * [339db43](https://github.com/numaproj/numaflow/commit/339db43b65136a2339d77b12edca166e78d47845) feat: track and expose oldest work yet to be done to the reduce loop (#617)
+ * [1ce4c38](https://github.com/numaproj/numaflow/commit/1ce4c383caf29384e937f6686e52f6561f8d93d2) feat: handle watermark barrier for map vertex (#607)
+ * [90dbe1f](https://github.com/numaproj/numaflow/commit/90dbe1fab20ad1421706bba772ae9352d16e674b) fix: corrected reduce vertex replica number. Fixes #593 (#616)
+ * [a155f2a](https://github.com/numaproj/numaflow/commit/a155f2af09522b8fb87a14af3559e1a2a74ddb57) feat: add API for pipeline status check. Fixes #407. (#599)
+ * [d9e3a56](https://github.com/numaproj/numaflow/commit/d9e3a56f6ed25f4baf05f22ab058ac393a3ab4a3) chore(deps): bump webpack from 5.74.0 to 5.76.1 in /ui (#610)
+
+### Contributors
+
+ * Alex Ang HH
+ * Derek Wang
+ * Dillen Padhiar
+ * Juanlu Yu
+ * Julie Vogelman
+ * Keran Yang
+ * Vedant Gupta
+ * Vigith Maurice
+ * Yashash H L
+ * dependabot[bot]
+
 ## v0.7.2 (2023-03-13)
 
  * [1196a24](https://github.com/numaproj/numaflow/commit/1196a24b7096273b60c072e803a336c79eef2c5b) Update manifests to v0.7.2
diff --git a/Makefile b/Makefile
index 8f15768b..51f4915f 100644
--- a/Makefile
+++ b/Makefile
@@ -110,7 +110,7 @@ test-e2e-suite-2:
 test-kafka-e2e:
 test-http-e2e:
 test-nats-e2e:
-test-redis-source-e2e:
+test-redis-streams-e2e:
 test-sdks-e2e:
 test-reduce-e2e:
 test-%: 
diff --git a/pkg/reduce/pbq/store/wal/bootstrap.go b/pkg/reduce/pbq/store/wal/bootstrap.go
index 62d4ce07..3dd64b25 100644
--- a/pkg/reduce/pbq/store/wal/bootstrap.go
+++ b/pkg/reduce/pbq/store/wal/bootstrap.go
@@ -17,9 +17,7 @@ limitations under the License.
 package wal
 
 import (
-	"bytes"
 	"encoding/binary"
-	"encoding/gob"
 	"fmt"
 	"io"
 	"os"
@@ -133,14 +131,14 @@ func (w *WAL) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 	return messages, false, nil
 }
 
-// decodeReadMessage decodes the header which is encoded by encodeReadMessage.
+// decodeReadMessage decodes the WALMessage which is encoded by encodeWALMessage.
 func decodeReadMessage(buf io.Reader) (*isb.ReadMessage, int64, error) {
-	entryHeader, err := decodeReadMessageHeader(buf)
+	entryHeader, err := decodeWALMessageHeader(buf)
 	if err != nil {
 		return nil, 0, err
 	}
 
-	entryBody, err := decodeReadMessageBody(buf, entryHeader)
+	entryBody, err := decodeWALBody(buf, entryHeader)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -153,8 +151,8 @@ func decodeReadMessage(buf io.Reader) (*isb.ReadMessage, int64, error) {
 	}, size, nil
 }
 
-// decodeReadMessageHeader decodes the header which is encoded by encodeReadMessageHeader.
-func decodeReadMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error) {
+// decodeWALMessageHeader decodes the WALMessage header which is encoded by encodeWALMessageHeader.
+func decodeWALMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error) {
 	// read the fixed vals
 	var entryHeader = new(readMessageHeaderPreamble)
 	err := binary.Read(buf, binary.LittleEndian, entryHeader)
@@ -164,9 +162,9 @@ func decodeReadMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error)
 	return entryHeader, nil
 }
 
-// decodeReadMessageBody decodes the header which is encoded by encodeReadMessageBody.
+// decodeWALBody decodes the WALMessage body which is encoded by encodeWALMessageBody.
 // Returns errChecksumMismatch to indicate if corrupted entry is found.
-func decodeReadMessageBody(buf io.Reader, entryHeader *readMessageHeaderPreamble) (*isb.Message, error) {
+func decodeWALBody(buf io.Reader, entryHeader *readMessageHeaderPreamble) (*isb.Message, error) {
 	var err error
 
 	body := make([]byte, entryHeader.MessageLen)
@@ -185,8 +183,7 @@ func decodeReadMessageBody(buf io.Reader, entryHeader *readMessageHeaderPreamble
 	}
 
 	var message = new(isb.Message)
-	dec := gob.NewDecoder(bytes.NewReader(body))
-	err = dec.Decode(&message)
+	err = message.UnmarshalBinary(body)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/reduce/pbq/store/wal/segment.go b/pkg/reduce/pbq/store/wal/segment.go
index b05aa029..ef3fb80a 100644
--- a/pkg/reduce/pbq/store/wal/segment.go
+++ b/pkg/reduce/pbq/store/wal/segment.go
@@ -19,7 +19,6 @@ package wal
 import (
 	"bytes"
 	"encoding/binary"
-	"encoding/gob"
 	"fmt"
 	"hash/crc32"
 	"os"
@@ -153,26 +152,27 @@ func (w *WAL) encodeWALHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
 	return buf, err
 }
 
-func (w *WAL) encodeReadMessage(message *isb.ReadMessage) (buf *bytes.Buffer, err error) {
+// encodeWALMessage builds the WAL message.
+func (w *WAL) encodeWALMessage(message *isb.ReadMessage) (buf *bytes.Buffer, err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
 				metrics.LabelPipeline:           w.walStores.pipelineName,
 				metrics.LabelVertex:             w.walStores.vertexName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-				labelErrorKind:                  "encodeReadMessage",
+				labelErrorKind:                  "encodeWALMessage",
 			}).Inc()
 		}
 	}()
 	buf = new(bytes.Buffer)
-	body, err := w.encodeReadMessageBody(message)
+	body, err := w.encodeWALMessageBody(message)
 	if err != nil {
 		return nil, err
 	}
-	checksum := calculateChecksum(body.Bytes())
+	checksum := calculateChecksum(body)
 
 	// Writes the message header
-	header, err := w.encodeReadMessageHeader(message, int64(body.Len()), checksum)
+	header, err := w.encodeWALMessageHeader(message, int64(len(body)), checksum)
 	if err != nil {
 		return nil, err
 	}
@@ -185,17 +185,19 @@ func (w *WAL) encodeReadMessage(message *isb.ReadMessage) (buf *bytes.Buffer, er
 	}
 
 	// Writes the message body
-	wrote, err = buf.Write(body.Bytes())
+	wrote, err = buf.Write(body)
 	if err != nil {
 		return nil, err
 	}
-	if wrote != body.Len() {
+	if wrote != len(body) {
 		return nil, fmt.Errorf("expected to write %d, but wrote only %d, %w", header.Len(), wrote, err)
 	}
 	return buf, nil
 }
 
-func (w *WAL) encodeReadMessageHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
+// encodeWALMessageHeader creates the header of the WAL message. The header is mainly for checksum, verifying the
+// correctness of the WALMessage.
+func (w *WAL) encodeWALMessageHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
 	watermark := message.Watermark.UnixMilli()
 
 	offset, err := message.ReadOffset.Sequence()
@@ -218,27 +220,27 @@ func (w *WAL) encodeReadMessageHeader(message *isb.ReadMessage, messageLen int64
 			metrics.LabelPipeline:           w.walStores.pipelineName,
 			metrics.LabelVertex:             w.walStores.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-			labelErrorKind:                  "encodeReadMessageHeader",
+			labelErrorKind:                  "encodeWALMessageHeader",
 		}).Inc()
 		return nil, err
 	}
 	return buf, nil
 }
 
-func (w *WAL) encodeReadMessageBody(message *isb.ReadMessage) (*bytes.Buffer, error) {
-	m := new(bytes.Buffer)
-	enc := gob.NewEncoder(m)
-	err := enc.Encode(message.Message)
+// encodeWALMessageBody uses ReadMessage.Message field as the body of the WAL message, encodes the
+// ReadMessage.Message, and returns.
+func (w *WAL) encodeWALMessageBody(readMsg *isb.ReadMessage) ([]byte, error) {
+	msgBinary, err := readMsg.Message.MarshalBinary()
 	if err != nil {
 		walErrors.With(map[string]string{
 			metrics.LabelPipeline:           w.walStores.pipelineName,
 			metrics.LabelVertex:             w.walStores.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-			labelErrorKind:                  "encodeReadMessageBody",
+			labelErrorKind:                  "encodeWALMessageBody",
 		}).Inc()
-		return nil, fmt.Errorf("encodeReadMessageBody encountered encode err: %w", err)
+		return nil, fmt.Errorf("encodeWALMessageBody encountered encode err: %w", err)
 	}
-	return m, nil
+	return msgBinary, nil
 }
 
 func calculateChecksum(data []byte) uint32 {
@@ -265,7 +267,7 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 		}
 	}()
 	encodeStart := time.Now()
-	entry, err := w.encodeReadMessage(message)
+	entry, err := w.encodeWALMessage(message)
 	entryEncodeLatency.With(map[string]string{
 		metrics.LabelPipeline:           w.walStores.pipelineName,
 		metrics.LabelVertex:             w.walStores.vertexName,
diff --git a/server/apis/interface.go b/server/apis/interface.go
index ba300ccb..7158bf2a 100644
--- a/server/apis/interface.go
+++ b/server/apis/interface.go
@@ -34,4 +34,5 @@ type Handler interface {
 	GetPipelineEdge(c *gin.Context)
 	GetPipelineWatermarks(c *gin.Context)
 	GetPipelineStatus(c *gin.Context)
+	ListNamespaces(c *gin.Context)
 }
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index da223f13..cc5c7cc6 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -322,6 +322,24 @@ func (h *handler) GetPipelineStatus(c *gin.Context) {
 	c.JSON(http.StatusOK, l)
 }
 
+// ListNamespaces is used to provide all the namespaces that have numaflow pipelines running
+func (h *handler) ListNamespaces(c *gin.Context) {
+	l, err := h.numaflowClient.Pipelines("").List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, err.Error())
+		return
+	}
+	m := make(map[string]bool)
+	for _, pl := range l.Items {
+		m[pl.Namespace] = true
+	}
+	var namespaces []string
+	for k := range m {
+		namespaces = append(namespaces, k)
+	}
+	c.JSON(http.StatusOK, namespaces)
+}
+
 func daemonSvcAddress(ns, pipeline string) string {
 	return fmt.Sprintf("%s.%s.svc.cluster.local:%d", fmt.Sprintf("%s-daemon-svc", pipeline), ns, dfv1.DaemonServicePort)
 }
diff --git a/server/routes/routes.go b/server/routes/routes.go
index 83a9bf9b..a3347a60 100644
--- a/server/routes/routes.go
+++ b/server/routes/routes.go
@@ -60,4 +60,5 @@ func v1Routes(r gin.IRouter) {
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/vertices/:vertex/metrics", handler.GetVertexMetrics)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/watermarks", handler.GetPipelineWatermarks)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/status", handler.GetPipelineStatus)
+	r.GET("/namespaces", handler.ListNamespaces)
 }
diff --git a/test/redis-source-e2e/testdata/redis-source-pipeline-from-beginning.yaml b/test/redis-streams-e2e/testdata/redis-source-pipeline-from-beginning.yaml
similarity index 100%
rename from test/redis-source-e2e/testdata/redis-source-pipeline-from-beginning.yaml
rename to test/redis-streams-e2e/testdata/redis-source-pipeline-from-beginning.yaml
diff --git a/test/redis-source-e2e/testdata/redis-source-pipeline-from-end.yaml b/test/redis-streams-e2e/testdata/redis-source-pipeline-from-end.yaml
similarity index 100%
rename from test/redis-source-e2e/testdata/redis-source-pipeline-from-end.yaml
rename to test/redis-streams-e2e/testdata/redis-source-pipeline-from-end.yaml
diff --git a/ui/src/components/namespaces/Namespaces.tsx b/ui/src/components/namespaces/Namespaces.tsx
index e69765bd..9c6fbc69 100644
--- a/ui/src/components/namespaces/Namespaces.tsx
+++ b/ui/src/components/namespaces/Namespaces.tsx
@@ -17,23 +17,35 @@ import ClearIcon from '@mui/icons-material/Clear'
 import SearchIcon from '@mui/icons-material/Search';
 import { useSystemInfoFetch } from "../../utils/fetchWrappers/systemInfoFetch";
 import {notifyError} from "../../utils/error";
+import { useNamespaceListFetch } from "../../utils/fetchWrappers/namespaceListFetch";
 
 export function Namespaces() {
-    const [nsArr, setnsArr] = useState([]);
+    const [nsSearch, setNsSearch] = useState([]);
+    const [nsCookie, setNsCookie] = useState([]);
     const [value, setValue] = useState("");
     const [disableSearch, setDisableSearch] = useState(false);
     const [namespace, setNamespace] = useState("");
     const { systemInfo, error: systemInfoError } = useSystemInfoFetch();
+    const { namespaceList, error: namespaceListError } = useNamespaceListFetch();
 
     useEffect(() => {
       if (systemInfoError) {
         notifyError([{
-          error: "Failed to fetch the namespace scope installation info",
+          error: "Failed to fetch the system info",
           options: {toastId: "ns-scope", autoClose: false}
         }]);
       }
     }, [systemInfoError])
 
+    useEffect(() => {
+      if (namespaceListError && systemInfo && systemInfo?.namespaced === false) {
+        notifyError([{
+          error: "Failed to fetch the available namespaces",
+          options: {toastId: "ns-search", autoClose: false}
+        }]);
+      }
+    }, [namespaceListError, systemInfo])
+
     useEffect(() => {
       if (systemInfo && systemInfo?.namespaced) {
         setValue(systemInfo?.managedNamespace);
@@ -41,6 +53,7 @@ export function Namespaces() {
         setDisableSearch(true);
       } else if (systemInfo && !systemInfo?.namespaced){
         setDisableSearch(false);
+
         // set namespace value in search box
         let curr_ns = localStorage.getItem("curr_namespace");
         if (!curr_ns) curr_ns = "";
@@ -52,7 +65,7 @@ export function Namespaces() {
         if (!ns_list) ns_list = "";
         const ns_arr = ns_list.split(",");
         ns_arr.pop();
-        setnsArr(ns_arr);
+        setNsCookie(ns_arr);
       }
     }, [systemInfo])
 
@@ -62,27 +75,36 @@ export function Namespaces() {
         setNamespace(namespaceVal);
         if (namespaceVal !== "") {
             let flag = 0;
-            for (let i = 0; i < nsArr.length; i++) {
-                if (namespaceVal === nsArr[i]) {
+            for (let i = 0; i < nsCookie.length; i++) {
+                if (namespaceVal === nsCookie[i]) {
                     flag = 1;
                     break;
                 }
             }
             if (flag === 1) {return;}
-            const arr = nsArr;
+            const arr = nsCookie;
             arr.unshift(namespaceVal);
             if (arr.length > 5) arr.pop();
-            setnsArr(arr);
+            setNsCookie(arr);
             let ns_list = "";
-            for (let i = 0; i < nsArr.length; i++) ns_list += nsArr[i] + ",";
+            for (let i = 0; i < nsCookie.length; i++) ns_list += nsCookie[i] + ",";
             localStorage.setItem("namespaces", ns_list);
         }
     };
 
-    const ns_List = [];
-    nsArr.forEach((namespace) => (
-        ns_List.push({label: namespace})
-    ))
+    useEffect(() => {
+      if (namespaceList) {
+        setNsSearch(namespaceList);
+      }
+    }, [namespaceList]);
+
+    const nsList = [];
+    nsCookie.forEach((namespace) => {
+      nsList.push({label: namespace, type: "Previously Searched"})
+    })
+    nsSearch.forEach((namespace) => {
+      nsList.push({label: namespace, type: "Available Namespaces"})
+    })
 
     const handleKeyPress = e => {
         if (e.key === 'Enter') {
@@ -107,9 +129,9 @@ export function Namespaces() {
                       blurOnSelect
                       disableClearable
                       id="curr_ns"
-                      options={ns_List}
+                      options={nsList}
+                      groupBy={(option) => option.type}
                       sx={{ width: 250, margin: "0 10px" }}
-                      filterOptions={(x) => x}
                       value={value}
                       onChange={(e, v) => {
                           setValue(v.label);
@@ -120,7 +142,6 @@ export function Namespaces() {
                           params.inputProps.onKeyPress = handleKeyPress;
                           return <TextField
                               {...params}
-                              autoComplete="off"
                               label="Namespace"
                               placeholder="enter a namespace"
                               InputLabelProps={{ shrink: true }}
diff --git a/ui/src/utils/fetchWrappers/namespaceListFetch.ts b/ui/src/utils/fetchWrappers/namespaceListFetch.ts
new file mode 100644
index 00000000..607933e5
--- /dev/null
+++ b/ui/src/utils/fetchWrappers/namespaceListFetch.ts
@@ -0,0 +1,33 @@
+import { useEffect, useState } from "react";
+import { useFetch } from "./fetch";
+
+export const useNamespaceListFetch = () => {
+  const [namespaceList, setNamespaceList] = useState<string[] | undefined>(undefined);
+  const [loading, setLoading] = useState<boolean>(true);
+
+  const {
+    data,
+    loading: fetchLoading,
+    error,
+  } = useFetch(
+    `/api/v1/namespaces`
+  );
+
+  useEffect(() => {
+    if (fetchLoading) {
+      setLoading(true);
+      return;
+    }
+    if (error) {
+      setLoading(false);
+      return;
+    }
+    if (data) {
+      setNamespaceList(data);
+      setLoading(false);
+      return;
+    }
+  }, [data, fetchLoading]);
+
+  return { namespaceList, error, loading };
+};


Test patch:
diff --git a/pkg/reduce/pbq/store/wal/segment_test.go b/pkg/reduce/pbq/store/wal/segment_test.go
index 1d23c8b4..721d3aea 100644
--- a/pkg/reduce/pbq/store/wal/segment_test.go
+++ b/pkg/reduce/pbq/store/wal/segment_test.go
@@ -165,7 +165,7 @@ func Test_writeReadEntry(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeWALMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
@@ -210,20 +210,20 @@ func Test_encodeDecodeEntry(t *testing.T) {
 			assert.NoError(t, err)
 			newWal := wal.(*WAL)
 
-			got, err := newWal.encodeReadMessage(tt.message)
-			if !tt.wantErr(t, err, fmt.Sprintf("encodeReadMessage(%v)", tt.message)) {
+			got, err := newWal.encodeWALMessage(tt.message)
+			if !tt.wantErr(t, err, fmt.Sprintf("encodeWALMessage(%v)", tt.message)) {
 				return
 			}
 
 			result, _, err := decodeReadMessage(bytes.NewReader(got.Bytes()))
 			assert.NoError(t, err)
-			assert.Equalf(t, tt.message.Message, result.Message, "encodeReadMessage(%v)", tt.message.Message)
+			assert.Equalf(t, tt.message.Message, result.Message, "encodeWALMessage(%v)", tt.message.Message)
 			expectedOffset, err := tt.message.ReadOffset.Sequence()
 			assert.NoError(t, err)
 			actualOffset, err := result.ReadOffset.Sequence()
 			assert.NoError(t, err)
-			assert.Equalf(t, expectedOffset, actualOffset, "encodeReadMessage(%v)", tt.message.ReadOffset)
-			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeReadMessage(%v)", tt.message.Watermark)
+			assert.Equalf(t, expectedOffset, actualOffset, "encodeWALMessage(%v)", tt.message.ReadOffset)
+			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeWALMessage(%v)", tt.message.Watermark)
 			err = newWal.Close()
 			assert.NoError(t, err)
 		})
@@ -259,7 +259,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	assert.NoError(t, err)
 	err = wal.Write(&message)
 	assert.NoError(t, err)
-	assert.Equal(t, int64(418), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(214), tempWAL.prevSyncedWOffset)
 
 	err = wal.Close()
 	assert.NoError(t, err)
@@ -286,7 +286,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeWALMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
@@ -316,7 +316,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	message := writeMessages[0]
 	storePrevSyncedTime := tempWAL.prevSyncedTime
 	err = wal.Write(&message)
-	assert.Equal(t, int64(228), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(126), tempWAL.prevSyncedWOffset)
 	assert.NotEqual(t, storePrevSyncedTime, tempWAL.prevSyncedTime)
 	assert.NoError(t, err)
 
@@ -352,7 +352,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeWALMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
diff --git a/test/redis-source-e2e/redis_source_test.go b/test/redis-streams-e2e/redis_streams_test.go
similarity index 95%
rename from test/redis-source-e2e/redis_source_test.go
rename to test/redis-streams-e2e/redis_streams_test.go
index 9758d503..e9e58bb6 100644
--- a/test/redis-source-e2e/redis_source_test.go
+++ b/test/redis-streams-e2e/redis_streams_test.go
@@ -16,7 +16,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package redis_source_e2e
+package redis_streams_e2e
 
 import (
 	"encoding/json"
@@ -28,7 +28,6 @@ import (
 	"github.com/numaproj/numaflow/test/fixtures"
 )
 
-//go:generate kubectl -n numaflow-system delete statefulset nats --ignore-not-found=true
 //go:generate kubectl apply -k ../../config/apps/redis -n numaflow-system
 type RedisSourceSuite struct {
 	fixtures.E2ESuite