use windower to fetch next window to be closed

Signed-off-by: Yashash H L <[email protected]>

pkg/reduce/data_forward.go

@@ -204,7 +204,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) {
 			return
 		}
 
-		nextWin := df.pbqManager.NextWindowToBeClosed()
+		nextWin := df.pbqManager.NextWindowToBeMaterialized()
 		if nextWin == nil {
 			// if all the windows are closed already, and the len(readBatch) == 0
 			// then it means there's an idle situation
@@ -353,7 +353,7 @@ func (df *DataForward) Process(ctx context.Context, messages []*isb.ReadMessage)
 	}
 
 	// solve Reduce withholding of watermark where we do not send WM until the window is closed.
-	if nextWin := df.pbqManager.NextWindowToBeClosed(); nextWin != nil {
+	if nextWin := df.pbqManager.NextWindowToBeMaterialized(); nextWin != nil {
 		// minus 1 ms because if it's the same as the end time the window would have already been closed
 		if watermark := time.Time(wm).Add(-1 * time.Millisecond); nextWin.EndTime().After(watermark) {
 			// publish idle watermark so that the next vertex doesn't need to wait for the window to close to
@@ -381,7 +381,7 @@ messagesLoop:
 		// drop the late messages only if there is no window open
 		if message.IsLate {
 			// we should be able to get the late message in as long as there is an open window
-			nextWin := df.pbqManager.NextWindowToBeClosed()
+			nextWin := df.windower.NextWindowToBeClosed()
 			// if there is no window open, drop the message
 			if nextWin == nil {
 				df.log.Warnw("Dropping the late message", zap.Time("eventTime", message.EventTime), zap.Time("watermark", message.Watermark))
@@ -429,7 +429,7 @@ messagesLoop:
 		windows := df.upsertWindowsAndKeys(message)
 
 		// for each window we will have a PBQ. A message could belong to multiple windows (e.g., sliding).
-		// We need to write the messages to these PBQs.
+		// We need to write the messages to these PBQs
 		for _, kw := range windows {
 
 			for _, partitionID := range kw.Partitions() {

pkg/reduce/pbq/pbqmanager.go

@@ -284,8 +284,9 @@ func (m *Manager) Replay(ctx context.Context) {
 	m.log.Infow("Finished replaying records from store", zap.Duration("took", time.Since(tm)), zap.Any("partitions", partitionsIds))
 }
 
-// NextWindowToBeClosed returns the next keyed window that is yet to be closed
-func (m *Manager) NextWindowToBeClosed() window.AlignedKeyedWindower {
+// NextWindowToBeMaterialized returns the next keyed window that is yet to be materialized(GCed)
+// will be used by the data forwarder to publish the idle watermark.
+func (m *Manager) NextWindowToBeMaterialized() window.AlignedKeyedWindower {
 	if m.yetToBeClosed.Len() == 0 {
 		return nil
 	}

pkg/reduce/pbq/pbqmanager_test.go

@@ -376,23 +376,23 @@ func TestManager_NextWindowToBeClosed(t *testing.T) {
 	pq4, err = pbqManager.CreateNewPBQ(ctx, partitionFour, kwFour)
 	assert.NoError(t, err)
 
-	aw := pbqManager.NextWindowToBeClosed()
+	aw := pbqManager.NextWindowToBeMaterialized()
 	assert.Equal(t, aw.EndTime(), time.Unix(120, 0))
 
 	_ = pq1.GC()
-	aw = pbqManager.NextWindowToBeClosed()
+	aw = pbqManager.NextWindowToBeMaterialized()
 	assert.Equal(t, aw.EndTime(), time.Unix(120, 0))
 
 	_ = pq2.GC()
-	aw = pbqManager.NextWindowToBeClosed()
+	aw = pbqManager.NextWindowToBeMaterialized()
 	assert.Equal(t, aw.EndTime(), time.Unix(240, 0))
 
 	_ = pq3.GC()
-	aw = pbqManager.NextWindowToBeClosed()
+	aw = pbqManager.NextWindowToBeMaterialized()
 	assert.Equal(t, aw.EndTime(), time.Unix(240, 0))
 
 	_ = pq4.GC()
-	aw = pbqManager.NextWindowToBeClosed()
+	aw = pbqManager.NextWindowToBeMaterialized()
 	assert.Nil(t, aw)
 
 }

pkg/window/strategy/fixed/fixed.go

@@ -83,3 +83,11 @@ func (f *Fixed) InsertIfNotPresent(kw window.AlignedKeyedWindower) (aw window.Al
 func (f *Fixed) RemoveWindows(wm time.Time) []window.AlignedKeyedWindower {
 	return f.entries.RemoveWindows(wm)
 }
+
+// NextWindowToBeClosed returns the next window which is yet to be closed.
+func (f *Fixed) NextWindowToBeClosed() window.AlignedKeyedWindower {
+	if f.entries.Len() == 0 {
+		return nil
+	}
+	return f.entries.Front()
+}

pkg/window/strategy/sliding/sliding.go

@@ -97,3 +97,11 @@ func (s *Sliding) InsertIfNotPresent(kw window.AlignedKeyedWindower) (window.Ali
 func (s *Sliding) RemoveWindows(wm time.Time) []window.AlignedKeyedWindower {
 	return s.entries.RemoveWindows(wm)
 }
+
+// NextWindowToBeClosed returns the next window which is yet to be closed.
+func (s *Sliding) NextWindowToBeClosed() window.AlignedKeyedWindower {
+	if s.entries.Len() == 0 {
+		return nil
+	}
+	return s.entries.Front()
+}

pkg/window/windower.go

@@ -51,4 +51,7 @@ type Windower interface {
 	InsertIfNotPresent(aw AlignedKeyedWindower) (AlignedKeyedWindower, bool)
 	// RemoveWindows returns list of window(s) that can be closed
 	RemoveWindows(time time.Time) []AlignedKeyedWindower
+	// NextWindowToBeClosed returns the next window yet to be closed.
+	// will be used by the data forwarder to check if the late message can be considered for processing.
+	NextWindowToBeClosed() AlignedKeyedWindower
 }