Repository: numaproj/numaflow
Pull request #250 (2022-10-21 15:29:13)

chore: introduce simple int offset
Signed-off-by: Vigith Maurice <vigith@gmail.com>

Offset is a monotonically increasing integer. For most of the internal representations, int64 makes more sense rather than the string representation. We can avoid an unwanted conversion if we have the support of int64. 

This will help when saving the message in WAL (a hot code path) where we can encode/decode the offset as int64.



Patch:
diff --git a/pkg/isb/interfaces.go b/pkg/isb/interfaces.go
index 5f324340..6aba756d 100644
--- a/pkg/isb/interfaces.go
+++ b/pkg/isb/interfaces.go
@@ -73,17 +73,32 @@ type Offset interface {
 	AckIt() error
 }
 
-// SimpleOffset is an Offset convenient function for implementations without needing AckIt()
-type SimpleOffset func() string
+// SimpleStringOffset is an Offset convenient function for implementations without needing AckIt() when offset is a string.
+type SimpleStringOffset func() string
 
-func (so SimpleOffset) String() string {
+func (so SimpleStringOffset) String() string {
 	return so()
 }
 
-func (so SimpleOffset) Sequence() (int64, error) {
+func (so SimpleStringOffset) Sequence() (int64, error) {
 	return strconv.ParseInt(so(), 10, 64)
 }
 
-func (so SimpleOffset) AckIt() error {
+func (so SimpleStringOffset) AckIt() error {
+	return nil
+}
+
+// SimpleIntOffset is an Offset convenient function for implementations without needing AckIt() when offset is a int64.
+type SimpleIntOffset func() int64
+
+func (si SimpleIntOffset) String() string {
+	return strconv.FormatInt(si(), 10)
+}
+
+func (si SimpleIntOffset) Sequence() (int64, error) {
+	return si(), nil
+}
+
+func (si SimpleIntOffset) AckIt() error {
 	return nil
 }
diff --git a/pkg/isb/stores/redis/read.go b/pkg/isb/stores/redis/read.go
index 74e7ed1d..ebed04e7 100644
--- a/pkg/isb/stores/redis/read.go
+++ b/pkg/isb/stores/redis/read.go
@@ -299,7 +299,7 @@ func (br *BufferRead) convertXStreamToMessages(xstreams []redis.XStream, message
 				}
 				readMessage := isb.ReadMessage{
 					Message:    msg,
-					ReadOffset: isb.SimpleOffset(func() string { return readOffset }),
+					ReadOffset: isb.SimpleStringOffset(func() string { return readOffset }),
 				}
 				messages = append(messages, &readMessage)
 			}
diff --git a/pkg/isb/stores/simplebuffer/buffer.go b/pkg/isb/stores/simplebuffer/buffer.go
index 96d1b293..6f62aa5e 100644
--- a/pkg/isb/stores/simplebuffer/buffer.go
+++ b/pkg/isb/stores/simplebuffer/buffer.go
@@ -112,8 +112,8 @@ func (b *InMemoryBuffer) Write(_ context.Context, messages []isb.Message) ([]isb
 			errs[idx] = nil
 			b.buffer[currentIdx].dirty = true
 			b.writeIdx = (currentIdx + 1) % b.size
-			writeOffsets = append(writeOffsets, isb.SimpleOffset(func() string {
-				return fmt.Sprintf("%d", currentIdx)
+			writeOffsets = append(writeOffsets, isb.SimpleIntOffset(func() int64 {
+				return currentIdx
 			}))
 			// access buffer via lock
 			b.rwlock.Unlock()
@@ -176,7 +176,7 @@ func (b *InMemoryBuffer) Read(ctx context.Context, count int64) ([]*isb.ReadMess
 			}
 		}
 
-		readMessage := isb.ReadMessage{Message: msg, ReadOffset: isb.SimpleOffset(func() string { return strconv.Itoa(int(currentIdx)) })}
+		readMessage := isb.ReadMessage{Message: msg, ReadOffset: isb.SimpleIntOffset(func() int64 { return currentIdx })}
 
 		readMessages = append(readMessages, &readMessage)
 	}
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 2dfda9fc..8a42be69 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -297,7 +297,7 @@ func newreadmessage(payload []byte, offset int64) *isb.ReadMessage {
 	}
 
 	return &isb.ReadMessage{
-		ReadOffset: isb.SimpleOffset(func() string { return strconv.FormatInt(offset, 10) }),
+		ReadOffset: isb.SimpleIntOffset(func() int64 { return offset }),
 		Message:    msg,
 	}
 }
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 637c1cc3..776d611b 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -147,7 +147,7 @@ func New(vertexInstance *dfv1.VertexInstance, writers []isb.BufferWriter, fetchW
 					Payload: msg,
 				},
 			},
-			ReadOffset: isb.SimpleOffset(func() string { return id }),
+			ReadOffset: isb.SimpleStringOffset(func() string { return id }),
 		}
 		h.messages <- m
 		w.WriteHeader(http.StatusNoContent)
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index d8ff511d..37d6c7ce 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -367,7 +367,7 @@ func toReadMessage(m *sarama.ConsumerMessage) *isb.ReadMessage {
 	}
 
 	return &isb.ReadMessage{
-		ReadOffset: isb.SimpleOffset(func() string { return offset }),
+		ReadOffset: isb.SimpleStringOffset(func() string { return offset }),
 		Message:    msg,
 	}
 }


Test patch:
diff --git a/pkg/isb/stores/simplebuffer/buffer_test.go b/pkg/isb/stores/simplebuffer/buffer_test.go
index 1373b30f..d9cfdb71 100644
--- a/pkg/isb/stores/simplebuffer/buffer_test.go
+++ b/pkg/isb/stores/simplebuffer/buffer_test.go
@@ -38,9 +38,9 @@ func TestNewSimpleBuffer(t *testing.T) {
 	// still full as we did not ack
 	assert.Equal(t, true, sb.IsFull())
 
-	err = sb.Ack(ctx, []isb.Offset{isb.SimpleOffset(func() string { return "not_a_number" })})[0]
+	err = sb.Ack(ctx, []isb.Offset{isb.SimpleStringOffset(func() string { return "not_a_number" })})[0]
 	assert.Error(t, err)
-	err = sb.Ack(ctx, []isb.Offset{isb.SimpleOffset(func() string { return "1000" })})[0]
+	err = sb.Ack(ctx, []isb.Offset{isb.SimpleStringOffset(func() string { return "1000" })})[0]
 	assert.Error(t, err)
 
 	errs := sb.Ack(ctx, []isb.Offset{readMessages[0].ReadOffset, readMessages[1].ReadOffset})
diff --git a/pkg/isb/testutils/rw.go b/pkg/isb/testutils/rw.go
index 42318e47..359a8271 100644
--- a/pkg/isb/testutils/rw.go
+++ b/pkg/isb/testutils/rw.go
@@ -50,7 +50,7 @@ func BuildTestReadMessages(count int64, startTime time.Time) []isb.ReadMessage {
 	for idx, writeMessage := range writeMessages {
 		readMessages[idx] = isb.ReadMessage{
 			Message:    writeMessage,
-			ReadOffset: isb.SimpleOffset(func() string { return fmt.Sprintf("read_%s", writeMessage.Header.ID) }),
+			ReadOffset: isb.SimpleStringOffset(func() string { return fmt.Sprintf("read_%s", writeMessage.Header.ID) }),
 		}
 	}
 
diff --git a/pkg/udf/function/uds_grpc_test.go b/pkg/udf/function/uds_grpc_test.go
index 0774bc34..5c59b1fc 100644
--- a/pkg/udf/function/uds_grpc_test.go
+++ b/pkg/udf/function/uds_grpc_test.go
@@ -107,7 +107,7 @@ func TestGRPCBasedUDF_BasicApplyWithMockClient(t *testing.T) {
 					Payload: []byte(`forward_message`),
 				},
 			},
-			ReadOffset: isb.SimpleOffset(func() string { return "0" }),
+			ReadOffset: isb.SimpleStringOffset(func() string { return "0" }),
 		},
 		)
 		assert.NoError(t, err)
@@ -151,7 +151,7 @@ func TestGRPCBasedUDF_BasicApplyWithMockClient(t *testing.T) {
 					Payload: []byte(`forward_message`),
 				},
 			},
-			ReadOffset: isb.SimpleOffset(func() string { return "0" }),
+			ReadOffset: isb.SimpleStringOffset(func() string { return "0" }),
 		},
 		)
 		assert.ErrorIs(t, err, ApplyUDFErr{
diff --git a/pkg/watermark/fetch/edge_fetcher_test.go b/pkg/watermark/fetch/edge_fetcher_test.go
index ab032edc..6d4deb79 100644
--- a/pkg/watermark/fetch/edge_fetcher_test.go
+++ b/pkg/watermark/fetch/edge_fetcher_test.go
@@ -122,7 +122,7 @@ func TestBuffer_GetWatermark(t *testing.T) {
 				processorManager: tt.processorManager,
 				log:              zaptest.NewLogger(t).Sugar(),
 			}
-			if got := b.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(tt.args.offset, 10) })); time.Time(got).In(location) != time.Unix(tt.want, 0).In(location) {
+			if got := b.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(tt.args.offset, 10) })); time.Time(got).In(location) != time.Unix(tt.want, 0).In(location) {
 				t.Errorf("GetWatermark() = %v, want %v", got, processor.Watermark(time.Unix(tt.want, 0)))
 			}
 			// this will always be 14 because the timeline has been populated ahead of time
diff --git a/pkg/watermark/fetch/offset_timeline_test.go b/pkg/watermark/fetch/offset_timeline_test.go
index 828153e1..c069e3f0 100644
--- a/pkg/watermark/fetch/offset_timeline_test.go
+++ b/pkg/watermark/fetch/offset_timeline_test.go
@@ -101,7 +101,7 @@ func TestTimeline_GetEventTime(t1 *testing.T) {
 	}
 	for _, tt := range tests {
 		t1.Run(tt.name, func(t1 *testing.T) {
-			if got := tt.args.timeline.GetEventTime(isb.SimpleOffset(func() string { return strconv.FormatInt(tt.args.inputOffset, 10) })); got != tt.want {
+			if got := tt.args.timeline.GetEventTime(isb.SimpleStringOffset(func() string { return strconv.FormatInt(tt.args.inputOffset, 10) })); got != tt.want {
 				t1.Errorf("GetEventTime() = %v, want %v", got, tt.want)
 			}
 		})
diff --git a/pkg/watermark/fetch/processor_manager_inmem_test.go b/pkg/watermark/fetch/processor_manager_inmem_test.go
index b00ca13b..91611777 100644
--- a/pkg/watermark/fetch/processor_manager_inmem_test.go
+++ b/pkg/watermark/fetch/processor_manager_inmem_test.go
@@ -115,14 +115,14 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
 
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" should be deleted after this GetWatermark offset=101
 	// because "p1" offsetTimeline's head offset=100, which is < inputOffset 103
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 1, len(allProcessors))
 	assert.True(t, allProcessors["p2"].IsActive())
@@ -159,7 +159,7 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" has been deleted from vertex.Processors
 	// so "p1" will be considered as a new processors and a new offsetTimeline watcher for "p1" will be created
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
 	p1 := testBuffer.processorManager.GetProcessor("p1")
 
 	assert.NotNil(t, p1)
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index 451e3a26..a8f0c7d6 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -159,14 +159,14 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
 
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" should be deleted after this GetWatermark offset=101
 	// because "p1" offsetTimeline's head offset=100, which is < inputOffset 103
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 1, len(allProcessors))
 	assert.True(t, allProcessors["p2"].IsActive())
@@ -203,7 +203,7 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" has been deleted from vertex.Processors
 	// so "p1" will be considered as a new processors and a new offsetTimeline watcher for "p1" will be created
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
 	p1 := testBuffer.processorManager.GetProcessor("p1")
 	assert.NotNil(t, p1)
 	assert.True(t, p1.IsActive())
@@ -403,14 +403,14 @@ func TestFetcherWithSeparateOTBucket(t *testing.T) {
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 2, len(allProcessors))
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" should be deleted after this GetWatermark offset=101
 	// because "p1" offsetTimeline's head offset=100, which is < inputOffset 103
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 1, len(allProcessors))
 	assert.True(t, allProcessors["p2"].IsActive())
@@ -444,7 +444,7 @@ func TestFetcherWithSeparateOTBucket(t *testing.T) {
 
 	// "p1" has been deleted from vertex.Processors
 	// so "p1" will be considered as a new processors and a new offsetTimeline watcher for "p1" will be created
-	_ = testBuffer.GetWatermark(isb.SimpleOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
+	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
 	newP1 := testBuffer.processorManager.GetProcessor("p1")
 	assert.NotNil(t, newP1)
 	assert.True(t, newP1.IsActive())
diff --git a/pkg/watermark/publish/publisher_inmem_test.go b/pkg/watermark/publish/publisher_inmem_test.go
index 5d0a4f42..366e3582 100644
--- a/pkg/watermark/publish/publisher_inmem_test.go
+++ b/pkg/watermark/publish/publisher_inmem_test.go
@@ -35,12 +35,12 @@ func TestPublisherWithSharedOTBuckets_InMem(t *testing.T) {
 	var epoch int64 = 1651161600
 	var location, _ = time.LoadLocation("UTC")
 	for i := 0; i < 3; i++ {
-		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(i) }))
+		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
 		epoch += 60
 		time.Sleep(time.Millisecond)
 	}
 	// publish a stale watermark (offset doesn't matter)
-	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(0) }))
+	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
 	keys := p.getAllOTKeysFromBucket()
 	assert.Equal(t, []string{"publisherTestPod1_1651161600", "publisherTestPod1_1651161660", "publisherTestPod1_1651161720"}, keys)
@@ -74,12 +74,12 @@ func TestPublisherWithSeparateOTBucket_InMem(t *testing.T) {
 	var epoch int64 = 1651161600
 	var location, _ = time.LoadLocation("UTC")
 	for i := 0; i < 3; i++ {
-		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(i) }))
+		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
 		epoch += 60
 		time.Sleep(time.Millisecond)
 	}
 	// publish a stale watermark (offset doesn't matter)
-	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(0) }))
+	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
 	keys := p.getAllOTKeysFromBucket()
 	assert.Equal(t, []string{"1651161600", "1651161660", "1651161720"}, keys)
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index 7963c4ba..92179499 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -62,12 +62,12 @@ func TestPublisherWithSeparateOTBuckets(t *testing.T) {
 	var epoch int64 = 1651161600
 	var location, _ = time.LoadLocation("UTC")
 	for i := 0; i < 3; i++ {
-		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(i) }))
+		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
 		epoch += 60
 		time.Sleep(time.Millisecond)
 	}
 	// publish a stale watermark (offset doesn't matter)
-	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(0) }))
+	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
 	keys := p.getAllOTKeysFromBucket()
 	assert.Equal(t, []string{"publisherTestPod1_1651161600", "publisherTestPod1_1651161660", "publisherTestPod1_1651161720"}, keys)
@@ -114,12 +114,12 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	var epoch int64 = 1651161600
 	var location, _ = time.LoadLocation("UTC")
 	for i := 0; i < 3; i++ {
-		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(i) }))
+		p.PublishWatermark(processor.Watermark(time.Unix(epoch, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
 		epoch += 60
 		time.Sleep(time.Millisecond)
 	}
 	// publish a stale watermark (offset doesn't matter)
-	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleOffset(func() string { return strconv.Itoa(0) }))
+	p.PublishWatermark(processor.Watermark(time.Unix(epoch-120, 0).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
 	keys := p.getAllOTKeysFromBucket()
 	assert.Equal(t, []string{"1651161600", "1651161660", "1651161720"}, keys)