Repository: numaproj/numaflow
Pull request #529 (2023-02-07 05:03:09)

feat: Offset time idle watermark put
Add a separate put offset timeline function for assumed idle watermark


Patch:
diff --git a/pkg/watermark/fetch/offset_timeline.go b/pkg/watermark/fetch/offset_timeline.go
index 144d9c60..e3cf875f 100644
--- a/pkg/watermark/fetch/offset_timeline.go
+++ b/pkg/watermark/fetch/offset_timeline.go
@@ -115,6 +115,34 @@ func (t *OffsetTimeline) Put(node OffsetWatermark) {
 	}
 }
 
+// PutIdle inserts the assumed OffsetWatermark which replaces the idle watermark into list. It ensures that the list will remain sorted after the insert.
+func (t *OffsetTimeline) PutIdle(node OffsetWatermark) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	// when inserting an assumed OffsetWatermark, we only need to compare with the head
+	// and can safely skip insertion when any condition doesn't meet
+	if e := t.watermarks.Front(); e != nil {
+		var elementNode = e.Value.(OffsetWatermark)
+		if node.watermark > elementNode.watermark {
+			if node.offset > elementNode.offset {
+				t.watermarks.InsertBefore(node, e)
+				t.watermarks.Remove(t.watermarks.Back())
+				return
+			}
+		} else if node.watermark == elementNode.watermark {
+			if node.offset > elementNode.offset {
+				e.Value = OffsetWatermark{
+					watermark: node.watermark,
+					offset:    node.offset,
+				}
+				return
+			}
+		} else {
+			return
+		}
+	}
+}
+
 // GetHeadOffset returns the head offset, that is the most recent offset which will have the highest
 // Watermark.
 func (t *OffsetTimeline) GetHeadOffset() int64 {
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index 4c71eef8..41a21178 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -259,7 +259,7 @@ func (v *ProcessorManager) startTimeLineWatcher() {
 							// this Vn processor, so it's safe to do nothing
 							watermarkOffset := processor.offsetTimeline.GetHeadOffsetWatermark()
 							if watermarkOffset.offset != -1 {
-								p.offsetTimeline.Put(watermarkOffset)
+								p.offsetTimeline.PutIdle(watermarkOffset)
 								break
 							}
 						}


Test patch:
diff --git a/pkg/watermark/fetch/offset_timeline_test.go b/pkg/watermark/fetch/offset_timeline_test.go
index 2935cd3e..8c6fbd66 100644
--- a/pkg/watermark/fetch/offset_timeline_test.go
+++ b/pkg/watermark/fetch/offset_timeline_test.go
@@ -259,4 +259,25 @@ func TestOffsetTimeline(t *testing.T) {
 	testTimeline.Put(OffsetWatermark{watermark: 30, offset: 35})
 	assert.Equal(t, "[33:36] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
 
+	testTimeline.PutIdle(OffsetWatermark{watermark: 32, offset: 36}) // ignored
+	assert.Equal(t, "[33:36] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(OffsetWatermark{watermark: 33, offset: 35}) // ignored
+	assert.Equal(t, "[33:36] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(OffsetWatermark{watermark: 33, offset: 36}) // ignored
+	assert.Equal(t, "[33:36] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(OffsetWatermark{watermark: 33, offset: 37}) // updated
+	assert.Equal(t, "[33:37] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(OffsetWatermark{watermark: 34, offset: 36}) // ignored
+	assert.Equal(t, "[33:37] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(OffsetWatermark{watermark: 34, offset: 37}) // ignored
+	assert.Equal(t, "[33:37] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(OffsetWatermark{watermark: 34, offset: 38}) // inserted
+	assert.Equal(t, "[34:38] -> [33:37] -> [32:36] -> [30:35] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21]", testTimeline.Dump())
+
 }