Repository: numaproj/numaflow (go)

Issue:
remove Start-time and Endtime from PaneInfo and use Start-time for Event-time and WM
This will make the watermark progression conservative but will not delay the processing time.

Pull request description:
chore: removing start and end time from PaneInfo
- removing start and end time from pane info
- publishing window start time as the watermark
- setting window start time as the event time for the message

fixes #547 

Patch:
diff --git a/pkg/isb/message.go b/pkg/isb/message.go
index 764791dc..4a906288 100644
--- a/pkg/isb/message.go
+++ b/pkg/isb/message.go
@@ -24,8 +24,6 @@ import (
 // PaneInfo is the time window of the payload.
 type PaneInfo struct {
 	EventTime time.Time
-	StartTime time.Time
-	EndTime   time.Time
 	// IsLate is used to indicate if it's a late data .
 	IsLate bool
 }
diff --git a/pkg/isb/stores/jetstream/reader.go b/pkg/isb/stores/jetstream/reader.go
index c1d0bbf5..3cd6b579 100644
--- a/pkg/isb/stores/jetstream/reader.go
+++ b/pkg/isb/stores/jetstream/reader.go
@@ -284,14 +284,6 @@ func convert2IsbMsgHeader(header nats.Header) isb.Header {
 		i, _ := strconv.ParseInt(x, 10, 64)
 		r.EventTime = time.UnixMilli(i)
 	}
-	if x := header.Get(_startTime); x != "" {
-		i, _ := strconv.ParseInt(x, 10, 64)
-		r.StartTime = time.UnixMilli(i)
-	}
-	if x := header.Get(_endTime); x != "" {
-		i, _ := strconv.ParseInt(x, 10, 64)
-		r.EndTime = time.UnixMilli(i)
-	}
 	return r
 }
 
diff --git a/pkg/isb/stores/jetstream/writer.go b/pkg/isb/stores/jetstream/writer.go
index 992d593a..822f67ad 100644
--- a/pkg/isb/stores/jetstream/writer.go
+++ b/pkg/isb/stores/jetstream/writer.go
@@ -313,14 +313,12 @@ const (
 	_id        = "i"
 	_late      = "l"
 	_eventTime = "pev"
-	_startTime = "ps"
-	_endTime   = "pen"
 )
 
 func convert2NatsMsgHeader(header isb.Header) nats.Header {
 	r := nats.Header{}
 	r.Add(_id, header.ID)
-	r.Add(_key, string(header.Key))
+	r.Add(_key, header.Key)
 	if header.IsLate {
 		r.Add(_late, "1")
 	} else {
@@ -329,11 +327,5 @@ func convert2NatsMsgHeader(header isb.Header) nats.Header {
 	if !header.EventTime.IsZero() {
 		r.Add(_eventTime, fmt.Sprint(header.EventTime.UnixMilli()))
 	}
-	if !header.StartTime.IsZero() {
-		r.Add(_startTime, fmt.Sprint(header.StartTime.UnixMilli()))
-	}
-	if !header.EndTime.IsZero() {
-		r.Add(_endTime, fmt.Sprint(header.EndTime.UnixMilli()))
-	}
 	return r
 }
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 6115bbbd..ff164e04 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -107,7 +107,7 @@ func (p *ProcessAndForward) Forward(ctx context.Context) error {
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
 	}).Observe(float64(time.Since(startTime).Microseconds()))
 
-	processorWM := processor.Watermark(p.PartitionID.End)
+	processorWM := processor.Watermark(p.PartitionID.Start)
 
 	messagesToStep := p.whereToStep()
 
diff --git a/pkg/sinks/logger/log.go b/pkg/sinks/logger/log.go
index 2506fef4..4fdfb27b 100644
--- a/pkg/sinks/logger/log.go
+++ b/pkg/sinks/logger/log.go
@@ -97,7 +97,7 @@ func (t *ToLog) Write(_ context.Context, messages []isb.Message) ([]isb.Offset,
 	prefix := "(" + t.GetName() + ")"
 	for _, message := range messages {
 		logSinkWriteCount.With(map[string]string{metrics.LabelVertex: t.name, metrics.LabelPipeline: t.pipelineName}).Inc()
-		log.Println(prefix, " Payload - ", string(message.Payload), " Key - ", message.Key, " Start - ", message.StartTime.UnixMilli(), " End - ", message.EndTime.UnixMilli(), " EventTime - ", message.EventTime.UnixMilli())
+		log.Println(prefix, " Payload - ", string(message.Payload), " Key - ", message.Key, " EventTime - ", message.EventTime.UnixMilli())
 	}
 	return nil, make([]error, len(messages))
 }
diff --git a/pkg/udf/function/uds_grpc.go b/pkg/udf/function/uds_grpc.go
index 75727c4e..a3ed2cce 100644
--- a/pkg/udf/function/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -189,9 +189,7 @@ readLoop:
 		writeMessage := &isb.Message{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{
-					EventTime: partitionID.End,
-					StartTime: partitionID.Start,
-					EndTime:   partitionID.End,
+					EventTime: partitionID.Start,
 					IsLate:    false,
 				},
 				Key: key,


Test patch:
diff --git a/pkg/isb/stores/jetstream/writer_test.go b/pkg/isb/stores/jetstream/writer_test.go
index 61cbb2b0..418ef810 100644
--- a/pkg/isb/stores/jetstream/writer_test.go
+++ b/pkg/isb/stores/jetstream/writer_test.go
@@ -269,8 +269,6 @@ func TestConvert2NatsMsgHeader(t *testing.T) {
 	isbHeader := isb.Header{
 		PaneInfo: isb.PaneInfo{
 			EventTime: time.Unix(1636470000, 0),
-			StartTime: time.Unix(1636470000, 0),
-			EndTime:   time.Unix(1636470060, 0),
 		},
 		ID: "1",
 	}
diff --git a/pkg/isb/testutils/rw.go b/pkg/isb/testutils/rw.go
index 3b10fe54..cd15e7d5 100644
--- a/pkg/isb/testutils/rw.go
+++ b/pkg/isb/testutils/rw.go
@@ -74,7 +74,7 @@ func BuildTestReadMessages(count int64, startTime time.Time) []isb.ReadMessage {
 	return readMessages
 }
 
-// BuildTestReadMessages builds test isb.ReadMessage which can be used for testing.
+// BuildTestReadMessagesIntOffset builds test isb.ReadMessage which can be used for testing.
 func BuildTestReadMessagesIntOffset(count int64, startTime time.Time) []isb.ReadMessage {
 	writeMessages := BuildTestWriteMessages(count, startTime)
 	var readMessages = make([]isb.ReadMessage, count)
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index 10b4bb6d..e5de4a92 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -120,8 +120,6 @@ func (f CounterReduceTest) ApplyReduce(_ context.Context, partitionID *partition
 	ret := &isb.Message{
 		Header: isb.Header{
 			PaneInfo: isb.PaneInfo{
-				StartTime: partitionID.Start,
-				EndTime:   partitionID.End,
 				EventTime: partitionID.End,
 			},
 			ID:  "msgID",
@@ -159,8 +157,6 @@ func (s SumReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		msg := &isb.Message{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{
-					StartTime: partitionID.Start,
-					EndTime:   partitionID.End,
 					EventTime: partitionID.End,
 				},
 				ID:  "msgID",
@@ -199,8 +195,6 @@ func (m MaxReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		ret := &isb.Message{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{
-					StartTime: partitionID.Start,
-					EndTime:   partitionID.End,
 					EventTime: partitionID.End,
 				},
 				ID:  "msgID",
diff --git a/pkg/reduce/pbq/store/wal/segment_test.go b/pkg/reduce/pbq/store/wal/segment_test.go
index 2c7c386f..71178114 100644
--- a/pkg/reduce/pbq/store/wal/segment_test.go
+++ b/pkg/reduce/pbq/store/wal/segment_test.go
@@ -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, tempWAL.prevSyncedWOffset, int64(844))
+	assert.Equal(t, tempWAL.prevSyncedWOffset, int64(650))
 
 	err = wal.Close()
 	assert.NoError(t, err)
@@ -316,7 +316,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	message := writeMessages[0]
 	storePrevSyncedTime := tempWAL.prevSyncedTime
 	err = wal.Write(&message)
-	assert.Equal(t, tempWAL.prevSyncedWOffset, int64(441))
+	assert.Equal(t, tempWAL.prevSyncedWOffset, int64(344))
 	assert.NotEqual(t, storePrevSyncedTime, tempWAL.prevSyncedTime)
 	assert.NoError(t, err)
 
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index f2246d86..0c0fbce4 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -212,7 +212,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			wmExpected: map[string]ot.Value{
 				"buffer1": {
 					Offset:    0,
-					Watermark: 120000,
+					Watermark: 60000,
 					Idle:      false,
 				},
 				"buffer2": {
@@ -236,12 +236,12 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			wmExpected: map[string]ot.Value{
 				"buffer1": {
 					Offset:    0,
-					Watermark: 120000,
+					Watermark: 60000,
 					Idle:      false,
 				},
 				"buffer2": {
 					Offset:    0,
-					Watermark: 120000,
+					Watermark: 60000,
 					Idle:      false,
 				},
 			},
diff --git a/pkg/reduce/readloop/readloop_test.go b/pkg/reduce/readloop/readloop_test.go
index 67fc2361..9966bf9c 100644
--- a/pkg/reduce/readloop/readloop_test.go
+++ b/pkg/reduce/readloop/readloop_test.go
@@ -65,8 +65,6 @@ func (s SumReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		msg := &isb.Message{
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{
-					StartTime: partitionID.Start,
-					EndTime:   partitionID.End,
 					EventTime: partitionID.End,
 				},
 				ID:  "msgID",
@@ -152,8 +150,6 @@ func TestReadLoop_Startup(t *testing.T) {
 			Header: isb.Header{
 				PaneInfo: isb.PaneInfo{
 					EventTime: time.Unix(300, 0),
-					StartTime: time.Time{},
-					EndTime:   time.Time{},
 					IsLate:    false,
 				},
 				ID:  "",
diff --git a/test/reduce-e2e/reduce_test.go b/test/reduce-e2e/reduce_test.go
index 59cc119a..36bddd26 100644
--- a/test/reduce-e2e/reduce_test.go
+++ b/test/reduce-e2e/reduce_test.go
@@ -229,8 +229,8 @@ func (r *ReduceSuite) TestComplexSlidingWindowPipeline() {
 	}()
 
 	w.Expect().
-		SinkContains("sink", "15").
-		SinkContains("sink", "45").
+		SinkContains("sink", "30").
+		SinkContains("sink", "60").
 		SinkContains("sink", "180")
 	done <- struct{}{}
 }