Repository: numaproj/numaflow
Pull request #582 (2023-03-02 17:03:39)

chore: refactor PaneInfo and add MessageType
Repurpose PaneInfo for other use cases.


Patch:
diff --git a/pkg/isb/message.go b/pkg/isb/message.go
index b6e59350..5f495e55 100644
--- a/pkg/isb/message.go
+++ b/pkg/isb/message.go
@@ -20,16 +20,43 @@ import (
 	"time"
 )
 
-// PaneInfo is the time window of the payload.
-type PaneInfo struct {
+// MessageKind represents the message type of the payload.
+type MessageKind int16
+
+const (
+	Data MessageKind = 1 << iota // Data payload
+	WMB                          // Watermark Barrier
+)
+
+func (mt MessageKind) String() string {
+	switch mt {
+	case mt & Data:
+		return "Data"
+	case mt & WMB:
+		return "WMB"
+	default:
+		return "Unknown"
+	}
+}
+
+// MessageInfo is the message information window of the payload.
+// The contents inside the MessageInfo can be interpreted differently based on the MessageKind.
+type MessageInfo struct {
+	// EventTime when
+	// MessageKind == Data represents the event time of the message
+	// MessageKind == WMB, value is ignored
 	EventTime time.Time
-	// IsLate is used to indicate if it's a late data .
+	// IsLate when
+	// MessageKind == Data, IsLate is used to indicate if the message is a late data (assignment happens at source)
+	// MessageKind == WMB, value is ignored
 	IsLate bool
 }
 
 // Header is the header of the message
 type Header struct {
-	PaneInfo
+	MessageInfo
+	// Kind indicates the kind of Message
+	Kind MessageKind
 	// ID is used for exactly-once-semantics. ID is usually populated from the offset, if offset is available.
 	ID string
 	// Key is (key,value) in the map-reduce paradigm which will be used for conditional forwarding.
diff --git a/pkg/isb/serde.go b/pkg/isb/serde.go
index 33d99bc2..2255b9a2 100644
--- a/pkg/isb/serde.go
+++ b/pkg/isb/serde.go
@@ -25,15 +25,15 @@ import (
 	"time"
 )
 
-type paneInfoPreamble struct {
+type messageInfoPreamble struct {
 	EventEpoch int64
 	IsLate     bool
 }
 
-// MarshalBinary encodes PaneInfo to the binary format
-func (p PaneInfo) MarshalBinary() (data []byte, err error) {
+// MarshalBinary encodes MessageInfo to the binary format
+func (p MessageInfo) MarshalBinary() (data []byte, err error) {
 	var buf = new(bytes.Buffer)
-	var preamble = paneInfoPreamble{
+	var preamble = messageInfoPreamble{
 		EventEpoch: p.EventTime.UnixMilli(),
 		IsLate:     p.IsLate,
 	}
@@ -43,10 +43,10 @@ func (p PaneInfo) MarshalBinary() (data []byte, err error) {
 	return buf.Bytes(), nil
 }
 
-// UnmarshalBinary decodes PaneInfo from the binary format
-func (p *PaneInfo) UnmarshalBinary(data []byte) (err error) {
+// UnmarshalBinary decodes MessageInfo from the binary format
+func (p *MessageInfo) UnmarshalBinary(data []byte) (err error) {
 	var r = bytes.NewReader(data)
-	var preamble = new(paneInfoPreamble)
+	var preamble = new(messageInfoPreamble)
 	err = binary.Read(r, binary.LittleEndian, preamble)
 	if err != nil {
 		return err
@@ -57,31 +57,33 @@ func (p *PaneInfo) UnmarshalBinary(data []byte) (err error) {
 }
 
 type headerPreamble struct {
-	PLen   int16
-	IDLen  int16
-	KeyLen int16
+	MLen    int16
+	MsgKind MessageKind
+	IDLen   int16
+	KeyLen  int16
 }
 
 // MarshalBinary encodes Header to a binary format
 func (h Header) MarshalBinary() (data []byte, err error) {
 	var buf = new(bytes.Buffer)
-	paneInfo, err := h.PaneInfo.MarshalBinary()
+	msgInfo, err := h.MessageInfo.MarshalBinary()
 	if err != nil {
 		return nil, err
 	}
 	var preamble = headerPreamble{
-		PLen:   int16(len(paneInfo)),
-		IDLen:  int16(len(h.ID)),
-		KeyLen: int16(len(h.Key)),
+		MLen:    int16(len(msgInfo)),
+		MsgKind: h.Kind,
+		IDLen:   int16(len(h.ID)),
+		KeyLen:  int16(len(h.Key)),
 	}
 	if err = binary.Write(buf, binary.LittleEndian, preamble); err != nil {
 		return nil, err
 	}
-	n, err := buf.Write(paneInfo)
+	n, err := buf.Write(msgInfo)
 	if err != nil {
 		return nil, err
-	} else if n != int(preamble.PLen) {
-		return nil, fmt.Errorf("expected to write paneInfo size of %d but got %d", preamble.PLen, n)
+	} else if n != int(preamble.MLen) {
+		return nil, fmt.Errorf("expected to write msgInfo size of %d but got %d", preamble.MLen, n)
 	}
 	if err = binary.Write(buf, binary.LittleEndian, []byte(h.ID)); err != nil {
 		return nil, err
@@ -99,15 +101,15 @@ func (h *Header) UnmarshalBinary(data []byte) (err error) {
 	if err = binary.Read(r, binary.LittleEndian, preamble); err != nil {
 		return err
 	}
-	var paneInfoByte = make([]byte, preamble.PLen)
-	n, err := r.Read(paneInfoByte)
+	var msgInfoByte = make([]byte, preamble.MLen)
+	n, err := r.Read(msgInfoByte)
 	if err != nil {
 		return err
-	} else if n != int(preamble.PLen) {
-		return fmt.Errorf("expected to read paneInfo size of %d but got %d", preamble.PLen, n)
+	} else if n != int(preamble.MLen) {
+		return fmt.Errorf("expected to read msgInfo size of %d but got %d", preamble.MLen, n)
 	}
-	var paneInfo = new(PaneInfo)
-	if err = paneInfo.UnmarshalBinary(paneInfoByte); err != nil {
+	var msgInfo = new(MessageInfo)
+	if err = msgInfo.UnmarshalBinary(msgInfoByte); err != nil {
 		return err
 	}
 	var id = make([]byte, preamble.IDLen)
@@ -118,7 +120,8 @@ func (h *Header) UnmarshalBinary(data []byte) (err error) {
 	if err = binary.Read(r, binary.LittleEndian, key); err != nil {
 		return err
 	}
-	h.PaneInfo = *paneInfo
+	h.MessageInfo = *msgInfo
+	h.Kind = preamble.MsgKind
 	h.ID = string(id)
 	h.Key = string(key)
 	return err
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 40441198..bf15c389 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -318,8 +318,8 @@ func (mg *memgen) newReadMessage(payload []byte, offset int64) *isb.ReadMessage
 	msg := isb.Message{
 		Header: isb.Header{
 			// TODO: insert the right time based on the generator
-			PaneInfo: isb.PaneInfo{EventTime: timeFromNanos(parseTime(payload))},
-			ID:       strconv.FormatInt(offset, 10) + "-" + strconv.FormatInt(int64(mg.vertexInstance.Replica), 10),
+			MessageInfo: isb.MessageInfo{EventTime: timeFromNanos(parseTime(payload))},
+			ID:          strconv.FormatInt(offset, 10) + "-" + strconv.FormatInt(int64(mg.vertexInstance.Replica), 10),
 		},
 		Body: isb.Body{Payload: payload},
 	}
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index a2b36d4d..4298f80c 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -159,8 +159,8 @@ func New(
 		m := &isb.ReadMessage{
 			Message: isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{EventTime: eventTime},
-					ID:       id,
+					MessageInfo: isb.MessageInfo{EventTime: eventTime},
+					ID:          id,
 				},
 				Body: isb.Body{
 					Payload: msg,
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index a291ec03..fb746bb6 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -392,9 +392,9 @@ func toReadMessage(m *sarama.ConsumerMessage) *isb.ReadMessage {
 	offset := toOffset(m.Topic, m.Partition, m.Offset)
 	msg := isb.Message{
 		Header: isb.Header{
-			PaneInfo: isb.PaneInfo{EventTime: m.Timestamp},
-			ID:       offset,
-			Key:      string(m.Key),
+			MessageInfo: isb.MessageInfo{EventTime: m.Timestamp},
+			ID:          offset,
+			Key:         string(m.Key),
 		},
 		Body: isb.Body{Payload: m.Value},
 	}
diff --git a/pkg/sources/nats/nats.go b/pkg/sources/nats/nats.go
index 7fd0ff84..b83ed694 100644
--- a/pkg/sources/nats/nats.go
+++ b/pkg/sources/nats/nats.go
@@ -167,8 +167,8 @@ func New(
 			Message: isb.Message{
 				Header: isb.Header{
 					// TODO: Be able to specify event time.
-					PaneInfo: isb.PaneInfo{EventTime: time.Now()},
-					ID:       id,
+					MessageInfo: isb.MessageInfo{EventTime: time.Now()},
+					ID:          id,
 				},
 				Body: isb.Body{
 					Payload: msg.Data,
diff --git a/pkg/sources/transformer/grpc_transformer.go b/pkg/sources/transformer/grpc_transformer.go
index a357e108..32606afd 100644
--- a/pkg/sources/transformer/grpc_transformer.go
+++ b/pkg/sources/transformer/grpc_transformer.go
@@ -84,11 +84,11 @@ func (u *gRPCBasedTransformer) ApplyMap(ctx context.Context, readMessage *isb.Re
 	key := readMessage.Key
 	payload := readMessage.Body.Payload
 	offset := readMessage.ReadOffset
-	parentPaneInfo := readMessage.PaneInfo
+	parentMessageInfo := readMessage.MessageInfo
 	var d = &functionpb.Datum{
 		Key:       key,
 		Value:     payload,
-		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentPaneInfo.EventTime)},
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentMessageInfo.EventTime)},
 		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(readMessage.Watermark)},
 	}
 
@@ -110,13 +110,13 @@ func (u *gRPCBasedTransformer) ApplyMap(ctx context.Context, readMessage *isb.Re
 		key := datum.Key
 		if datum.EventTime != nil {
 			// Transformer supports changing event time.
-			parentPaneInfo.EventTime = datum.EventTime.EventTime.AsTime()
+			parentMessageInfo.EventTime = datum.EventTime.EventTime.AsTime()
 		}
 		writeMessage := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: parentPaneInfo,
-				ID:       fmt.Sprintf("%s-%d", offset.String(), i),
-				Key:      key,
+				MessageInfo: parentMessageInfo,
+				ID:          fmt.Sprintf("%s-%d", offset.String(), i),
+				Key:         key,
 			},
 			Body: isb.Body{
 				Payload: datum.Value,
diff --git a/pkg/udf/function/uds_grpc.go b/pkg/udf/function/uds_grpc.go
index a3ed2cce..4b2020a8 100644
--- a/pkg/udf/function/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -82,11 +82,11 @@ func (u *UDSgRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 	key := readMessage.Key
 	payload := readMessage.Body.Payload
 	offset := readMessage.ReadOffset
-	parentPaneInfo := readMessage.PaneInfo
+	parentMessageInfo := readMessage.MessageInfo
 	var d = &functionpb.Datum{
 		Key:       key,
 		Value:     payload,
-		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentPaneInfo.EventTime)},
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentMessageInfo.EventTime)},
 		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(readMessage.Watermark)},
 	}
 
@@ -108,9 +108,9 @@ func (u *UDSgRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 		key := datum.Key
 		writeMessage := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: parentPaneInfo,
-				ID:       fmt.Sprintf("%s-%d", offset.String(), i),
-				Key:      key,
+				MessageInfo: parentMessageInfo,
+				ID:          fmt.Sprintf("%s-%d", offset.String(), i),
+				Key:         key,
 			},
 			Body: isb.Body{
 				Payload: datum.Value,
@@ -188,7 +188,7 @@ readLoop:
 		key := datum.Key
 		writeMessage := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: partitionID.Start,
 					IsLate:    false,
 				},
@@ -206,12 +206,12 @@ readLoop:
 func createDatum(readMessage *isb.ReadMessage) *functionpb.Datum {
 	key := readMessage.Key
 	payload := readMessage.Body.Payload
-	parentPaneInfo := readMessage.PaneInfo
+	parentMessageInfo := readMessage.MessageInfo
 
 	var d = &functionpb.Datum{
 		Key:       key,
 		Value:     payload,
-		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentPaneInfo.EventTime)},
+		EventTime: &functionpb.EventTime{EventTime: timestamppb.New(parentMessageInfo.EventTime)},
 		Watermark: &functionpb.Watermark{Watermark: timestamppb.New(readMessage.Watermark)},
 	}
 	return d


Test patch:
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index 3807408f..ec8a924f 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -152,7 +152,7 @@ func (f mySourceForwardTest) ApplyMap(ctx context.Context, message *isb.ReadMess
 		_ = ctx
 		offset := readMessage.ReadOffset
 		payload := readMessage.Body.Payload
-		parentPaneInfo := readMessage.PaneInfo
+		parentPaneInfo := readMessage.MessageInfo
 
 		// apply source data transformer
 		_ = payload
@@ -164,9 +164,9 @@ func (f mySourceForwardTest) ApplyMap(ctx context.Context, message *isb.ReadMess
 
 		writeMessage := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: parentPaneInfo,
-				ID:       offset.String(),
-				Key:      key,
+				MessageInfo: parentPaneInfo,
+				ID:          offset.String(),
+				Key:         key,
 			},
 			Body: isb.Body{
 				Payload: result,
diff --git a/pkg/isb/serde_test.go b/pkg/isb/serde_test.go
index 7dc7e7a4..d26f26f5 100644
--- a/pkg/isb/serde_test.go
+++ b/pkg/isb/serde_test.go
@@ -16,7 +16,7 @@ func TestPaneInfo(t *testing.T) {
 	tests := []struct {
 		name               string
 		fields             fields
-		wantData           PaneInfo
+		wantData           MessageInfo
 		wantMarshalError   bool
 		wantUnmarshalError bool
 	}{
@@ -26,7 +26,7 @@ func TestPaneInfo(t *testing.T) {
 				EventTime: time.UnixMilli(1676617200000),
 				IsLate:    false,
 			},
-			wantData: PaneInfo{
+			wantData: MessageInfo{
 				EventTime: time.UnixMilli(1676617200000).UTC(),
 				IsLate:    false,
 			},
@@ -39,7 +39,7 @@ func TestPaneInfo(t *testing.T) {
 				EventTime: time.UnixMilli(1676617200000),
 				IsLate:    true,
 			},
-			wantData: PaneInfo{
+			wantData: MessageInfo{
 				EventTime: time.UnixMilli(1676617200000).UTC(),
 				IsLate:    true,
 			},
@@ -49,7 +49,7 @@ func TestPaneInfo(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			p := PaneInfo{
+			p := MessageInfo{
 				EventTime: tt.fields.EventTime,
 				IsLate:    tt.fields.IsLate,
 			}
@@ -58,7 +58,7 @@ func TestPaneInfo(t *testing.T) {
 				t.Errorf("MarshalBinary() error = %v, wantMarshalError %v", err, tt.wantMarshalError)
 				return
 			}
-			var newP = new(PaneInfo)
+			var newP = new(MessageInfo)
 			err = newP.UnmarshalBinary(gotData)
 			if (err != nil) != tt.wantUnmarshalError {
 				t.Errorf("UnmarshalBinary() error = %v, wantUnmarshalError %v", err, tt.wantMarshalError)
@@ -73,9 +73,10 @@ func TestPaneInfo(t *testing.T) {
 
 func TestHeader(t *testing.T) {
 	type fields struct {
-		PaneInfo PaneInfo
-		ID       string
-		Key      string
+		MessageInfo MessageInfo
+		Kind        MessageKind
+		ID          string
+		Key         string
 	}
 	tests := []struct {
 		name               string
@@ -87,20 +88,22 @@ func TestHeader(t *testing.T) {
 		{
 			name: "good",
 			fields: fields{
-				PaneInfo: PaneInfo{
+				MessageInfo: MessageInfo{
 					EventTime: time.UnixMilli(1676617200000),
 					IsLate:    true,
 				},
-				ID:  "TestID",
-				Key: "TestKey",
+				Kind: Data,
+				ID:   "TestID",
+				Key:  "TestKey",
 			},
 			wantData: Header{
-				PaneInfo: PaneInfo{
+				MessageInfo: MessageInfo{
 					EventTime: time.UnixMilli(1676617200000).UTC(),
 					IsLate:    true,
 				},
-				ID:  "TestID",
-				Key: "TestKey",
+				Kind: Data,
+				ID:   "TestID",
+				Key:  "TestKey",
 			},
 			wantMarshalError:   false,
 			wantUnmarshalError: false,
@@ -109,9 +112,10 @@ func TestHeader(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			h := &Header{
-				PaneInfo: tt.fields.PaneInfo,
-				ID:       tt.fields.ID,
-				Key:      tt.fields.Key,
+				MessageInfo: tt.fields.MessageInfo,
+				Kind:        tt.fields.Kind,
+				ID:          tt.fields.ID,
+				Key:         tt.fields.Key,
 			}
 			gotData, err := h.MarshalBinary()
 			if (err != nil) != tt.wantMarshalError {
@@ -202,12 +206,13 @@ func TestMessage(t *testing.T) {
 			name: "good",
 			fields: fields{
 				Header: Header{
-					PaneInfo: PaneInfo{
+					MessageInfo: MessageInfo{
 						EventTime: time.UnixMilli(1676617200000),
 						IsLate:    true,
 					},
-					ID:  "TestID",
-					Key: "TestKey",
+					Kind: Data,
+					ID:   "TestID",
+					Key:  "TestKey",
 				},
 				Body: Body{
 					Payload: []byte("TestBODY"),
@@ -215,12 +220,13 @@ func TestMessage(t *testing.T) {
 			},
 			wantData: Message{
 				Header: Header{
-					PaneInfo: PaneInfo{
+					MessageInfo: MessageInfo{
 						EventTime: time.UnixMilli(1676617200000).UTC(),
 						IsLate:    true,
 					},
-					ID:  "TestID",
-					Key: "TestKey",
+					Kind: Data,
+					ID:   "TestID",
+					Key:  "TestKey",
 				},
 				Body: Body{
 					Payload: []byte("TestBODY"),
@@ -279,12 +285,13 @@ func TestReadMessage(t *testing.T) {
 			fields: fields{
 				Message: Message{
 					Header: Header{
-						PaneInfo: PaneInfo{
+						MessageInfo: MessageInfo{
 							EventTime: time.UnixMilli(1676617200000),
 							IsLate:    true,
 						},
-						ID:  "TestID",
-						Key: "TestKey",
+						Kind: Data,
+						ID:   "TestID",
+						Key:  "TestKey",
 					},
 					Body: Body{
 						Payload: []byte("TestBODY"),
@@ -298,12 +305,13 @@ func TestReadMessage(t *testing.T) {
 			wantData: ReadMessage{
 				Message: Message{
 					Header: Header{
-						PaneInfo: PaneInfo{
+						MessageInfo: MessageInfo{
 							EventTime: time.UnixMilli(1676617200000).UTC(),
 							IsLate:    true,
 						},
-						ID:  "TestID",
-						Key: "TestKey",
+						Kind: Data,
+						ID:   "TestID",
+						Key:  "TestKey",
 					},
 					Body: Body{
 						Payload: []byte("TestBODY"),
diff --git a/pkg/isb/stores/jetstream/writer_test.go b/pkg/isb/stores/jetstream/writer_test.go
index 418ef810..066c7520 100644
--- a/pkg/isb/stores/jetstream/writer_test.go
+++ b/pkg/isb/stores/jetstream/writer_test.go
@@ -267,7 +267,7 @@ func TestWriteClose(t *testing.T) {
 // TestConvert2NatsMsgHeader is used to convert nats header
 func TestConvert2NatsMsgHeader(t *testing.T) {
 	isbHeader := isb.Header{
-		PaneInfo: isb.PaneInfo{
+		MessageInfo: isb.MessageInfo{
 			EventTime: time.Unix(1636470000, 0),
 		},
 		ID: "1",
diff --git a/pkg/isb/stores/redis/write_test.go b/pkg/isb/stores/redis/write_test.go
index c9bb1a4c..6b1bcbfc 100644
--- a/pkg/isb/stores/redis/write_test.go
+++ b/pkg/isb/stores/redis/write_test.go
@@ -197,7 +197,7 @@ func buildTestWriteMessages(rqw *BufferWrite, count int64, startTime time.Time)
 func TestLua(t *testing.T) {
 	ctx := context.Background()
 	client := redis.NewUniversalClient(redisOptions)
-	message := isb.Message{Header: isb.Header{ID: "0", PaneInfo: isb.PaneInfo{EventTime: testStartTime}}, Body: isb.Body{Payload: []byte("foo")}}
+	message := isb.Message{Header: isb.Header{ID: "0", MessageInfo: isb.MessageInfo{EventTime: testStartTime}}, Body: isb.Body{Payload: []byte("foo")}}
 	script := redis.NewScript(exactlyOnceInsertLuaScript)
 
 	var hashName = "{step-1}:1234567890:hash-foo"
@@ -565,7 +565,7 @@ func forwardDataAndVerify(ctx context.Context, t *testing.T, fromStepWrite *Buff
 
 	// Read messages and validate the content
 	assert.Len(t, readMessages, 17)
-	assert.Equal(t, []interface{}{writeMessages[0].Header.PaneInfo, writeMessages[1].Header.PaneInfo, writeMessages[2].Header.PaneInfo, writeMessages[3].Header.PaneInfo, writeMessages[4].Header.PaneInfo}, []interface{}{readMessages[0].Header.PaneInfo, readMessages[1].Header.PaneInfo, readMessages[2].Header.PaneInfo, readMessages[3].Header.PaneInfo, readMessages[4].Header.PaneInfo})
+	assert.Equal(t, []interface{}{writeMessages[0].Header.MessageInfo, writeMessages[1].Header.MessageInfo, writeMessages[2].Header.MessageInfo, writeMessages[3].Header.MessageInfo, writeMessages[4].Header.MessageInfo}, []interface{}{readMessages[0].Header.MessageInfo, readMessages[1].Header.MessageInfo, readMessages[2].Header.MessageInfo, readMessages[3].Header.MessageInfo, readMessages[4].Header.MessageInfo})
 	assert.Equal(t, []interface{}{writeMessages[0].Body.Payload, writeMessages[1].Body.Payload, writeMessages[2].Body.Payload, writeMessages[3].Body.Payload, writeMessages[4].Body.Payload}, []interface{}{readMessages[0].Body.Payload, readMessages[1].Body.Payload, readMessages[2].Body.Payload, readMessages[3].Body.Payload, readMessages[4].Body.Payload})
 
 	_, _ = fromStepWrite.Write(ctx, writeMessages[18:20])
diff --git a/pkg/isb/testutils/rw.go b/pkg/isb/testutils/rw.go
index 2eb02ded..fb63e6bd 100644
--- a/pkg/isb/testutils/rw.go
+++ b/pkg/isb/testutils/rw.go
@@ -47,7 +47,7 @@ func BuildTestWriteMessages(count int64, startTime time.Time) []isb.Message {
 		messages = append(messages,
 			isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
+					MessageInfo: isb.MessageInfo{
 						EventTime: tmpTime,
 					},
 					ID: fmt.Sprintf("%d", i),
diff --git a/pkg/isb/testutils/udf.go b/pkg/isb/testutils/udf.go
index 307e56cf..d6b4c056 100644
--- a/pkg/isb/testutils/udf.go
+++ b/pkg/isb/testutils/udf.go
@@ -27,7 +27,7 @@ func CopyUDFTestApply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb
 	_ = ctx
 	offset := readMessage.ReadOffset
 	payload := readMessage.Body.Payload
-	parentPaneInfo := readMessage.PaneInfo
+	parentPaneInfo := readMessage.MessageInfo
 
 	// apply UDF
 	_ = payload
@@ -37,9 +37,9 @@ func CopyUDFTestApply(ctx context.Context, readMessage *isb.ReadMessage) ([]*isb
 
 	writeMessage := &isb.Message{
 		Header: isb.Header{
-			PaneInfo: parentPaneInfo,
-			ID:       offset.String(),
-			Key:      key,
+			MessageInfo: parentPaneInfo,
+			ID:          offset.String(),
+			Key:         key,
 		},
 		Body: isb.Body{
 			Payload: result,
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index fec028c1..124b2cd9 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -119,7 +119,7 @@ func (f CounterReduceTest) ApplyReduce(_ context.Context, partitionID *partition
 	b, _ := json.Marshal(payload)
 	ret := &isb.Message{
 		Header: isb.Header{
-			PaneInfo: isb.PaneInfo{
+			MessageInfo: isb.MessageInfo{
 				EventTime: partitionID.End,
 			},
 			ID:  "msgID",
@@ -156,7 +156,7 @@ func (s SumReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		b, _ := json.Marshal(payload)
 		msg := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: partitionID.End,
 				},
 				ID:  "msgID",
@@ -194,7 +194,7 @@ func (m MaxReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		b, _ := json.Marshal(payload)
 		ret := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: partitionID.End,
 				},
 				ID:  "msgID",
@@ -806,7 +806,7 @@ func buildMessagesForReduce(count int, key string, publishTime time.Time) []isb.
 		})
 		messages[i] = isb.Message{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: publishTime,
 				},
 				ID:  fmt.Sprintf("%d", i),
@@ -894,7 +894,7 @@ func buildIsbMessage(messageValue int, eventTime time.Time) isb.Message {
 	})
 	return isb.Message{
 		Header: isb.Header{
-			PaneInfo: isb.PaneInfo{
+			MessageInfo: isb.MessageInfo{
 				EventTime: eventTime,
 			},
 			ID:  fmt.Sprintf("%d", messageValue),
diff --git a/pkg/reduce/pbq/store/wal/segment_test.go b/pkg/reduce/pbq/store/wal/segment_test.go
index 476aaa25..1d23c8b4 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, int64(408), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(418), tempWAL.prevSyncedWOffset)
 
 	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, int64(223), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(228), tempWAL.prevSyncedWOffset)
 	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 8180e88e..8a9a3ed2 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -314,7 +314,7 @@ func createProcessAndForwardAndOTStore(ctx context.Context, key string, pbqManag
 	var result = []*isb.Message{
 		{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: time.UnixMilli(60000),
 				},
 				ID:  "1",
diff --git a/pkg/reduce/readloop/readloop_test.go b/pkg/reduce/readloop/readloop_test.go
index 9966bf9c..080d3552 100644
--- a/pkg/reduce/readloop/readloop_test.go
+++ b/pkg/reduce/readloop/readloop_test.go
@@ -64,7 +64,7 @@ func (s SumReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		b, _ := json.Marshal(payload)
 		msg := &isb.Message{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: partitionID.End,
 				},
 				ID:  "msgID",
@@ -148,7 +148,7 @@ func TestReadLoop_Startup(t *testing.T) {
 	latestMessage := &isb.ReadMessage{
 		Message: isb.Message{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{
+				MessageInfo: isb.MessageInfo{
 					EventTime: time.Unix(300, 0),
 					IsLate:    false,
 				},
@@ -204,7 +204,7 @@ func createStoreMessages(_ context.Context, key string, value int, eventTime tim
 		readMessage := &isb.ReadMessage{
 			Message: isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
+					MessageInfo: isb.MessageInfo{
 						EventTime: eventTime,
 					},
 					ID:  fmt.Sprintf("%d", value+1),
diff --git a/pkg/sinks/kafka/kafka_test.go b/pkg/sinks/kafka/kafka_test.go
index 891c74e2..c92890bf 100644
--- a/pkg/sinks/kafka/kafka_test.go
+++ b/pkg/sinks/kafka/kafka_test.go
@@ -65,15 +65,15 @@ func TestWriteSuccessToKafka(t *testing.T) {
 	msgs := []isb.Message{
 		{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{},
-				Key:      "key1",
+				MessageInfo: isb.MessageInfo{},
+				Key:         "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
 		{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{},
-				Key:      "key1",
+				MessageInfo: isb.MessageInfo{},
+				Key:         "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
@@ -117,15 +117,15 @@ func TestWriteFailureToKafka(t *testing.T) {
 	msgs := []isb.Message{
 		{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{},
-				Key:      "key1",
+				MessageInfo: isb.MessageInfo{},
+				Key:         "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
 		{
 			Header: isb.Header{
-				PaneInfo: isb.PaneInfo{},
-				Key:      "key1",
+				MessageInfo: isb.MessageInfo{},
+				Key:         "key1",
 			},
 			Body: isb.Body{Payload: []byte("welcome1")},
 		},
diff --git a/pkg/sources/transformer/grpc_transformer_test.go b/pkg/sources/transformer/grpc_transformer_test.go
index e1a2f952..eb92b55d 100644
--- a/pkg/sources/transformer/grpc_transformer_test.go
+++ b/pkg/sources/transformer/grpc_transformer_test.go
@@ -27,10 +27,11 @@ import (
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
 	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
 	"github.com/numaproj/numaflow-go/pkg/function/clienttest"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/testutils"
 	"github.com/numaproj/numaflow/pkg/udf/function"
-	"github.com/stretchr/testify/assert"
 
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
@@ -114,7 +115,7 @@ func TestGRPCBasedTransformer_BasicApplyWithMockClient(t *testing.T) {
 		got, err := u.ApplyMap(ctx, &isb.ReadMessage{
 			Message: isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
+					MessageInfo: isb.MessageInfo{
 						EventTime: time.Unix(1661169600, 0),
 					},
 					ID:  "test_id",
@@ -158,7 +159,7 @@ func TestGRPCBasedTransformer_BasicApplyWithMockClient(t *testing.T) {
 		_, err := u.ApplyMap(ctx, &isb.ReadMessage{
 			Message: isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
+					MessageInfo: isb.MessageInfo{
 						EventTime: time.Unix(1661169660, 0),
 					},
 					ID:  "test_id",
diff --git a/pkg/udf/function/uds_grpc_test.go b/pkg/udf/function/uds_grpc_test.go
index a286f125..05bff5f6 100644
--- a/pkg/udf/function/uds_grpc_test.go
+++ b/pkg/udf/function/uds_grpc_test.go
@@ -116,7 +116,7 @@ func TestGRPCBasedUDF_BasicApplyWithMockClient(t *testing.T) {
 		got, err := u.ApplyMap(ctx, &isb.ReadMessage{
 			Message: isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
+					MessageInfo: isb.MessageInfo{
 						EventTime: time.Unix(1661169600, 0),
 					},
 					ID:  "test_id",
@@ -160,7 +160,7 @@ func TestGRPCBasedUDF_BasicApplyWithMockClient(t *testing.T) {
 		_, err := u.ApplyMap(ctx, &isb.ReadMessage{
 			Message: isb.Message{
 				Header: isb.Header{
-					PaneInfo: isb.PaneInfo{
+					MessageInfo: isb.MessageInfo{
 						EventTime: time.Unix(1661169660, 0),
 					},
 					ID:  "test_id",