Repository: numaproj/numaflow
Pull request #557 (2023-02-15 20:38:26)

feat: implement watermark propagation for source data transformer
See https://github.com/numaproj/numaflow/pull/544#issuecomment-1432009459 about this change.

Explain what this PR does.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


KeranYang: @yhl25 @vigith @ashwinidulams @whynowy @chromevoid I performed some manual tests to verify the functionality of this PR and got the following results, which gave me extra confidence to merge this PR. Please review and see if you are good with it and we can target to merge it some time this week. Thanks!

# Source Data Transformer Test Note

This note documents the manual tests I performed to verify the functionality of numaflow [[PR #557](https://github.com/numaproj/numaflow/pull/557)](https://github.com/numaproj/numaflow/pull/557) . The testing pipeline utilized was the [[osam-apigw-numa waterfront at pre-prod stage](https://github.intuit.com/oss-analytics/osam-apigw-numa/tree/master/spec/wf/ppd)](https://github.intuit.com/oss-analytics/osam-apigw-numa/tree/master/spec/wf/ppd).

### Test 1 - Regression Test

During this test, I verified that the functionality of the existing features was not impacted by my PR, by comparing metrics before and after applying my changes, using pipeline [[apigw-wf-go.yaml](https://github.intuit.com/oss-analytics/osam-apigw-numa/blob/master/spec/wf/ppd/apigw-wf-go.yaml)](https://github.intuit.com/oss-analytics/osam-apigw-numa/blob/master/spec/wf/ppd/apigw-wf-go.yaml).

[[Metric board](https://tinyurl.com/yckrzx5c)](https://tinyurl.com/yckrzx5c)

* Deployed the change at Friday noon.
* Monitored the continuity of metric propagations.
* No performance(memory usage/latency/throughput) downgrade.

### Test 2 - Performance Test

During this test, I moved the filtering UDF to source data transformer and compared metrics before and after the change. This is to verify saving one filtering vertex doesn't introduce any performance downgrade.

* Worked with Yashash and verified no performance downgrade.

### Test 3 - Watermark Assignment Test

The goal of this test is to verify after extracing message event time from payload, source vertex is able to use the new event time to publish watermark.

The approach I used to verify is that I created two pipeline with watermark maxDelay disabled. One pipeline is the original pipeline where we use Kafka event time, the other pipeline extract new event time from payload. With watermark maxDelay disabled, we started seeing late messages. I monitored the `no. of dropped messages` metric and observed that the event-time-from-payload pipeline dropped way more messages than original one, proving that the new event time was used to generate watermark.

**Notes**: The reason why extracting event time from payload introduces more late messages is because the event time from payload is at second level while the kafka event time is at millisecond level. With the event-time-from-payload pipeline, multiple messages are group together to share same event time and if within a group, one of the message is late, then the entire group are marked as late.

### Test 4 - Output comparison

During this test, I verified the output of the transformer pipeline is the same as the original pipeline, using Splunk. I spot-checked three [window, key] combinations and verified both pipelines are producing exact same data.

#### Splunk Search Criteria

Pipeline **apigw-wf-go** without transformer:

```
(index=iks) ((sourcetype=eventrouter AND event.involvedObject.namespace=oss-analytics-osamapigwnuma-usw2-qal) OR ( kubernetes_namespace=oss-analytics-osamapigwnuma-usw2-qal)) kubernetes_cluster=ip-msaas-ppd-usw2-k8s host="apigw-wf-go-sink-0-oh3lq" "Intuit.reporting.spreadsheetsyncadmin:Intuit.identity.authz.authzdecisionservice"
```

Pipeline **extract-event-time-from-payload** with transformer:

```
(index=iks) ((sourcetype=eventrouter AND event.involvedObject.namespace=oss-analytics-osamapigwnuma-usw2-qal) OR ( kubernetes_namespace=oss-analytics-osamapigwnuma-usw2-qal)) kubernetes_cluster=ip-msaas-ppd-usw2-k8s host="extract-event-time-from-payload-sink-0-zur9f" "Intuit.reporting.spreadsheetsyncadmin:Intuit.identity.authz.authzdecisionservice"
```

#### Results

For the same window, outputs from both pipelines are exactly same.

https://www.diffchecker.com/A8y2rbQN/

https://www.diffchecker.com/wr1h1QuY/

#### Using a different key

```
"Intuit.identity.relationship.icnccpa:Intuit.platform.servicesplatform.config:servicetoservice"
```

For the same window, outputs from both pipelineS are exactly same.

https://www.diffchecker.com/dXQxEdSz/
KeranYang: @yhl25 @vigith @ashwinidulams @whynowy @chromevoid I performed some manual tests to verify the functionality of this PR and got the following results, which gave me extra confidence to merge this PR. Please review and see if you are good with it and we can target to merge it some time this week. Thanks!

# Source Data Transformer Test Note

This note documents the manual tests I performed to verify the functionality of numaflow [[PR #557](https://github.com/numaproj/numaflow/pull/557)](https://github.com/numaproj/numaflow/pull/557) . The testing pipeline utilized was the osam-apigw-numa waterfront pipeline at pre-prod stage

### Test 1 - Regression Test

During this test, I verified that the functionality of the existing features was not impacted by my PR, by comparing metrics before and after applying my changes.

* Deployed the change at Friday noon.
* Monitored the continuity of metric propagations.
* No performance(memory usage/latency/throughput) downgrade.

### Test 2 - Performance Test

During this test, I moved the filtering UDF to source data transformer and compared metrics before and after the change. This is to verify saving one filtering vertex doesn't introduce any performance downgrade.

* Worked with Yashash and verified no performance downgrade.

### Test 3 - Watermark Assignment Test

The goal of this test is to verify after extracing message event time from payload, source vertex is able to use the new event time to publish watermark.

The approach I used to verify is that I created two pipeline with watermark maxDelay disabled. One pipeline is the original pipeline where we use Kafka event time, the other pipeline extract new event time from payload. With watermark maxDelay disabled, we started seeing late messages. I monitored the `no. of dropped messages` metric and observed that the event-time-from-payload pipeline dropped way more messages than original one, proving that the new event time was used to generate watermark.

**Notes**: The reason why extracting event time from payload introduces more late messages is because the event time from payload is at second level while the kafka event time is at millisecond level. With the event-time-from-payload pipeline, multiple messages are group together to share same event time and if within a group, one of the message is late, then the entire group are marked as late.

### Test 4 - Output comparison

During this test, I verified the output of the transformer pipeline is the same as the original pipeline, using Splunk. I spot-checked three [window, key] combinations and verified both pipelines are producing exact same data.
vigith: could you please resolve the conflicts? you can delete those lines, we moved those code to `serde.go`.

Patch:
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 {


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() {