Repository: numaproj/numaflow
Pull request #1465 (2024-01-18 06:07:34)

chore: update Windower interface


Patch:
diff --git a/pkg/reduce/pbq/store/errors.go b/pkg/reduce/pbq/store/errors.go
index 5dce1cf9..8f4e6a0b 100644
--- a/pkg/reduce/pbq/store/errors.go
+++ b/pkg/reduce/pbq/store/errors.go
@@ -18,6 +18,6 @@ package store
 
 import "errors"
 
-var WriteStoreFullErr error = errors.New("error writing, store is full")
-var WriteStoreClosedErr error = errors.New("error writing, store is closed")
-var ReadStoreEmptyErr error = errors.New("error reading, store is empty")
+var ErrWriteStoreFull error = errors.New("error writing, store is full")
+var ErrWriteStoreClosed error = errors.New("error writing, store is closed")
+var ErrReadStoreEmpty error = errors.New("error reading, store is empty")
diff --git a/pkg/reduce/pbq/store/memory/store.go b/pkg/reduce/pbq/store/memory/store.go
index 877ecc76..5c7355f1 100644
--- a/pkg/reduce/pbq/store/memory/store.go
+++ b/pkg/reduce/pbq/store/memory/store.go
@@ -41,7 +41,7 @@ type memoryStore struct {
 // this function will be invoked during bootstrap if there is a restart
 func (m *memoryStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 	if m.isEmpty() || m.readPos >= m.writePos {
-		m.log.Errorw(store.ReadStoreEmptyErr.Error())
+		m.log.Errorw(store.ErrReadStoreEmpty.Error())
 		return []*isb.ReadMessage{}, true, nil
 	}
 
@@ -56,12 +56,12 @@ func (m *memoryStore) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 // WriteToStore writes a message to store
 func (m *memoryStore) Write(msg *isb.ReadMessage) error {
 	if m.writePos >= m.storeSize {
-		m.log.Errorw(store.WriteStoreFullErr.Error(), zap.Any("msg header", msg.Header))
-		return store.WriteStoreFullErr
+		m.log.Errorw(store.ErrWriteStoreFull.Error(), zap.Any("msg header", msg.Header))
+		return store.ErrWriteStoreFull
 	}
 	if m.closed {
-		m.log.Errorw(store.WriteStoreClosedErr.Error(), zap.Any("msg header", msg.Header))
-		return store.WriteStoreClosedErr
+		m.log.Errorw(store.ErrWriteStoreClosed.Error(), zap.Any("msg header", msg.Header))
+		return store.ErrWriteStoreClosed
 	}
 	m.storage[m.writePos] = msg
 	m.writePos += 1
diff --git a/pkg/reduce/pbq/store/wal/bootstrap.go b/pkg/reduce/pbq/store/wal/bootstrap.go
index f4f16649..5d5cb624 100644
--- a/pkg/reduce/pbq/store/wal/bootstrap.go
+++ b/pkg/reduce/pbq/store/wal/bootstrap.go
@@ -82,9 +82,9 @@ func decodeWALHeader(buf io.Reader) (*partition.ID, error) {
 	if err != nil {
 		return nil, err
 	}
-	// read the variadic key
-	var key = make([]rune, hp.SLen)
-	err = binary.Read(buf, binary.LittleEndian, key)
+	// read the variadic slot
+	var slot = make([]rune, hp.SLen)
+	err = binary.Read(buf, binary.LittleEndian, slot)
 	if err != nil {
 		return nil, err
 	}
@@ -92,7 +92,7 @@ func decodeWALHeader(buf io.Reader) (*partition.ID, error) {
 	return &partition.ID{
 		Start: time.UnixMilli(hp.S).In(location),
 		End:   time.UnixMilli(hp.E).In(location),
-		Slot:  string(key),
+		Slot:  string(slot),
 	}, nil
 }
 
diff --git a/pkg/reduce/pbq/store/wal/segment.go b/pkg/reduce/pbq/store/wal/segment.go
index ef3fb80a..f7b88ab1 100644
--- a/pkg/reduce/pbq/store/wal/segment.go
+++ b/pkg/reduce/pbq/store/wal/segment.go
@@ -121,7 +121,7 @@ type readMessageHeaderPreamble struct {
 //	| start time (int64) | end time (int64) | slot-len (int16) | slot []rune |
 //	+--------------------+------------------+------------------+-------------+
 //
-// We require the key-len because key is variadic.
+// We require the slot-len because slot is variadic.
 func (w *WAL) encodeWALHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
 	defer func() {
 		if err != nil {
diff --git a/pkg/reduce/pbq/store/wal/stores.go b/pkg/reduce/pbq/store/wal/stores.go
index f58c4f9f..24ae953b 100644
--- a/pkg/reduce/pbq/store/wal/stores.go
+++ b/pkg/reduce/pbq/store/wal/stores.go
@@ -106,6 +106,7 @@ func (ws *walStores) openOrCreateWAL(id *partition.ID) (*WAL, error) {
 			metrics.LabelVertex:             ws.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(ws.replicaIndex)),
 		}).Inc()
+
 		wal = &WAL{
 			fp:                fp,
 			openMode:          os.O_WRONLY,
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 05322b30..a20d346f 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -154,7 +154,7 @@ outerLoop:
 				p.publishWM(ctx, response.Window.Partition())
 
 				// delete the closed windows which are tracked by the windower
-				p.windower.DeleteClosedWindow(response)
+				p.windower.DeleteClosedWindow(response.Window)
 				continue
 			}
 
@@ -192,7 +192,7 @@ outerLoop:
 	// once we have received all the responses from the UDF.
 	p.publishWM(ctx, p.partitionId)
 	// delete the closed windows which are tracked by the windower
-	p.windower.DeleteClosedWindow(&window.TimedWindowResponse{Window: window.NewWindowFromPartition(p.partitionId)})
+	p.windower.DeleteClosedWindow(window.NewWindowFromPartition(p.partitionId))
 
 	// Since we have successfully processed all the messages for a window, we can now delete the persisted messages.
 	err := p.pbqReader.GC()
diff --git a/pkg/window/strategy/fixed/fixed.go b/pkg/window/strategy/fixed/fixed.go
index 28ee298e..d54b0bb7 100644
--- a/pkg/window/strategy/fixed/fixed.go
+++ b/pkg/window/strategy/fixed/fixed.go
@@ -195,8 +195,8 @@ func (w *Windower) NextWindowToBeClosed() window.TimedWindow {
 }
 
 // DeleteClosedWindow deletes the window from the closed windows list.
-func (w *Windower) DeleteClosedWindow(response *window.TimedWindowResponse) {
-	w.closedWindows.Delete(response.Window)
+func (w *Windower) DeleteClosedWindow(window window.TimedWindow) {
+	w.closedWindows.Delete(window)
 }
 
 // OldestWindowEndTime returns the end time of the oldest window among both active and closed windows.
diff --git a/pkg/window/strategy/session/session.go b/pkg/window/strategy/session/session.go
index 4d1fef1a..adb5020c 100644
--- a/pkg/window/strategy/session/session.go
+++ b/pkg/window/strategy/session/session.go
@@ -351,8 +351,8 @@ func (w *Windower) NextWindowToBeClosed() window.TimedWindow {
 }
 
 // DeleteClosedWindow deletes the window from the closed windows list.
-func (w *Windower) DeleteClosedWindow(response *window.TimedWindowResponse) {
-	w.closedWindows.Delete(response.Window)
+func (w *Windower) DeleteClosedWindow(window window.TimedWindow) {
+	w.closedWindows.Delete(window)
 }
 
 // OldestWindowEndTime returns the end time of the oldest window among both active and closed windows.
diff --git a/pkg/window/strategy/sliding/sliding.go b/pkg/window/strategy/sliding/sliding.go
index c3e60fd5..5a119b61 100644
--- a/pkg/window/strategy/sliding/sliding.go
+++ b/pkg/window/strategy/sliding/sliding.go
@@ -214,8 +214,8 @@ func (w *Windower) NextWindowToBeClosed() window.TimedWindow {
 }
 
 // DeleteClosedWindow deletes the window from the closed windows list
-func (w *Windower) DeleteClosedWindow(response *window.TimedWindowResponse) {
-	w.closedWindows.Delete(response.Window)
+func (w *Windower) DeleteClosedWindow(window window.TimedWindow) {
+	w.closedWindows.Delete(window)
 }
 
 // OldestWindowEndTime returns the end time of the oldest window among both active and closed windows.
diff --git a/pkg/window/windower.go b/pkg/window/windower.go
index cbd50a2d..efb5079f 100644
--- a/pkg/window/windower.go
+++ b/pkg/window/windower.go
@@ -41,7 +41,7 @@ type TimedWindower interface {
 	// NextWindowToBeClosed returns the next window yet to be closed.
 	NextWindowToBeClosed() TimedWindow
 	// DeleteClosedWindow deletes the window from the closed windows list.
-	DeleteClosedWindow(response *TimedWindowResponse)
+	DeleteClosedWindow(tw TimedWindow)
 	// OldestWindowEndTime returns the end time of the oldest window among both active and closed windows.
 	// If there are no windows, it returns -1.
 	OldestWindowEndTime() time.Time


Test patch:
diff --git a/pkg/reduce/pbq/pbq_test.go b/pkg/reduce/pbq/pbq_test.go
index 87da0b3c..1300982b 100644
--- a/pkg/reduce/pbq/pbq_test.go
+++ b/pkg/reduce/pbq/pbq_test.go
@@ -193,5 +193,5 @@ func TestPBQ_WriteWithStoreFull(t *testing.T) {
 	}
 	pq.CloseOfBook()
 
-	assert.Error(t, err, store.WriteStoreFullErr)
+	assert.Error(t, err, store.ErrWriteStoreFull)
 }
diff --git a/pkg/window/strategy/fixed/fixed_test.go b/pkg/window/strategy/fixed/fixed_test.go
index 97f44520..18d000ff 100644
--- a/pkg/window/strategy/fixed/fixed_test.go
+++ b/pkg/window/strategy/fixed/fixed_test.go
@@ -178,13 +178,11 @@ func TestFixed_DeleteWindows(t *testing.T) {
 	windower.CloseWindows(baseTime.Add(180 * time.Second))
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(&window.TimedWindowResponse{
-		Window: window.NewWindowFromPartition(&partition.ID{
-			Start: baseTime,
-			End:   baseTime.Add(60 * time.Second),
-			Slot:  "slot-0",
-		}),
-	})
+	windower.DeleteClosedWindow(window.NewWindowFromPartition(&partition.ID{
+		Start: baseTime,
+		End:   baseTime.Add(60 * time.Second),
+		Slot:  "slot-0",
+	}))
 
 	// since we deleted one of the windows, the closed windows should be 2
 	assert.Equal(t, 2, windower.closedWindows.Len())
diff --git a/pkg/window/strategy/session/session_test.go b/pkg/window/strategy/session/session_test.go
index 31a36727..b4b81360 100644
--- a/pkg/window/strategy/session/session_test.go
+++ b/pkg/window/strategy/session/session_test.go
@@ -470,24 +470,20 @@ func TestSession_DeleteWindows(t *testing.T) {
 	windower.CloseWindows(baseTime.Add(180 * time.Second))
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(&window.TimedWindowResponse{
-		Window: window.NewWindowFromPartitionAndKeys(&partition.ID{
-			Start: baseTime,
-			End:   baseTime.Add(60 * time.Second),
-			Slot:  "slot-0",
-		}, []string{"key-1"}),
-	})
+	windower.DeleteClosedWindow(window.NewWindowFromPartitionAndKeys(&partition.ID{
+		Start: baseTime,
+		End:   baseTime.Add(60 * time.Second),
+		Slot:  "slot-0",
+	}, []string{"key-1"}))
 
 	// since we deleted one of the windows, the closed windows should be 3
 	assert.Equal(t, 3, windower.closedWindows.Len())
 
-	windower.DeleteClosedWindow(&window.TimedWindowResponse{
-		Window: window.NewWindowFromPartitionAndKeys(&partition.ID{
-			Start: baseTime.Add(60 * time.Second),
-			End:   baseTime.Add(120 * time.Second),
-			Slot:  "slot-0",
-		}, []string{"key-1"}),
-	})
+	windower.DeleteClosedWindow(window.NewWindowFromPartitionAndKeys(&partition.ID{
+		Start: baseTime.Add(60 * time.Second),
+		End:   baseTime.Add(120 * time.Second),
+		Slot:  "slot-0",
+	}, []string{"key-1"}))
 	// since we deleted two windows, the closed windows should be 2
 	assert.Equal(t, 2, windower.closedWindows.Len())
 }
@@ -533,13 +529,11 @@ func TestWindower_OldestClosedWindowEndTime(t *testing.T) {
 	assert.Equal(t, baseTime.Add(70*time.Second), windower.OldestWindowEndTime())
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(&window.TimedWindowResponse{
-		Window: window.NewWindowFromPartitionAndKeys(&partition.ID{
-			Start: baseTime,
-			End:   baseTime.Add(70 * time.Second),
-			Slot:  "slot-0",
-		}, []string{"key-1"}),
-	})
+	windower.DeleteClosedWindow(window.NewWindowFromPartitionAndKeys(&partition.ID{
+		Start: baseTime,
+		End:   baseTime.Add(70 * time.Second),
+		Slot:  "slot-0",
+	}, []string{"key-1"}))
 
 	// since we deleted (60, 130) window, now the oldest window end time will be 90
 	assert.Equal(t, baseTime.Add(90*time.Second), windower.OldestWindowEndTime())
diff --git a/pkg/window/strategy/sliding/sliding_test.go b/pkg/window/strategy/sliding/sliding_test.go
index 89b8993e..59d6e4db 100644
--- a/pkg/window/strategy/sliding/sliding_test.go
+++ b/pkg/window/strategy/sliding/sliding_test.go
@@ -230,13 +230,11 @@ func TestSliding_DeleteWindows(t *testing.T) {
 	windower.CloseWindows(baseTime.Add(120 * time.Second))
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(&window.TimedWindowResponse{
-		Window: window.NewWindowFromPartition(&partition.ID{
-			Start: baseTime,
-			End:   baseTime.Add(60 * time.Second),
-			Slot:  "slot-0",
-		}),
-	})
+	windower.DeleteClosedWindow(window.NewWindowFromPartition(&partition.ID{
+		Start: baseTime,
+		End:   baseTime.Add(60 * time.Second),
+		Slot:  "slot-0",
+	}))
 
 	// since we deleted one of the windows, the closed windows should be 2
 	assert.Equal(t, 2, windower.closedWindows.Len())