Repository: numaproj/numaflow
Pull request #750 (2023-05-25 07:03:22)

fix: use int32 for message length
Fixes: https://github.com/numaproj/numaflow/issues/749

<!--

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.

-->



Patch:
diff --git a/pkg/isb/serde.go b/pkg/isb/serde.go
index fa85d016..9ecd6a26 100644
--- a/pkg/isb/serde.go
+++ b/pkg/isb/serde.go
@@ -55,7 +55,7 @@ func (p *MessageInfo) UnmarshalBinary(data []byte) (err error) {
 }
 
 type headerPreamble struct {
-	MLen    int16
+	MLen    int32
 	MsgKind MessageKind
 	IDLen   int16
 	KeyLen  int16
@@ -69,7 +69,7 @@ func (h Header) MarshalBinary() (data []byte, err error) {
 		return nil, err
 	}
 	var preamble = headerPreamble{
-		MLen:    int16(len(msgInfo)),
+		MLen:    int32(len(msgInfo)),
 		MsgKind: h.Kind,
 		IDLen:   int16(len(h.ID)),
 		KeyLen:  int16(len(h.Keys)),
@@ -145,14 +145,14 @@ func (h *Header) UnmarshalBinary(data []byte) (err error) {
 }
 
 type bodyPreamble struct {
-	PLen int64
+	PLen int32
 }
 
 // MarshalBinary encodes Body to a binary format
 func (b Body) MarshalBinary() (data []byte, err error) {
 	var buf = new(bytes.Buffer)
 	var preamble = bodyPreamble{
-		PLen: int64(len(b.Payload)),
+		PLen: int32(len(b.Payload)),
 	}
 	if err = binary.Write(buf, binary.LittleEndian, preamble); err != nil {
 		return nil, err
@@ -181,8 +181,8 @@ func (b *Body) UnmarshalBinary(data []byte) (err error) {
 }
 
 type messagePreamble struct {
-	HLen int16
-	BLen int16
+	HLen int32
+	BLen int32
 }
 
 // MarshalBinary encodes Message to the binary format
@@ -197,8 +197,8 @@ func (m Message) MarshalBinary() (data []byte, err error) {
 		return nil, err
 	}
 	var preamble = messagePreamble{
-		HLen: int16(len(header)),
-		BLen: int16(len(body)),
+		HLen: int32(len(header)),
+		BLen: int32(len(body)),
 	}
 	if err = binary.Write(buf, binary.LittleEndian, preamble); err != nil {
 		return nil, err
@@ -253,7 +253,7 @@ func (m *Message) UnmarshalBinary(data []byte) (err error) {
 }
 
 type readMessagePreamble struct {
-	MLen int16
+	MLen int32
 	// TODO: currently only support simple int offset
 	SimpleIntOffset int64
 	WMEpoch         int64
@@ -279,7 +279,7 @@ func (rm ReadMessage) MarshalBinary() (data []byte, err error) {
 	}
 
 	var preamble = readMessagePreamble{
-		MLen:            int16(len(message)),
+		MLen:            int32(len(message)),
 		SimpleIntOffset: offset,
 		WMEpoch:         rm.Watermark.UnixMilli(),
 		NumDelivered:    rm.Metadata.NumDelivered,


Test patch:
diff --git a/pkg/reduce/pbq/store/wal/segment_test.go b/pkg/reduce/pbq/store/wal/segment_test.go
index 73d90415..e358506d 100644
--- a/pkg/reduce/pbq/store/wal/segment_test.go
+++ b/pkg/reduce/pbq/store/wal/segment_test.go
@@ -261,7 +261,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	assert.NoError(t, err)
 	err = wal.Write(&message)
 	assert.NoError(t, err)
-	assert.Equal(t, int64(218), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(222), tempWAL.prevSyncedWOffset)
 
 	err = wal.Close()
 	assert.NoError(t, err)
@@ -318,7 +318,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	message := writeMessages[0]
 	storePrevSyncedTime := tempWAL.prevSyncedTime
 	err = wal.Write(&message)
-	assert.Equal(t, int64(128), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(130), tempWAL.prevSyncedWOffset)
 	assert.NotEqual(t, storePrevSyncedTime, tempWAL.prevSyncedTime)
 	assert.NoError(t, err)