Repository: numaproj/numaflow
Pull request #574 (2023-02-28 11:49:40)

chore: edge label
Fixing styling of edge label so as to get a consistent view across pipelines

<img width="1486" alt="image" src="https://user-images.githubusercontent.com/49195734/222097593-b467f074-d271-40f3-80f1-f53a28779e56.png">

<img width="1455" alt="image" src="https://user-images.githubusercontent.com/49195734/222097338-005dab03-4e9c-4217-852b-a19387e0f71f.png">


vigith: In the second figure, the labels are on the edge when there are multiple edges. Isn't that inconsistent?
veds-g: > In the second figure, the labels are on the edge when there are multiple edges. Isn't that inconsistent?

So for vertical edges, are we expecting it to be left and right? If yes, then I need to take a look about how to position the watermark and pending in a custom way, because right now i am moving the pieces up and down based on an initial coordinate that is defined by the graph algo  - which I am not sure we can modify. Since right now there is no specific way to identify if an edge is vertical / horizontal, I need to think of a way to identify them and then place the pieces accordingly.
veds-g: > In the second figure, the labels are on the edge when there are multiple edges. Isn't that inconsistent?

Also initially the label was just a single element on the center co-ordinate of the edge so I went ahead with that view in mind

Patch:
diff --git a/config/advanced-install/numaflow-server.yaml b/config/advanced-install/numaflow-server.yaml
index 1f37f332..9133b54e 100644
--- a/config/advanced-install/numaflow-server.yaml
+++ b/config/advanced-install/numaflow-server.yaml
@@ -33,7 +33,6 @@ rules:
   - pods/exec
   - pods/log
   - configmaps
-  - secrets
   - services
   verbs:
   - get
diff --git a/config/cluster-install/rbac/numaflow-server/numaflow-server-cluster-role.yaml b/config/cluster-install/rbac/numaflow-server/numaflow-server-cluster-role.yaml
index 500489f3..8dad780a 100644
--- a/config/cluster-install/rbac/numaflow-server/numaflow-server-cluster-role.yaml
+++ b/config/cluster-install/rbac/numaflow-server/numaflow-server-cluster-role.yaml
@@ -27,7 +27,6 @@ rules:
       - pods/exec
       - pods/log
       - configmaps
-      - secrets
       - services
     verbs:
       - get
diff --git a/config/install.yaml b/config/install.yaml
index 41df0b1a..59126d66 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -11427,7 +11427,6 @@ rules:
   - pods/exec
   - pods/log
   - configmaps
-  - secrets
   - services
   verbs:
   - get
diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
index e6e0bdd5..0c24a72d 100644
--- a/config/namespace-install.yaml
+++ b/config/namespace-install.yaml
@@ -11340,7 +11340,6 @@ rules:
   - pods/exec
   - pods/log
   - configmaps
-  - secrets
   - services
   verbs:
   - get
diff --git a/config/namespace-install/rbac/numaflow-server/numaflow-server-role.yaml b/config/namespace-install/rbac/numaflow-server/numaflow-server-role.yaml
index 9ae78141..cec3d45e 100644
--- a/config/namespace-install/rbac/numaflow-server/numaflow-server-role.yaml
+++ b/config/namespace-install/rbac/numaflow-server/numaflow-server-role.yaml
@@ -27,7 +27,6 @@ rules:
       - pods/exec
       - pods/log
       - configmaps
-      - secrets
       - services
     verbs:
       - get
diff --git a/docs/development/debugging.md b/docs/development/debugging.md
index e035220f..c5cc18bf 100644
--- a/docs/development/debugging.md
+++ b/docs/development/debugging.md
@@ -46,6 +46,15 @@ kubectl port-forward simple-pipeline-p1-0-7jzbn 2469
 go tool pprof -http localhost:8081 https+insecure://localhost:2469/debug/pprof/heap
 ```
 
+`Tracing` is also available with commands below.
+
+```sh
+# Add optional "&seconds=n" to specify the duration.
+curl -skq https://localhost:2469/debug/pprof/trace?debug=1 -o trace.out
+
+go tool trace -http localhost:8082 trace.out
+```
+
 ## Debug Inside the Container
 
 When doing local [development](development.md) using command lines such as `make start`, or `make image`, the built `numaflow` docker image is based on `alpine`, which allows you to execute into the container for debugging with `kubectl exec -it {pod-name} -c {container-name} -- sh`.
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index 3cd1c7e0..1f8e0815 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -36,6 +36,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
+	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 )
 
@@ -103,6 +104,12 @@ func NewInterStepDataForward(vertex *dfv1.Vertex,
 	// Add logger from parent ctx to child context.
 	isdf.ctx = logging.WithLogger(ctx, options.logger)
 
+	if isdf.opts.vertexType == dfv1.VertexTypeSource {
+		if isdf.opts.srcWatermarkPublisher == nil {
+			return nil, fmt.Errorf("failed to assign a non-nil source watermark publisher for source vertex data forwarder")
+		}
+	}
+
 	return &isdf, nil
 }
 
@@ -199,19 +206,6 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		return
 	}
 
-	// fetch watermark if available
-	// TODO: make it async (concurrent and wait later)
-	// let's track only the first element's watermark. This is important because we reassign the watermark we fetch
-	// to all the elements in the batch. If we were to assign last element's watermark, we will wrongly mark on-time data as late.
-	processorWM := isdf.fetchWatermark.GetWatermark(readMessages[0].ReadOffset)
-	for _, m := range readMessages {
-		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(m.Payload)))
-		m.Watermark = time.Time(processorWM)
-		if isdf.opts.vertexType == dfv1.VertexTypeSource && processorWM.After(m.EventTime) { // Set late data at source level
-			m.IsLate = true
-		}
-	}
-
 	// create space for writeMessages specific to each step as we could forward to all the steps too.
 	var messageToStep = make(map[string][]isb.Message)
 	var toBuffers string // logging purpose
@@ -238,9 +232,28 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		}()
 	}
 	concurrentUDFProcessingStart := time.Now()
-	// send UDF processing work to the channel
-	for idx, readMessage := range readMessages {
-		udfResults[idx].readMessage = readMessage
+
+	var processorWM processor.Watermark
+	if isdf.opts.vertexType == dfv1.VertexTypeSource {
+		// for source vertex, the udf is the source data transformer.
+		// in this case, we assign time.UnixMilli(-1) to processorWM.
+		// source data transformer applies filtering and assigns event time to source data, which doesn't require watermarks.
+		processorWM = processor.Watermark(time.UnixMilli(-1))
+	} else {
+		// fetch watermark if available
+		// TODO: make it async (concurrent and wait later)
+		// let's track only the first element's watermark. This is important because we reassign the watermark we fetch
+		// to all the elements in the batch. If we were to assign last element's watermark, we will wrongly mark on-time data as late.
+		processorWM = isdf.fetchWatermark.GetWatermark(readMessages[0].ReadOffset)
+	}
+
+	for idx, m := range readMessages {
+		// emit message size metric
+		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(m.Payload)))
+		// assign watermark to the message. assign time.UnixMilli(-1) as watermark when we are at source vertex.
+		m.Watermark = time.Time(processorWM)
+		// send UDF processing work to the channel
+		udfResults[idx].readMessage = m
 		udfCh <- &udfResults[idx]
 	}
 	// let the go routines know that there is no more work
@@ -250,9 +263,37 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	wg.Wait()
 	isdf.opts.logger.Debugw("concurrent applyUDF completed", zap.Int("concurrency", isdf.opts.udfConcurrency), zap.Duration("took", time.Since(concurrentUDFProcessingStart)))
 	concurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Observe(float64(time.Since(concurrentUDFProcessingStart).Microseconds()))
+	// UDF processing is done.
+
+	// if vertex type is source, it means we have finished the source data transformation.
+	// let's publish source watermark and assign IsLate attribute based on new event time.
+	if isdf.opts.vertexType == dfv1.VertexTypeSource {
+		var writeMessages []*isb.Message
+		var transformedReadMessages []*isb.ReadMessage
+		for _, m := range udfResults {
+			writeMessages = append(writeMessages, m.writeMessages...)
+			for _, message := range m.writeMessages {
+				// we convert each writeMessage to isb.ReadMessage by providing its parent ReadMessage's ReadOffset.
+				// since we use message event time instead of the watermark to determine and publish source watermarks, time.UnixMilli(-1) is assigned to the message watermark.
+				// transformedReadMessages are immediately used below for publishing source watermarks.
+				transformedReadMessages = append(transformedReadMessages, message.ToReadMessage(m.readMessage.ReadOffset, time.UnixMilli(-1)))
+			}
+		}
+		// publish source watermark
+		isdf.opts.srcWatermarkPublisher.PublishSourceWatermarks(transformedReadMessages)
+		// fetch the source watermark again, we might not get the latest watermark because of publishing delay,
+		// but ideally we should use the latest to determine the IsLate attribute.
+		processorWM = isdf.fetchWatermark.GetWatermark(readMessages[0].ReadOffset)
+		// assign isLate
+		for _, m := range writeMessages {
+			if processorWM.After(m.EventTime) { // Set late data at source level
+				m.IsLate = true
+			}
+		}
+	}
 
-	// Now that we know the UDF processing is done, let's figure out which vertex to send the results to.
-	// Update the toBuffer(s) with writeMessages.
+	// let's figure out which vertex to send the results to.
+	// update the toBuffer(s) with writeMessages.
 	for _, m := range udfResults {
 		// look for errors in udf processing, if we see even 1 error let's return. handling partial retrying is not worth ATM.
 		if m.udfError != nil {
@@ -268,6 +309,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 			}
 		}
 	}
+
 	// forward the message to the edge buffer (could be multiple edges)
 	writeOffsets, err := isdf.writeToBuffers(ctx, messageToStep)
 	if err != nil {
diff --git a/pkg/forward/options.go b/pkg/forward/options.go
index c1ff09e9..001b3c91 100644
--- a/pkg/forward/options.go
+++ b/pkg/forward/options.go
@@ -22,6 +22,7 @@ import (
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/isb"
 )
 
 // options for forwarding the message
@@ -34,6 +35,8 @@ type options struct {
 	retryInterval time.Duration
 	// vertexType indicates the type of the vertex
 	vertexType dfv1.VertexType
+	// srcWatermarkPublisher is used to publish source watermark
+	srcWatermarkPublisher isb.SourceWatermarkPublisher
 	// logger is used to pass the logger variable
 	logger *zap.SugaredLogger
 }
@@ -79,3 +82,11 @@ func WithVertexType(t dfv1.VertexType) Option {
 		return nil
 	}
 }
+
+// WithSourceWatermarkPublisher sets the source watermark publisher
+func WithSourceWatermarkPublisher(p isb.SourceWatermarkPublisher) Option {
+	return func(o *options) error {
+		o.srcWatermarkPublisher = p
+		return nil
+	}
+}
diff --git a/pkg/isb/interfaces.go b/pkg/isb/interfaces.go
index 0d225097..20a680f1 100644
--- a/pkg/isb/interfaces.go
+++ b/pkg/isb/interfaces.go
@@ -77,6 +77,11 @@ type BufferWriterInformation interface {
 	GetName() string
 }
 
+// SourceWatermarkPublisher publishes source watermarks based on a list of isb.ReadMessage
+type SourceWatermarkPublisher interface {
+	PublishSourceWatermarks([]*ReadMessage)
+}
+
 // Offset is an interface used in the ReadMessage referencing offset information.
 type Offset interface {
 	// String return the offset identifier
diff --git a/pkg/isb/message.go b/pkg/isb/message.go
index 2a09a700..b6e59350 100644
--- a/pkg/isb/message.go
+++ b/pkg/isb/message.go
@@ -53,3 +53,8 @@ type ReadMessage struct {
 	ReadOffset Offset
 	Watermark  time.Time
 }
+
+// ToReadMessage converts Message to a ReadMessage by providing the offset and watermark
+func (m *Message) ToReadMessage(ot Offset, wm time.Time) *ReadMessage {
+	return &ReadMessage{Message: *m, ReadOffset: ot, Watermark: wm}
+}
diff --git a/pkg/reconciler/pipeline/validate.go b/pkg/reconciler/pipeline/validate.go
index 1703c87a..b8b81461 100644
--- a/pkg/reconciler/pipeline/validate.go
+++ b/pkg/reconciler/pipeline/validate.go
@@ -159,10 +159,6 @@ func ValidatePipeline(pl *dfv1.Pipeline) error {
 				// We only support single partition non-keyed windowing.
 				return fmt.Errorf(`invalid edge (%s - %s), "parallelism" should not > 1 for non-keyed windowing`, e.From, e.To)
 			}
-			if _, ok := sources[e.From]; ok && reduceUdfs[e.To].UDF.GroupBy.Keyed {
-				// Source vertex can not lead to a keyed reduce vertex, because the keys coming from sources are undeterminable.
-				return fmt.Errorf(`invalid spec (%s - %s), "keyed" should not be true for a reduce vertex which has data coming from a source vertex`, e.From, e.To)
-			}
 		}
 		namesInEdges[e.From] = true
 		namesInEdges[e.To] = true
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 13549e19..40441198 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -179,7 +179,7 @@ func NewMemGen(
 		destinations[w.GetName()] = w
 	}
 
-	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(gensrc.logger)}
+	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(gensrc.logger), forward.WithSourceWatermarkPublisher(gensrc)}
 	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
 		if x.ReadBatchSize != nil {
 			forwardOpts = append(forwardOpts, forward.WithReadBatchSize(int64(*x.ReadBatchSize)))
@@ -231,19 +231,19 @@ loop:
 			break loop
 		}
 	}
-	if len(msgs) > 0 {
-		// publish the last message's offset with watermark, this is an optimization to avoid too many insert calls
-		// into the offset timeline store.
-		// Please note that we are inserting the watermark before the data has been persisted into ISB by the forwarder.
-		o := msgs[len(msgs)-1].ReadOffset
-		// use the first event time as watermark to make it conservative
-		nanos, _ := msgs[0].ReadOffset.Sequence()
-		// remove the nanosecond precision
-		mg.sourcePublishWM.PublishWatermark(processor.Watermark(time.Unix(0, nanos)), o)
-	}
 	return msgs, nil
 }
 
+func (mg *memgen) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
+	if len(msgs) <= 0 {
+		return
+	}
+	// use the first event time as watermark to make it conservative
+	nanos, _ := msgs[0].ReadOffset.Sequence()
+	// remove the nanosecond precision
+	mg.sourcePublishWM.PublishWatermark(processor.Watermark(time.Unix(0, nanos)), nil) // Source publisher does not care about the offset
+}
+
 // Ack acknowledges an array of offset.
 func (mg *memgen) Ack(_ context.Context, offsets []isb.Offset) []error {
 	return make([]error, len(offsets))
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 76a1c81f..a2b36d4d 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -194,7 +194,7 @@ func New(
 		destinations[w.GetName()] = w
 	}
 
-	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(h.logger)}
+	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(h.logger), forward.WithSourceWatermarkPublisher(h)}
 	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
 		if x.ReadBatchSize != nil {
 			forwardOpts = append(forwardOpts, forward.WithReadBatchSize(int64(*x.ReadBatchSize)))
@@ -220,29 +220,35 @@ func (h *httpSource) GetName() string {
 }
 
 func (h *httpSource) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error) {
-	msgs := []*isb.ReadMessage{}
-	var oldest time.Time
+	var msgs []*isb.ReadMessage
 	timeout := time.After(h.readTimeout)
 loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-h.messages:
-			if oldest.IsZero() || m.EventTime.Before(oldest) {
-				oldest = m.EventTime
-			}
-			msgs = append(msgs, m)
 			fmt.Println("got a message to read")
 			httpSourceReadCount.With(map[string]string{metrics.LabelVertex: h.name, metrics.LabelPipeline: h.pipelineName}).Inc()
+			msgs = append(msgs, m)
 		case <-timeout:
 			h.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", h.readTimeout), zap.Int("read", len(msgs)))
 			break loop
 		}
 	}
 	h.logger.Debugf("Read %d messages.", len(msgs))
+	return msgs, nil
+}
+
+func (h *httpSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
+	var oldest time.Time
+	for _, m := range msgs {
+		if oldest.IsZero() || m.EventTime.Before(oldest) {
+			oldest = m.EventTime
+		}
+	}
 	if len(msgs) > 0 && !oldest.IsZero() {
-		h.sourcePublishWM.PublishWatermark(processor.Watermark(oldest), msgs[len(msgs)-1].ReadOffset)
+		h.logger.Debugf("Publishing watermark %v to source", oldest)
+		h.sourcePublishWM.PublishWatermark(processor.Watermark(oldest), nil) // Source publisher does not care about the offset
 	}
-	return msgs, nil
 }
 
 func (h *httpSource) Ack(_ context.Context, offsets []isb.Offset) []error {
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index 8173417c..a291ec03 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -122,8 +122,6 @@ func (r *KafkaSource) GetName() string {
 }
 
 func (r *KafkaSource) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error) {
-	// It stores latest timestamps for different partitions
-	oldestTimestamps := make(map[int32]time.Time)
 	msgs := make([]*isb.ReadMessage, 0, count)
 	timeout := time.After(r.readTimeout)
 loop:
@@ -131,23 +129,30 @@ loop:
 		select {
 		case m := <-r.handler.messages:
 			kafkaSourceReadCount.With(map[string]string{metrics.LabelVertex: r.name, metrics.LabelPipeline: r.pipelineName}).Inc()
-			_m := toReadMessage(m)
-			msgs = append(msgs, _m)
-			// Get latest timestamps for different partitions
-			if t, ok := oldestTimestamps[m.Partition]; !ok || m.Timestamp.Before(t) {
-				oldestTimestamps[m.Partition] = m.Timestamp
-			}
+			msgs = append(msgs, toReadMessage(m))
 		case <-timeout:
 			// log that timeout has happened and don't return an error
 			r.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", r.readTimeout))
 			break loop
 		}
 	}
+	return msgs, nil
+}
+
+func (r *KafkaSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
+	// oldestTimestamps stores the latest timestamps for different partitions
+	oldestTimestamps := make(map[int32]time.Time)
+	for _, m := range msgs {
+		// Get latest timestamps for different partitions
+		_, partition, _, _ := offsetFrom(m.ReadOffset.String())
+		if t, ok := oldestTimestamps[partition]; !ok || m.EventTime.Before(t) {
+			oldestTimestamps[partition] = m.EventTime
+		}
+	}
 	for p, t := range oldestTimestamps {
 		publisher := r.loadSourceWatermarkPublisher(p)
 		publisher.PublishWatermark(processor.Watermark(t), nil) // Source publisher does not care about the offset
 	}
-	return msgs, nil
 }
 
 // loadSourceWatermarkPublisher does a lazy load on the watermark publisher
@@ -331,7 +336,7 @@ func NewKafkaSource(
 		destinations[w.GetName()] = w
 	}
 
-	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(kafkasource.logger)}
+	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(kafkasource.logger), forward.WithSourceWatermarkPublisher(kafkasource)}
 	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
 		if x.ReadBatchSize != nil {
 			forwardOpts = append(forwardOpts, forward.WithReadBatchSize(int64(*x.ReadBatchSize)))
diff --git a/pkg/sources/nats/nats.go b/pkg/sources/nats/nats.go
index 569a7bcb..7fd0ff84 100644
--- a/pkg/sources/nats/nats.go
+++ b/pkg/sources/nats/nats.go
@@ -87,7 +87,7 @@ func New(
 	for _, w := range writers {
 		destinations[w.GetName()] = w
 	}
-	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(n.logger)}
+	forwardOpts := []forward.Option{forward.WithVertexType(dfv1.VertexTypeSource), forward.WithLogger(n.logger), forward.WithSourceWatermarkPublisher(n)}
 	if x := vertexInstance.Vertex.Spec.Limits; x != nil {
 		if x.ReadBatchSize != nil {
 			forwardOpts = append(forwardOpts, forward.WithReadBatchSize(int64(*x.ReadBatchSize)))
@@ -218,28 +218,33 @@ func (ns *natsSource) GetName() string {
 }
 
 func (ns *natsSource) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error) {
-	msgs := []*isb.ReadMessage{}
-	var oldest time.Time
+	var msgs []*isb.ReadMessage
 	timeout := time.After(ns.readTimeout)
 loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-ns.messages:
-			if oldest.IsZero() || m.EventTime.Before(oldest) {
-				oldest = m.EventTime
-			}
-			msgs = append(msgs, m)
 			natsSourceReadCount.With(map[string]string{metrics.LabelVertex: ns.name, metrics.LabelPipeline: ns.pipelineName}).Inc()
+			msgs = append(msgs, m)
 		case <-timeout:
 			ns.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", ns.readTimeout), zap.Int("read", len(msgs)))
 			break loop
 		}
 	}
 	ns.logger.Debugf("Read %d messages.", len(msgs))
+	return msgs, nil
+}
+
+func (ns *natsSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
+	var oldest time.Time
+	for _, m := range msgs {
+		if oldest.IsZero() || m.EventTime.Before(oldest) {
+			oldest = m.EventTime
+		}
+	}
 	if len(msgs) > 0 && !oldest.IsZero() {
-		ns.sourcePublishWM.PublishWatermark(processor.Watermark(oldest), msgs[len(msgs)-1].ReadOffset)
+		ns.sourcePublishWM.PublishWatermark(processor.Watermark(oldest), nil) // Source publisher does not care about the offset
 	}
-	return msgs, nil
 }
 
 func (ns *natsSource) Ack(_ context.Context, offsets []isb.Offset) []error {
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index 2daa6881..dd8c329e 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -99,25 +99,6 @@ func (h *handler) GetPipeline(c *gin.Context) {
 	c.JSON(http.StatusOK, pl)
 }
 
-// ListNamespaces is used to provide all the namespaces that have numaflow pipelines running
-func (h *handler) ListNamespaces(c *gin.Context) {
-	ns := c.GetString("namespace")
-	l, err := h.numaflowClient.Pipelines(ns).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
-	}
-	namespaces := []string{}
-	for k := range m {
-		namespaces = append(namespaces, k)
-	}
-	c.JSON(http.StatusOK, namespaces)
-}
-
 // ListInterStepBufferServices is used to provide all the interstepbuffer services in a namespace
 func (h *handler) ListInterStepBufferServices(c *gin.Context) {
 	limit, _ := strconv.ParseInt(c.Query("limit"), 10, 64)
diff --git a/server/cmd/start.go b/server/cmd/start.go
index 42ede476..cfe8ef10 100644
--- a/server/cmd/start.go
+++ b/server/cmd/start.go
@@ -42,9 +42,6 @@ func Start(insecure bool, port int, namespaced bool, managedNamespace string) {
 	router.Use(gin.Logger())
 	router.RedirectTrailingSlash = true
 	router.Use(static.Serve("/", static.LocalFile("./ui/build", true)))
-	if namespaced {
-		router.Use(Namespace(managedNamespace))
-	}
 	routes.Routes(router)
 	router.Use(UrlRewrite(router))
 	server := http.Server{
@@ -89,10 +86,3 @@ func UrlRewrite(r *gin.Engine) gin.HandlerFunc {
 		c.Next()
 	}
 }
-
-func Namespace(ns string) gin.HandlerFunc {
-	return func(c *gin.Context) {
-		c.Set("namespace", ns)
-		c.Next()
-	}
-}
diff --git a/server/routes/routes.go b/server/routes/routes.go
index 837e6f4b..2cec43f3 100644
--- a/server/routes/routes.go
+++ b/server/routes/routes.go
@@ -36,7 +36,6 @@ func v1Routes(r gin.IRouter) {
 	if err != nil {
 		panic(err)
 	}
-	r.GET("/namespaces", handler.ListNamespaces)
 	r.GET("/namespaces/:namespace/pipelines", handler.ListPipelines)
 	r.GET("/namespaces/:namespace/isbsvcs", handler.ListInterStepBufferServices)
 	r.GET("/namespaces/:namespace/isbsvcs/:isbsvc", handler.GetInterStepBufferService)
diff --git a/ui/src/components/namespaces/NamespaceRowContent.tsx b/ui/src/components/namespaces/NamespaceRowContent.tsx
index 2885bf43..f8270502 100644
--- a/ui/src/components/namespaces/NamespaceRowContent.tsx
+++ b/ui/src/components/namespaces/NamespaceRowContent.tsx
@@ -10,31 +10,55 @@ interface NamespaceRowContentProps {
 
 export function NamespaceRowContent(props: NamespaceRowContentProps) {
   const { namespaceId } = props;
-  const { pipelines } = useNamespaceFetch(namespaceId);
-  return (
-    <div className={"NamespaceRowContent"} data-testid="namespace-row-content">
-      <Box
-        sx={{
-          margin: 1,
-          fontWeight: 500,
-          fontSize: "1rem",
-        }}
-      >
-        <List>
-          {pipelines &&
-            pipelines.map((pipelineId) => {
-              return (
-                <ListItem key={pipelineId}>
-                  <Link
-                    to={`/namespaces/${namespaceId}/pipelines/${pipelineId}`}
-                  >
-                    {pipelineId}
-                  </Link>
-                </ListItem>
-              );
-            })}
-        </List>
-      </Box>
-    </div>
-  );
+
+  if (namespaceId === "") {
+      return (
+          <div className={"NamespaceRowContent"} data-testid="namespace-row-content">
+              <Box
+                  sx={{
+                      fontWeight: 500,
+                      fontSize: "1rem",
+                  }}
+              >
+                  <List>
+                      <ListItem>
+                          <div>Search for a namespace to get the pipelines</div>
+                      </ListItem>
+                  </List>
+              </Box>
+          </div>
+      );
+  } else {
+      const {pipelines} = useNamespaceFetch(namespaceId);
+      return (
+          <div className={"NamespaceRowContent"} data-testid="namespace-row-content">
+              <Box
+                  sx={{
+                      fontWeight: 500,
+                      fontSize: "1rem",
+                  }}
+              >
+                  <List>
+                      {pipelines &&
+                          pipelines.map((pipelineId) => {
+                              return (
+                                  <ListItem key={pipelineId}>
+                                      <Link
+                                          to={`/namespaces/${namespaceId}/pipelines/${pipelineId}`}
+                                      >
+                                          {pipelineId}
+                                      </Link>
+                                  </ListItem>
+                              );
+                          })}
+                      {!pipelines.length &&
+                          <ListItem>
+                              <div>No pipelines in the provided namespaces</div>
+                          </ListItem>
+                      }
+                  </List>
+              </Box>
+          </div>
+      );
+  }
 }
diff --git a/ui/src/components/namespaces/Namespaces.tsx b/ui/src/components/namespaces/Namespaces.tsx
index cf222415..bfb43698 100644
--- a/ui/src/components/namespaces/Namespaces.tsx
+++ b/ui/src/components/namespaces/Namespaces.tsx
@@ -1,22 +1,76 @@
-import Table from "@mui/material/Table";
-import TableBody from "@mui/material/TableBody";
-import TableCell from "@mui/material/TableCell";
-import TableContainer from "@mui/material/TableContainer";
-import TableHead from "@mui/material/TableHead";
-import TableRow from "@mui/material/TableRow";
-import Paper from "@mui/material/Paper";
-import { useFetch } from "../../utils/fetchWrappers/fetch";
-import { useState } from "react";
+import {useEffect, useState} from "react";
 import { NamespaceRowContent } from "./NamespaceRowContent";
-import IconButton from "@mui/material/IconButton";
-import KeyboardArrowUpIcon from "@mui/icons-material/KeyboardArrowUp";
-import KeyboardArrowDownIcon from "@mui/icons-material/KeyboardArrowDown";
-import Collapse from "@mui/material/Collapse";
 import "./Namespaces.css";
-
+import {
+    Button,
+    TableBody,
+    Table,
+    TableCell,
+    TableContainer,
+    TableHead,
+    TableRow,
+    Paper,
+    TextField,
+    Autocomplete
+} from "@mui/material";
+import ClearIcon from '@mui/icons-material/Clear'
+import SearchIcon from '@mui/icons-material/Search';
 
 export function Namespaces() {
-  const { data } = useFetch("/api/v1/namespaces");
+    const [nsArr, setnsArr] = useState([]);
+    const [value, setValue] = useState("");
+    const [namespace, setNamespace] = useState("");
+
+    const handle = (namespaceVal) => {
+        localStorage.setItem("curr_namespace", namespaceVal);
+        setValue(namespaceVal);
+        setNamespace(namespaceVal);
+        if (namespaceVal !== "") {
+            let flag = 0;
+            for (let i = 0; i < nsArr.length; i++) {
+                if (namespaceVal === nsArr[i]) {
+                    flag = 1;
+                    break;
+                }
+            }
+            if (flag === 1) {return;}
+            const arr = nsArr;
+            arr.unshift(namespaceVal);
+            if (arr.length > 5) arr.pop();
+            setnsArr(arr);
+            let ns_list = "";
+            for (let i = 0; i < nsArr.length; i++) ns_list += nsArr[i] + ",";
+            localStorage.setItem("namespaces", ns_list);
+        }
+    };
+
+    useEffect(() => {
+        // set namespace value in search box
+        let curr_ns = localStorage.getItem("curr_namespace");
+        if (!curr_ns) curr_ns = "";
+        setValue(curr_ns);
+        setNamespace(curr_ns);
+
+        // set drop-down for previously entered namespaces
+        let ns_list = localStorage.getItem("namespaces");
+        if (!ns_list) ns_list = "";
+        const ns_arr = ns_list.split(",");
+        ns_arr.pop();
+        setnsArr(ns_arr);
+    }, []);
+
+    const ns_List = [];
+    nsArr.forEach((namespace) => (
+        ns_List.push({label: namespace})
+    ))
+
+    const handleKeyPress = e => {
+        if (e.key === 'Enter') {
+            handle(value);
+            setNamespace(value);
+            e.target.blur();
+        }
+    }
 
   return (
     <div className="Namespaces">
@@ -24,71 +78,70 @@ export function Namespaces() {
         <Table aria-label="collapsible table">
           <TableHead>
             <TableRow>
-              <TableCell />
               <TableCell
-                style={{
-                  fontWeight: "bold",
-                  fontSize: "1rem",
-                }}
               >
-                Namespace
+                  <div style={{display: "flex", flexDirection: "row"}}>
+                  <Autocomplete
+                      data-testid="namespace-input"
+                      freeSolo
+                      blurOnSelect
+                      disableClearable
+                      id="curr_ns"
+                      options={ns_List}
+                      sx={{ width: 250, margin: "0 10px" }}
+                      filterOptions={(x) => x}
+                      value={value}
+                      onChange={(e, v) => {
+                          setValue(v.label);
+                          handle(v.label);
+                          setNamespace(v.label);
+                      }}
+                      renderInput={(params) => {
+                          params.inputProps.onKeyPress = handleKeyPress;
+                          return <TextField
+                              {...params}
+                              autoComplete="off"
+                              label="Namespace"
+                              placeholder="enter a namespace"
+                              InputLabelProps={{ shrink: true }}
+                              onChange={e => {setValue(e.target.value)}}
+                          />
+                      }}
+                  />
+                  <Button
+                      data-testid="namespace-search"
+                      onClick={() => {
+                          handle(value);
+                          setNamespace(value);
+                      }}
+                      style={{marginTop: "15px", height: "30px"}}
+                  >
+                      <SearchIcon/>
+                  </Button>
+                  <Button
+                      data-testid="namespace-clear"
+                      onClick={() => {
+                          handle("");
+                          setNamespace("");
+                          setValue("");
+                      }}
+                      style={{marginTop: "15px", height: "30px"}}
+                  >
+                      <ClearIcon/>
+                  </Button>
+                  </div>
               </TableCell>
             </TableRow>
           </TableHead>
-            {data &&
-              data.map((namespace: string) => {
-                return <NamespaceRow key={namespace} namespaceId={namespace} />;
-              })}
+          <TableBody>
+            <TableRow>
+                <TableCell>
+                    <NamespaceRowContent namespaceId={namespace} />
+                </TableCell>
+            </TableRow>
+          </TableBody>
         </Table>
       </TableContainer>
     </div>
   );
 }
-
-interface NamespaceRowProps {
-    namespaceId: string;
-}
-
-export function NamespaceRow(props: NamespaceRowProps) {
-    const { namespaceId } = props;
-    const [open, setOpen] = useState(false);
-
-    return (
-        <TableBody data-testid={`namespace-row-body-${namespaceId}`}>
-            <TableRow sx={{ "& > *": { borderBottom: "unset" } }}>
-                <TableCell>
-                    <IconButton
-                        data-testid={`namespace-row-${namespaceId}`}
-                        aria-label="expand row"
-                        size="small"
-                        onClick={() => setOpen(!open)}
-                    >
-                        {open ? <KeyboardArrowUpIcon /> : <KeyboardArrowDownIcon />}
-                    </IconButton>
-                </TableCell>
-                <TableCell
-                    sx={{
-                        fontWeight: 500,
-                        fontSize: "1rem",
-                    }}
-                    component="th"
-                    scope="row"
-                >
-                    {namespaceId}
-                </TableCell>
-            </TableRow>
-            <TableRow>
-                <TableCell
-                    data-testid="table-cell"
-                    style={{ paddingBottom: 0, paddingTop: 0 }}
-                    colSpan={6}
-                >
-                    <Collapse in={open} timeout="auto" unmountOnExit>
-                        <NamespaceRowContent namespaceId={namespaceId} />
-                    </Collapse>
-                </TableCell>
-            </TableRow>
-        </TableBody>
-    );
-}
-
diff --git a/ui/src/components/pipeline/edgeinfo/EdgeInfo.tsx b/ui/src/components/pipeline/edgeinfo/EdgeInfo.tsx
index bce04b54..2b2504f8 100644
--- a/ui/src/components/pipeline/edgeinfo/EdgeInfo.tsx
+++ b/ui/src/components/pipeline/edgeinfo/EdgeInfo.tsx
@@ -106,7 +106,7 @@ export default function EdgeInfo(props: EdgeInfoProps) {
                     if (typeof singleEdge?.data?.bufferUsage !== "undefined") {
                       bufferUsage = (singleEdge?.data?.bufferUsage * 100).toFixed(2);
                     }
-                    return <TableRow>
+                    return <TableRow key={singleEdge.id}>
                       <TableCell>{singleEdge.data.bufferName.slice(singleEdge.data.bufferName.indexOf('-') + 1)}</TableCell>
                       <TableCell data-testid="isFull">{isFull}</TableCell>
                       <TableCell data-testid="ackPending">{singleEdge.data.ackPending}</TableCell>
@@ -155,6 +155,7 @@ export default function EdgeInfo(props: EdgeInfoProps) {
           {edges.map((singleEdge) => {
             if (singleEdge?.data?.conditions && (singleEdge?.source == edge?.data?.fromVertex && singleEdge?.target == edge?.data?.toVertex)) {
               return <ReactJson
+                  key={singleEdge.id}
                   name="conditions"
                   enableClipboard={handleCopy}
                   theme="apathy:inverted"
diff --git a/ui/src/components/pipeline/graph/CustomEdge.css b/ui/src/components/pipeline/graph/CustomEdge.css
index 24d78a32..c7762aba 100644
--- a/ui/src/components/pipeline/graph/CustomEdge.css
+++ b/ui/src/components/pipeline/graph/CustomEdge.css
@@ -4,15 +4,12 @@
 
 .edge-watermark {
     width: fit-content;
-    height: 12px;
-    background: #ccd6dd;
-    border: 1px solid #8fa4b1;
+    background: white;
     position: absolute;
-    border-radius: 5px;
+    border-radius: 2px;
     padding: 0px 5px;
     font-size: 0.8em;
     text-transform: lowercase;
-    margin-top: -20px;
     text-align: center;
     font-family: "IBM Plex Sans", sans-serif;
     pointer-events: all;
@@ -20,15 +17,12 @@
 
 .edge-backpressure {
     width: fit-content;
-    height: 12px;
-    background: #ccd6dd;
-    border: 1px solid #8fa4b1;
+    background: white;
     position: absolute;
-    border-radius: 5px;
+    border-radius: 2px;
     padding: 0px 5px;
     font-size: 0.8em;
     text-transform: lowercase;
-    margin-top: 8px;
     text-align: center;
     font-family: "IBM Plex Sans", sans-serif;
     pointer-events: all;
diff --git a/ui/src/components/pipeline/graph/CustomEdge.tsx b/ui/src/components/pipeline/graph/CustomEdge.tsx
index f5fb67f8..a2e4de88 100644
--- a/ui/src/components/pipeline/graph/CustomEdge.tsx
+++ b/ui/src/components/pipeline/graph/CustomEdge.tsx
@@ -31,6 +31,15 @@ const CustomEdge: FC<EdgeProps> = ({
             fontWeight: 700,
         };
     }
+    let wmStyle;
+    let pendingStyle;
+    if (sourceY === targetY) {
+        wmStyle = { marginTop: "-12px" }
+        pendingStyle = { marginTop: "1px" }
+    } else {
+        wmStyle = { right: "1px" }
+        pendingStyle = { left: "1px" }
+    }
 
     return (
         <>
@@ -44,10 +53,9 @@ const CustomEdge: FC<EdgeProps> = ({
                         fontFamily: "IBM Plex Sans",
                         fontWeight: 400,
                         fontSize: "0.6rem",
-                        width: "40px",
+                        width: "fit-content",
                         display: "flex",
                         justifyContent: "center",
-
                     }}
                     className="nodrag nopan"
                 >
@@ -64,7 +72,7 @@ const CustomEdge: FC<EdgeProps> = ({
                             arrow
                             placement={"top"}
                         >
-                            <div className={"edge-watermark"}>
+                            <div className={"edge-watermark"} style={wmStyle}>
                                 {Math.min.apply(null, data.edgeWatermark.watermarks)}
                             </div>
                         </Tooltip>
@@ -78,7 +86,7 @@ const CustomEdge: FC<EdgeProps> = ({
                         arrow
                         placement={"bottom"}
                     >
-                        <div className={"edge-backpressure"}>
+                        <div className={"edge-backpressure"} style={pendingStyle}>
                             {data.backpressureLabel}
                         </div>
                     </Tooltip>


Test patch:
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index f005429b..3807408f 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -49,7 +49,8 @@ const (
 )
 
 var (
-	testStartTime = time.Unix(1636470000, 0).UTC()
+	testStartTime       = time.Unix(1636470000, 0).UTC()
+	testSourceWatermark = time.Unix(1636460000, 0).UTC()
 )
 
 type testForwardFetcher struct {
@@ -64,7 +65,7 @@ func (t testForwardFetcher) Close() error {
 // GetWatermark uses current time as the watermark because we want to make sure
 // the test publisher is publishing watermark
 func (t testForwardFetcher) GetWatermark(_ isb.Offset) processor.Watermark {
-	return processor.Watermark(time.Now())
+	return processor.Watermark(testSourceWatermark)
 }
 
 func (t testForwardFetcher) GetHeadWatermark() processor.Watermark {
@@ -134,6 +135,111 @@ func TestNewInterStepDataForward(t *testing.T) {
 	<-stopped
 }
 
+// mySourceForwardTest tests source data transformer by updating message event time, and then verifying new event time and IsLate assignments.
+type mySourceForwardTest struct {
+}
+
+func (f mySourceForwardTest) WhereTo(_ string) ([]string, error) {
+	return []string{"to1"}, nil
+}
+
+// for source forward test, in transformer, we assign to the message a new event time that is before test source watermark,
+// such that we can verify message IsLate attribute gets set to true.
+var testSourceNewEventTime = testSourceWatermark.Add(time.Duration(-1) * time.Minute)
+
+func (f mySourceForwardTest) ApplyMap(ctx context.Context, message *isb.ReadMessage) ([]*isb.Message, error) {
+	return func(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb.Message, error) {
+		_ = ctx
+		offset := readMessage.ReadOffset
+		payload := readMessage.Body.Payload
+		parentPaneInfo := readMessage.PaneInfo
+
+		// apply source data transformer
+		_ = payload
+		// copy the payload
+		result := payload
+		// assign new event time
+		parentPaneInfo.EventTime = testSourceNewEventTime
+		var key string
+
+		writeMessage := &isb.Message{
+			Header: isb.Header{
+				PaneInfo: parentPaneInfo,
+				ID:       offset.String(),
+				Key:      key,
+			},
+			Body: isb.Body{
+				Payload: result,
+			},
+		}
+		return []*isb.Message{writeMessage}, nil
+	}(ctx, message)
+}
+
+// TestSourceWatermarkPublisher is a dummy implementation of isb.SourceWatermarkPublisher interface
+type TestSourceWatermarkPublisher struct {
+}
+
+func (p TestSourceWatermarkPublisher) PublishSourceWatermarks([]*isb.ReadMessage) {
+	// PublishSourceWatermarks is not tested in forwarder_test.go
+}
+
+func TestSourceInterStepDataForward(t *testing.T) {
+	fromStep := simplebuffer.NewInMemoryBuffer("from", 25)
+	to1 := simplebuffer.NewInMemoryBuffer("to1", 10, simplebuffer.WithReadTimeOut(time.Second*10))
+	toSteps := map[string]isb.BufferWriter{
+		"to1": to1,
+	}
+	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
+		PipelineName: "testPipeline",
+		AbstractVertex: dfv1.AbstractVertex{
+			Name: "testVertex",
+		},
+	}}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime)
+	fetchWatermark := testForwardFetcher{}
+	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
+
+	// verify if source watermark publisher is not set, NewInterStepDataForward throws.
+	failedForwarder, err := NewInterStepDataForward(vertex, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeSource))
+	assert.True(t, failedForwarder == nil)
+	assert.Error(t, err)
+	assert.Equal(t, fmt.Errorf("failed to assign a non-nil source watermark publisher for source vertex data forwarder"), err)
+
+	// create a valid source forwarder
+	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, mySourceForwardTest{}, mySourceForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(5), WithVertexType(dfv1.VertexTypeSource), WithSourceWatermarkPublisher(TestSourceWatermarkPublisher{}))
+	assert.NoError(t, err)
+	assert.False(t, to1.IsFull())
+	assert.True(t, to1.IsEmpty())
+
+	stopped := f.Start()
+	count := int64(2)
+	// write some data
+	_, errs := fromStep.Write(ctx, writeMessages[0:count])
+	assert.Equal(t, make([]error, count), errs)
+
+	// read some data
+	readMessages, err := to1.Read(ctx, count)
+	assert.NoError(t, err, "expected no error")
+	assert.Len(t, readMessages, int(count))
+	assert.Equal(t, []interface{}{writeMessages[0].Header.Key, writeMessages[1].Header.Key}, []interface{}{readMessages[0].Header.Key, readMessages[1].Header.Key})
+	assert.Equal(t, []interface{}{writeMessages[0].Header.ID, writeMessages[1].Header.ID}, []interface{}{readMessages[0].Header.ID, readMessages[1].Header.ID})
+	for _, m := range readMessages {
+		// verify new event time gets assigned to messages.
+		assert.Equal(t, testSourceNewEventTime, m.EventTime)
+		// verify messages are marked as late because of event time update.
+		assert.Equal(t, true, m.IsLate)
+	}
+	f.Stop()
+	time.Sleep(1 * time.Millisecond)
+	// only for shutdown will work as from buffer is not empty
+	f.ForceStop()
+	<-stopped
+}
+
 // TestWriteToBufferError explicitly tests the case of retrying failed messages
 func TestWriteToBufferError(t *testing.T) {
 	fromStep := simplebuffer.NewInMemoryBuffer("from", 25)
diff --git a/pkg/isb/testutils/rw.go b/pkg/isb/testutils/rw.go
index cd15e7d5..2eb02ded 100644
--- a/pkg/isb/testutils/rw.go
+++ b/pkg/isb/testutils/rw.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/go-redis/redis/v8"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 )
diff --git a/pkg/reconciler/pipeline/validate_test.go b/pkg/reconciler/pipeline/validate_test.go
index 9ed3bc01..b4d531dc 100644
--- a/pkg/reconciler/pipeline/validate_test.go
+++ b/pkg/reconciler/pipeline/validate_test.go
@@ -377,11 +377,6 @@ func TestValidateReducePipeline(t *testing.T) {
 		err = ValidatePipeline(testObj)
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), `"parallelism" is < 1`)
-		testObj.Spec.Edges[0].Parallelism = pointer.Int32(1)
-		testObj.Spec.Vertices[1].UDF.GroupBy.Keyed = true
-		err = ValidatePipeline(testObj)
-		assert.Error(t, err)
-		assert.Contains(t, err.Error(), `"keyed" should not be true for a reduce vertex which has data coming from a source vertex`)
 	})
 
 	t.Run("no storage", func(t *testing.T) {
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index c9321649..7c58ccc2 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -177,15 +177,48 @@ func (s *FunctionalSuite) TestBuiltinEventTimeExtractor() {
 	pipelineName := "extract-event-time"
 
 	// wait for all the pods to come up
-	w.Expect().VertexPodsRunning()
+	w.Expect().VertexPodsRunning().DaemonPodsRunning()
+
+	defer w.DaemonPodPortForward(pipelineName, 1234, dfv1.DaemonServicePort).
+		TerminateAllPodPortForwards()
+
+	// Use daemon client to verify watermark propagation.
+	client, err := daemonclient.NewDaemonServiceClient("localhost:1234")
+	assert.NoError(s.T(), err)
+	defer func() {
+		_ = client.Close()
+	}()
 
 	// In this test, we send a message with event time being now, apply event time extractor and verify from log that the message event time gets updated.
 	timeNow := strconv.FormatInt(time.Now().UnixMilli(), 10)
-	testMsg := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-02-18T21:54:42.123Z"}]}`
-
-	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte(testMsg)).WithHeader("X-Numaflow-Event-Time", timeNow))
+	testMsgOne := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-01-18T21:54:42.123Z"}]}`
+	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte(testMsgOne)).WithHeader("X-Numaflow-Event-Time", timeNow))
+	w.Expect().VertexPodLogContains("out", fmt.Sprintf("EventTime -  %d", time.Date(2021, 1, 18, 21, 54, 42, 123000000, time.UTC).UnixMilli()), PodLogCheckOptionWithCount(1))
+
+	// Verify watermark is generated based on the new event time.
+	testMsgTwo := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-02-18T21:54:42.123Z"}]}`
+	testMsgThree := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-03-18T21:54:42.123Z"}]}`
+	testMsgFour := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-04-18T21:54:42.123Z"}]}`
+	testMsgFive := `{"test": 21, "item": [{"id": 1, "name": "numa", "time": "2022-02-18T21:54:42.123Z"},{"id": 2, "name": "numa", "time": "2021-05-18T21:54:42.123Z"}]}`
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
 
-	w.Expect().VertexPodLogContains("out", fmt.Sprintf("EventTime -  %d", time.Date(2021, 2, 18, 21, 54, 42, 123000000, time.UTC).UnixMilli()), PodLogCheckOptionWithCount(1))
+	// TODO - Figure out a better way to wait for watermark to propagate.
+	time.Sleep(time.Second * 2)
+	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte(testMsgTwo)).WithHeader("X-Numaflow-Event-Time", timeNow))
+	time.Sleep(time.Second * 2)
+	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte(testMsgThree)).WithHeader("X-Numaflow-Event-Time", timeNow))
+	time.Sleep(time.Second * 2)
+	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte(testMsgFour)).WithHeader("X-Numaflow-Event-Time", timeNow))
+	time.Sleep(time.Second * 2)
+	w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte(testMsgFive)).WithHeader("X-Numaflow-Event-Time", timeNow))
+	time.Sleep(time.Second * 2)
+
+	wm, err := client.GetPipelineWatermarks(ctx, pipelineName)
+	assert.NoError(s.T(), err)
+	edgeWM := wm[0].Watermarks[0]
+	// Watermark propagation can delay, we consider the test as passed as long as the retrieved watermark matches one of the assigned event times.
+	assert.True(s.T(), edgeWM == time.Date(2021, 4, 18, 21, 54, 42, 123000000, time.UTC).UnixMilli() || edgeWM == time.Date(2021, 3, 18, 21, 54, 42, 123000000, time.UTC).UnixMilli() || edgeWM == time.Date(2021, 2, 18, 21, 54, 42, 123000000, time.UTC).UnixMilli())
 }
 
 func (s *FunctionalSuite) TestConditionalForwarding() {
diff --git a/ui/src/components/namespaces/Namespaces.test.tsx b/ui/src/components/namespaces/Namespaces.test.tsx
index 22677fa4..d4e38340 100644
--- a/ui/src/components/namespaces/Namespaces.test.tsx
+++ b/ui/src/components/namespaces/Namespaces.test.tsx
@@ -1,26 +1,20 @@
 import {Namespaces} from "./Namespaces"
 import {fireEvent, render, screen, waitFor} from "@testing-library/react"
-import {useFetch} from "../../utils/fetchWrappers/fetch";
 import {useNamespaceFetch} from "../../utils/fetchWrappers/namespaceFetch";
 import {BrowserRouter} from "react-router-dom";
 
-jest.mock("../../utils/fetchWrappers/fetch");
-const mockedUseFetch = useFetch as jest.MockedFunction<typeof useFetch>;
 jest.mock("../../utils/fetchWrappers/namespaceFetch");
 const mockedUseNamespaceFetch = useNamespaceFetch as jest.MockedFunction<typeof useNamespaceFetch>;
 
 
 describe("Namespaces screen", () => {
     it("Load namespaces screen", async () => {
-        mockedUseFetch.mockReturnValueOnce({data: ["namespace-1", "namespace-2"], error: false, loading: false});
         mockedUseNamespaceFetch.mockReturnValue({pipelines: ["simple-pipeline"], error: false, loading: false});
         render(<BrowserRouter><Namespaces/></BrowserRouter>)
-        expect(screen.getByText("namespace-1")).toBeVisible();
-        expect(screen.getByText("namespace-2")).toBeVisible();
-        fireEvent.click(screen.getByTestId("namespace-row-namespace-1"))
-        await waitFor(() => expect(screen.getByTestId("namespace-row-body-namespace-1")).toBeVisible());
+        fireEvent.click(screen.getByTestId("namespace-search"))
         await waitFor(() => expect(screen.getByTestId("namespace-row-content")).toBeVisible());
-
-
+        fireEvent.keyDown(screen.getByTestId("namespace-input"),{ key: "Enter"})
+        await waitFor(() => expect(screen.getByTestId("namespace-row-content")).toBeVisible());
+        fireEvent.click(screen.getByTestId("namespace-clear"))
     })
 })