Repository: numaproj/numaflow
Pull request #607 (2023-03-13 20:18:26)

feat: handle watermark barrier for map vertex
Issue: our current implementation won't propagate the watermark when no events flow into the processing pods. This issue causes the pipeline to stop/stuck and also creates latency in data processing. 

We have added some new concepts:
- WMB: short for watermark barrier. A WMB is a watermark with the "Idle=true/false" information. We use the new "idle" field to determine if the watermark is idle (meaning no incoming or outgoing events) or not.
- control message: a control message is a empty message with it's messageKind set to "WMB".  We use control message to identify if the processing pod is idle or not. 

In this PR, we solve the following use case for map vertex.
1. We read a batch, and the len(batch) != 0.
<img width="319" alt="image" src="https://user-images.githubusercontent.com/19543684/225714264-9b4a1181-ed6e-467b-a5a7-16dd91b00e86.png">
2. We read a batch, and the len(batch) == 0. 
(Note: if the WMB fetched doesn't pass the validation, the flow will return early)
<img width="319" alt="image" src="https://user-images.githubusercontent.com/19543684/225717791-6940b773-a617-4c1e-b779-c3f45838f1a6.png">

Test
In this pipeline, we have both even and odd branches.
The source vertex only produces odd number, so there's no data flow in the even branch.
We can see that the watermark is still propagating in the even branch, and the odd sink logs the messages as expected.
Then we let the source vertex send only string values instead of odd numbers, the odd branch also gets no data now. We can still observe watermarks getting updated.
Later we resume source vertex, and odd sink starts to log again.
<img width="1370" alt="Screenshot 2023-03-08 at 2 27 48 PM" src="https://user-images.githubusercontent.com/19543684/225723165-553a699a-4fa8-404f-b2f7-df1a79646642.png">
<img width="727" alt="image" src="https://user-images.githubusercontent.com/19543684/225723660-521152b0-784b-4457-a3a1-739256f50366.png">



TODO: 
1. solve the idling situation for reduce vertex.
    - a partition is idle
    - reduce is withholding the watermark (this is the first problem that triggered all of this)
3. solve the idling situation for source vertex.


yhl25: I know that pr is WIP, was just understanding the code, so left a few comments.

Patch:
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index 1a257d29..cad0614b 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -56,6 +56,10 @@ type InterStepDataForward struct {
 	opts             options
 	vertexName       string
 	pipelineName     string
+	// wmbOffset is a toBufferName to the write offset of the idle watermark map.
+	wmbOffset map[string]isb.Offset
+	// wmbChecker checks if the idle watermark is valid.
+	wmbChecker wmb.WMBChecker
 	Shutdown
 }
 
@@ -83,7 +87,7 @@ func NewInterStepDataForward(vertex *dfv1.Vertex,
 	// creating a context here which is managed by the forwarder's lifecycle
 	ctx, cancel := context.WithCancel(context.Background())
 
-	isdf := InterStepDataForward{
+	var isdf = InterStepDataForward{
 		ctx:              ctx,
 		cancelFn:         cancel,
 		fromBuffer:       fromStep,
@@ -95,6 +99,8 @@ func NewInterStepDataForward(vertex *dfv1.Vertex,
 		// should we do a check here for the values not being null?
 		vertexName:   vertex.Spec.Name,
 		pipelineName: vertex.Spec.PipelineName,
+		wmbOffset:    make(map[string]isb.Offset, len(toSteps)),
+		wmbChecker:   wmb.NewWMBChecker(2), // TODO: make configurable
 		Shutdown: Shutdown{
 			rwlock: new(sync.RWMutex),
 		},
@@ -203,15 +209,46 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	// process only if we have any read messages. There is a natural looping here if there is an internal error while
 	// reading, and we are not able to proceed.
 	if len(readMessages) == 0 {
+		// When the read length is zero, the write length is definitely zero too
+		// meaning there's no data to be published to the next vertex, and we consider this
+		// situation as idling.
+		// In order to continue propagating watermark, we will set watermark idle=true and publish it.
+		// We also publish a control message if this is the first time we get this idle situation.
+
+		// we use the HeadWMB as the watermark for the idle
+		var processorWMB = isdf.fetchWatermark.GetHeadWMB()
+		if !isdf.wmbChecker.ValidateHeadWMB(processorWMB) {
+			// validation failed, skip publishing
+			isdf.opts.logger.Debugw("skip publishing idle watermark",
+				zap.Int("counter", isdf.wmbChecker.GetCounter()),
+				zap.Int64("offset", processorWMB.Offset),
+				zap.Int64("watermark", processorWMB.Watermark),
+				zap.Bool("Idle", processorWMB.Idle))
+			return
+		}
+		for _, toBuffer := range isdf.toBuffers {
+			isdf.publishIdleWatermark(toBuffer, wmb.Watermark(time.UnixMilli(processorWMB.Watermark)))
+		}
 		return
 	}
 
+	var dataMessages = make([]*isb.ReadMessage, 0, len(readMessages))
+
+	// store the offsets of the messages we read from ISB
+	var readOffsets = make([]isb.Offset, len(readMessages))
+	for idx, m := range readMessages {
+		readOffsets[idx] = m.ReadOffset
+		if m.Kind == isb.Data {
+			dataMessages = append(dataMessages, m)
+		}
+	}
+
 	// create space for writeMessages specific to each step as we could forward to all the steps too.
 	var messageToStep = make(map[string][]isb.Message)
 	var toBuffers string // logging purpose
 	for buffer := range isdf.toBuffers {
 		// over allocating to have a predictable pattern
-		messageToStep[buffer] = make([]isb.Message, 0, len(readMessages))
+		messageToStep[buffer] = make([]isb.Message, 0, len(dataMessages))
 		toBuffers += buffer + ","
 	}
 
@@ -219,7 +256,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	udfCh := make(chan *readWriteMessagePair)
 	// udfResults stores the results after UDF processing for all read messages. It indexes
 	// a read message to the corresponding write message
-	udfResults := make([]readWriteMessagePair, len(readMessages))
+	udfResults := make([]readWriteMessagePair, len(dataMessages))
 	// applyUDF, if there is an Internal error it is a blocking call and will return only if shutdown has been initiated.
 
 	// create a pool of UDF Processors
@@ -247,7 +284,8 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		processorWM = isdf.fetchWatermark.GetWatermark(readMessages[0].ReadOffset)
 	}
 
-	for idx, m := range readMessages {
+	// send to UDF only the data messages
+	for idx, m := range dataMessages {
 		// emit message size metric
 		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, "buffer": isdf.fromBuffer.GetName()}).Add(float64(len(m.Payload)))
 		// assign watermark to the message. assign time.UnixMilli(-1) as watermark when we are at source vertex.
@@ -330,32 +368,37 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 				if len(offsets) > 0 {
 					publisher.PublishWatermark(processorWM, offsets[len(offsets)-1])
 					activeWatermarkBuffers[bufferName] = true
+					// reset because the toBuffer is not idling
+					isdf.wmbOffset[bufferName] = nil
 				}
 				// This (len(offsets) == 0) happens at conditional forwarding, there's no data written to the buffer
-				// TODO: Should also publish to those edges without writing (fall out of conditional forwarding)
 			} else { // For Sink vertex, and it does not care about the offset during watermark publishing
 				publisher.PublishWatermark(processorWM, nil)
 				activeWatermarkBuffers[bufferName] = true
+				// reset because the toBuffer is not idling
+				isdf.wmbOffset[bufferName] = nil
 			}
 		}
 	}
-	if len(activeWatermarkBuffers) < len(isdf.publishWatermark) {
-		// if there's any buffers that haven't received any watermark during this
-		// batch processing cycle, send an idle watermark
+	if len(dataMessages) > 0 && len(activeWatermarkBuffers) < len(isdf.publishWatermark) {
+		// - condition1 "len(dataMessages) > 0" :
+		//   Meaning, we do have some data messages, but not all out buffers get written.
+		//   Tt could be all data messages are dropped, or conditional forwarding to part of the out buffers.
+		//   If we don't have this condition check, when dataMessages is zero but ctrlMessages > 0, we will
+		//   wrongly publish an idle watermark without the ctrl message and the ctrl message tracking map.
+		// - condition 2 "len(activeWatermarkBuffers) < len(isdf.publishWatermark)" :
+		//   send idle watermark only if we have idle out buffers
 		for bufferName := range isdf.publishWatermark {
 			if !activeWatermarkBuffers[bufferName] {
 				// use the watermark of the current read batch for the idle watermark
-				isdf.publishWatermark[bufferName].PublishIdleWatermark(processorWM)
+				// same as read len==0 because there's no event published to the buffer
+				isdf.publishIdleWatermark(isdf.toBuffers[bufferName], processorWM)
 			}
 		}
 	}
 
-	// let us ack the chunk only if we have successfully forwarded all the messages.
-	// we need the readOffsets to acknowledge later
-	var readOffsets = make([]isb.Offset, len(readMessages))
-	for idx, m := range udfResults {
-		readOffsets[idx] = m.readMessage.ReadOffset
-	}
+	// when we apply udf, we don't handle partial errors (it's either non or all, non will return early),
+	// so we should be able to ack all the readOffsets including data messages and control messages
 	err = isdf.ackFromBuffer(ctx, readOffsets)
 	// implicit return for posterity :-)
 	if err != nil {
@@ -564,6 +607,50 @@ func (isdf *InterStepDataForward) whereToStep(writeMessage *isb.Message, message
 	return nil
 }
 
+// publishIdleWatermark publishes a ctrl message with isb.Kind set to WMB when write length is zero
+// We only send one ctrl message when we see a new WMB; later we only update the WMB without a ctrl
+// message.
+func (isdf *InterStepDataForward) publishIdleWatermark(toBuffer isb.BufferWriter, wm wmb.Watermark) {
+	var bufferName = toBuffer.GetName()
+
+	if isdf.wmbOffset[bufferName] == nil {
+		// the map is nil means we get a new idle situation
+		if isdf.opts.vertexType == dfv1.VertexTypeSink {
+			// for Sink vertex, we don't need to write any ctrl message
+			// and because when we publish the watermark, offset is not important for sink
+			// so, we do nothing here
+		} else {
+			// if wmbOffset is nil, create a new WMB and write a ctrl message to ISB
+			var ctrlMessage = []isb.Message{{Header: isb.Header{Kind: isb.WMB}}}
+			writeOffsets, err := isdf.writeToBuffer(isdf.ctx, toBuffer, ctrlMessage)
+			if err != nil {
+				isdf.opts.logger.Errorw("failed to write ctrl message to buffer", zap.String("bufferName", bufferName), zap.Error(err))
+				return
+			}
+			isdf.opts.logger.Debug("succeeded to write ctrl message to buffer", zap.String("bufferName", bufferName), zap.Error(err))
+
+			if len(writeOffsets) == 1 {
+				// we only write one ctrl message, so there's only one offset in the array, use index=0 to get the offset
+				isdf.wmbOffset[bufferName] = writeOffsets[0]
+			} else {
+				// if sink vertex, then there's no write offset
+				isdf.wmbOffset[bufferName] = isb.SimpleIntOffset(func() int64 { return 0 })
+			}
+		}
+	}
+
+	// publish WMB (this will naturally incr or set the timestamp of isdf.wmbOffset)
+	if publisher, ok := isdf.publishWatermark[bufferName]; ok {
+		if isdf.opts.vertexType == dfv1.VertexTypeSource || isdf.opts.vertexType == dfv1.VertexTypeMapUDF ||
+			isdf.opts.vertexType == dfv1.VertexTypeReduceUDF {
+			publisher.PublishIdleWatermark(wm, isdf.wmbOffset[bufferName])
+		} else {
+			// for Sink vertex, and it does not care about the offset during watermark publishing
+			publisher.PublishIdleWatermark(wm, nil)
+		}
+	}
+}
+
 // errorArrayToMap summarizes an error array to map
 func errorArrayToMap(errs []error) map[string]int64 {
 	result := make(map[string]int64)
diff --git a/pkg/isb/message.go b/pkg/isb/message.go
index 5933f03e..9e5f54ec 100644
--- a/pkg/isb/message.go
+++ b/pkg/isb/message.go
@@ -24,18 +24,15 @@ import (
 type MessageKind int16
 
 const (
-	Data MessageKind = 1 << iota // Data payload
-	WMB                          // Watermark Barrier
+	Data MessageKind = iota // Data payload
+	WMB                     // Watermark Barrier
 )
 
-func (mta MessageKind) String() string {
-	var mt MessageKind
+func (mt MessageKind) String() string {
 	switch mt {
-	case 0:
-		return "Empty"
-	case mt & Data:
+	case Data:
 		return "Data"
-	case mt & WMB:
+	case WMB:
 		return "WMB"
 	default:
 		return "Unknown"
diff --git a/pkg/isb/stores/simplebuffer/buffer.go b/pkg/isb/stores/simplebuffer/buffer.go
index 4c7e667a..9a202f4f 100644
--- a/pkg/isb/stores/simplebuffer/buffer.go
+++ b/pkg/isb/stores/simplebuffer/buffer.go
@@ -253,3 +253,14 @@ func (b *InMemoryBuffer) Ack(_ context.Context, offsets []isb.Offset) []error {
 
 	return errs
 }
+
+// GetMessages gets the first num messages in the in mem buffer
+// this function is for testing purpose
+func (b *InMemoryBuffer) GetMessages(num int) []*isb.Message {
+	var msgs = make([]*isb.Message, 0, num)
+	for i := 0; i < num && i < len(b.buffer); i++ {
+		msg, _ := buildMessage(b.buffer[i].header, b.buffer[i].body)
+		msgs = append(msgs, &msg)
+	}
+	return msgs
+}
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index 48a2db7d..c40f95cd 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -185,5 +185,7 @@ func (d *DataForward) forwardAChunk(ctx context.Context) {
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(d.vertexReplica)),
 		"buffer":                        d.fromBuffer.GetName(),
 	}).Add(float64(totalBytes))
+
+	// readMessages has to be written to PBQ, acked, etc.
 	d.readloop.Process(ctx, readMessages)
 }
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 1ae4296f..036779a8 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -267,7 +267,9 @@ func (p *ProcessAndForward) publishWM(wm wmb.Watermark, writeOffsets map[string]
 		for bufferName := range p.publishWatermark {
 			if !activeWatermarkBuffers[bufferName] {
 				// use the watermark of the current read batch for the idle watermark
-				p.publishWatermark[bufferName].PublishIdleWatermark(wm)
+				// we don't care about the offset here because, in this use case,
+				// we will use a referred watermark to replace this idle watermark
+				p.publishWatermark[bufferName].PublishIdleWatermark(wm, isb.SimpleIntOffset(func() int64 { return -1 }))
 			}
 		}
 	}
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index b1c1d40b..697e6ade 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -137,18 +137,34 @@ func (rl *ReadLoop) Startup(ctx context.Context) error {
 // Process is one iteration of the read loop which writes the messages to the PBQs followed by acking the messages, and
 // then closing the windows that can closed.
 func (rl *ReadLoop) Process(ctx context.Context, messages []*isb.ReadMessage) {
+	var dataMessages = make([]*isb.ReadMessage, 0, len(messages))
+	var ctrlMessages = make([]*isb.ReadMessage, 0) // for a high TPS pipeline, 0 is the most optimal value
+
+	for _, message := range messages {
+		if message.Kind == isb.Data {
+			dataMessages = append(dataMessages, message)
+		} else {
+			ctrlMessages = append(ctrlMessages, message)
+		}
+	}
+
 	// write messages to windows based by PBQs.
-	successfullyWrittenMessages, err := rl.writeMessagesToWindows(ctx, messages)
+	successfullyWrittenMessages, err := rl.writeMessagesToWindows(ctx, dataMessages)
 	if err != nil {
 		rl.log.Errorw("Failed to write messages", zap.Int("totalMessages", len(messages)), zap.Int("writtenMessage", len(successfullyWrittenMessages)))
 	}
 
+	// ack the control messages
+	if len(ctrlMessages) != 0 {
+		rl.ackMessages(ctx, ctrlMessages)
+	}
+
 	if len(successfullyWrittenMessages) == 0 {
 		return
 	}
-
 	// ack successful messages
 	rl.ackMessages(ctx, successfullyWrittenMessages)
+
 	// close any windows that need to be closed.
 	// since the watermark will be same for all the messages in the batch
 	// we can invoke remove windows only once per batch
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 6a7986c5..729c575d 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -227,7 +227,6 @@ loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-h.messages:
-			fmt.Println("got a message to read")
 			httpSourceReadCount.With(map[string]string{metrics.LabelVertex: h.name, metrics.LabelPipeline: h.pipelineName}).Inc()
 			msgs = append(msgs, m)
 		case <-timeout:
diff --git a/pkg/watermark/fetch/edge_fetcher.go b/pkg/watermark/fetch/edge_fetcher.go
index 79c35d55..03876c58 100644
--- a/pkg/watermark/fetch/edge_fetcher.go
+++ b/pkg/watermark/fetch/edge_fetcher.go
@@ -53,9 +53,40 @@ func NewEdgeFetcher(ctx context.Context, bufferName string, storeWatcher store.W
 	}
 }
 
-// GetHeadWatermark returns the watermark using the HeadOffset (the latest offset among all processors). This
-// can be used in showing the watermark progression for a vertex when not consuming the messages
-// directly (eg. UX, tests)
+// GetWatermark gets the smallest timestamp for the given offset
+func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) wmb.Watermark {
+	var offset, err = inputOffset.Sequence()
+	if err != nil {
+		e.log.Errorw("Unable to get offset from isb.Offset.Sequence()", zap.Error(err))
+		return wmb.Watermark(time.Unix(-1, 0))
+	}
+	var debugString strings.Builder
+	var epoch int64 = math.MaxInt64
+	var allProcessors = e.processorManager.GetAllProcessors()
+	for _, p := range allProcessors {
+		debugString.WriteString(fmt.Sprintf("[Processor: %v] \n", p))
+		var t = p.offsetTimeline.GetEventTime(inputOffset)
+		if t == -1 { // watermark cannot be computed, perhaps a new processing unit was added or offset fell off the timeline
+			epoch = t
+		} else if t < epoch {
+			epoch = t
+		}
+		if p.IsDeleted() && (offset > p.offsetTimeline.GetHeadOffset()) {
+			// if the pod is not active and the current offset is ahead of all offsets in Timeline
+			e.processorManager.DeleteProcessor(p.entity.GetName())
+		}
+	}
+	// if there are no processors
+	if epoch == math.MaxInt64 {
+		epoch = -1
+	}
+	e.log.Debugf("%s[%s] get watermark for offset %d: %+v", debugString.String(), e.bufferName, offset, epoch)
+
+	return wmb.Watermark(time.UnixMilli(epoch))
+}
+
+// GetHeadWatermark returns the latest watermark among all processors. This can be used in showing the watermark
+// progression for a vertex when not consuming the messages directly (eg. UX, tests)
 // NOTE
 //   - We don't use this function in the regular pods in the vertex.
 //   - UX only uses GetHeadWatermark, so the `p.IsDeleted()` check in the GetWatermark never happens.
@@ -70,12 +101,12 @@ func (e *edgeFetcher) GetHeadWatermark() wmb.Watermark {
 		if !p.IsActive() {
 			continue
 		}
-		var o = p.offsetTimeline.GetHeadOffset()
-		e.log.Debugf("Processor: %v (headoffset:%d)", p, o)
-		debugString.WriteString(fmt.Sprintf("[Processor:%v] (headoffset:%d) \n", p, o))
-		if o != -1 && o > headOffset {
-			headOffset = o
-			epoch = p.offsetTimeline.GetEventTimeFromInt64(o)
+		var w = p.offsetTimeline.GetHeadWMB()
+		e.log.Debugf("Processor: %v (headOffset:%d) (headWatermark:%d) (headIdle:%t)", p, w.Offset, w.Watermark, w.Idle)
+		debugString.WriteString(fmt.Sprintf("[Processor:%v] (headOffset:%d) (headWatermark:%d) (headIdle:%t) \n", p, w.Offset, w.Watermark, w.Idle))
+		if w.Offset != -1 && w.Offset > headOffset {
+			headOffset = w.Offset
+			epoch = w.Watermark
 		}
 	}
 	e.log.Debugf("GetHeadWatermark: %s", debugString.String())
@@ -86,36 +117,31 @@ func (e *edgeFetcher) GetHeadWatermark() wmb.Watermark {
 	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
-// GetWatermark gets the smallest timestamp for the given offset
-func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) wmb.Watermark {
-	var offset, err = inputOffset.Sequence()
-	if err != nil {
-		e.log.Errorw("Unable to get offset from isb.Offset.Sequence()", zap.Error(err))
-		return wmb.Watermark(time.Unix(-1, 0))
+// GetHeadWMB returns the latest idle WMB among all processors
+func (e *edgeFetcher) GetHeadWMB() wmb.WMB {
+	var headWMB = wmb.WMB{
+		// we find the head WMB only based on Offset
+		Offset: math.MinInt64,
 	}
-	var debugString strings.Builder
-	var epoch int64 = math.MaxInt64
+	// if any head wmb from Vn-1 processors is not idle, we skip publishing
 	var allProcessors = e.processorManager.GetAllProcessors()
 	for _, p := range allProcessors {
-		debugString.WriteString(fmt.Sprintf("[Processor: %v] \n", p))
-		var t = p.offsetTimeline.GetEventTime(inputOffset)
-		if t == -1 { // watermark cannot be computed, perhaps a new processing unit was added or offset fell off the timeline
-			epoch = t
-		} else if t < epoch {
-			epoch = t
+		if !p.IsActive() {
+			continue
 		}
-		if p.IsDeleted() && (offset > p.offsetTimeline.GetHeadOffset()) {
-			// if the pod is not active and the current offset is ahead of all offsets in Timeline
-			e.processorManager.DeleteProcessor(p.entity.GetName())
+		var curHeadWMB = p.offsetTimeline.GetHeadWMB()
+		if !curHeadWMB.Idle {
+			return wmb.WMB{}
+		}
+		if curHeadWMB.Offset != -1 && curHeadWMB.Offset > headWMB.Offset {
+			headWMB = curHeadWMB
 		}
 	}
-	// if there are no processors
-	if epoch == math.MaxInt64 {
-		epoch = -1
+	if headWMB.Offset == math.MinInt64 {
+		// there is no valid watermark yet
+		return wmb.WMB{}
 	}
-	e.log.Debugf("%s[%s] get watermark for offset %d: %+v", debugString.String(), e.bufferName, offset, epoch)
-
-	return wmb.Watermark(time.UnixMilli(epoch))
+	return headWMB
 }
 
 // Close function closes the watchers.
diff --git a/pkg/watermark/fetch/interface.go b/pkg/watermark/fetch/interface.go
index 706e53a4..e6b7f539 100644
--- a/pkg/watermark/fetch/interface.go
+++ b/pkg/watermark/fetch/interface.go
@@ -28,6 +28,8 @@ type Fetcher interface {
 	io.Closer
 	// GetWatermark returns the inorder monotonically increasing watermark of the edge connected to Vn-1.
 	GetWatermark(offset isb.Offset) wmb.Watermark
-	// GetHeadWatermark returns the latest watermark based on the head offset
+	// GetHeadWatermark returns the latest watermark among all processors
 	GetHeadWatermark() wmb.Watermark
+	// GetHeadWMB returns the latest idle WMB among all processors
+	GetHeadWMB() wmb.WMB
 }
diff --git a/pkg/watermark/fetch/offset_timeline.go b/pkg/watermark/fetch/offset_timeline.go
index ad355cc9..5456a483 100644
--- a/pkg/watermark/fetch/offset_timeline.go
+++ b/pkg/watermark/fetch/offset_timeline.go
@@ -23,6 +23,7 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 	"go.uber.org/zap"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -51,24 +52,15 @@ func NewOffsetTimeline(ctx context.Context, c int) *OffsetTimeline {
 	}
 
 	for i := 0; i < c; i++ {
-		offsetTimeline.watermarks.PushBack(OffsetWatermark{
-			watermark: -1,
-			offset:    -1,
+		offsetTimeline.watermarks.PushBack(wmb.WMB{
+			Watermark: -1,
+			Offset:    -1,
 		})
 	}
 
 	return &offsetTimeline
 }
 
-// OffsetWatermark stores the maximum offset for the given event time
-// we use basic data type int64 to compare the value
-type OffsetWatermark struct {
-	// watermark is derived from fetch.Watermark
-	watermark int64
-	// offset is derived from isb.Offset
-	offset int64
-}
-
 // Capacity returns the capacity of the OffsetTimeline list.
 func (t *OffsetTimeline) Capacity() int {
 	t.lock.RLock()
@@ -76,31 +68,31 @@ func (t *OffsetTimeline) Capacity() int {
 	return t.capacity
 }
 
-// Put inserts the OffsetWatermark into list. It ensures that the list will remain sorted after the insert.
-func (t *OffsetTimeline) Put(node OffsetWatermark) {
+// Put inserts the WMB into list. It ensures that the list will remain sorted after the insert.
+func (t *OffsetTimeline) Put(node wmb.WMB) {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 	// The `for` loop's amortized time complexity should be O(1). Since the OffsetTimeline is sorted by time and
 	// our access pattern will always hit the most recent data.
 	for e := t.watermarks.Front(); e != nil; e = e.Next() {
-		var elementNode = e.Value.(OffsetWatermark)
-		if node.watermark == elementNode.watermark {
-			// we store the maximum offset for the given event time
-			if node.offset > elementNode.offset {
-				e.Value = OffsetWatermark{
-					watermark: node.watermark,
-					offset:    node.offset,
+		var elementNode = e.Value.(wmb.WMB)
+		if node.Watermark == elementNode.Watermark {
+			// we store the maximum Offset for the given event time
+			if node.Offset > elementNode.Offset {
+				e.Value = wmb.WMB{
+					Watermark: node.Watermark,
+					Offset:    node.Offset,
 				}
 				return
 			} else {
-				t.log.Errorw("The new input offset should never be smaller than the existing offset", zap.Int64("watermark", node.watermark),
-					zap.Int64("existingOffset", elementNode.offset), zap.Int64("inputOffset", node.offset))
+				t.log.Errorw("The new input offset should never be smaller than the existing offset", zap.Int64("watermark", node.Watermark),
+					zap.Int64("existingOffset", elementNode.Offset), zap.Int64("inputOffset", node.Offset))
 				return
 			}
-		} else if node.watermark > elementNode.watermark {
-			if node.offset < elementNode.offset {
-				t.log.Errorw("The new input offset should never be smaller than the existing offset", zap.Int64("watermark", node.watermark),
-					zap.Int64("existingOffset", elementNode.offset), zap.Int64("inputOffset", node.offset))
+		} else if node.Watermark > elementNode.Watermark {
+			if node.Offset < elementNode.Offset {
+				t.log.Errorw("The new input offset should never be smaller than the existing offset", zap.Int64("watermark", node.Watermark),
+					zap.Int64("existingOffset", elementNode.Offset), zap.Int64("inputOffset", node.Offset))
 				return
 			}
 			// our list is sorted by event time from highest to lowest
@@ -115,25 +107,79 @@ 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) {
+// PutIdle inserts the idle WMB into list or update the current idle WMB.
+func (t *OffsetTimeline) PutIdle(node wmb.WMB) {
 	t.lock.Lock()
 	defer t.lock.Unlock()
-	// when inserting an assumed OffsetWatermark, we only need to compare with the head
+	// when inserting an idle WMB, 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 {
+		var elementNode = e.Value.(wmb.WMB)
+		if elementNode.Idle {
+			if node.Watermark < elementNode.Watermark {
+				// should never happen
+				return
+			}
+			if elementNode.Offset == node.Offset {
+				e.Value = wmb.WMB{
+					Idle:      true,
+					Offset:    node.Offset,
+					Watermark: node.Watermark,
+				}
+			} else if node.Offset > elementNode.Offset {
+				// NOTE: Valid case for the warning:
+				// If we have an active watermark between two idle watermarks, but this active watermark is skipped
+				// publishing due to "skip publishing same watermark", we will see this warning.
+				// In this case, it's safe to insert the idle watermark.
+				t.log.Warnw("The idle watermark has a larger offset from the head idle watermark", zap.Int64("idleWatermark", node.Watermark),
+					zap.Int64("existingOffset", elementNode.Offset), zap.Int64("inputOffset", node.Offset))
+				t.watermarks.InsertBefore(node, e)
+				t.watermarks.Remove(t.watermarks.Back())
+			}
+			return
+		}
+		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,
+		} else if node.Watermark == elementNode.Watermark {
+			if node.Offset > elementNode.Offset {
+				e.Value = wmb.WMB{
+					Idle:      true,
+					Watermark: node.Watermark,
+					Offset:    node.Offset,
+				}
+				return
+			}
+		} else {
+			return
+		}
+	}
+}
+
+// PutReferred inserts the referred WMB, which is a validated watermark copied from other timelines, to replace
+// the idle watermark into list. It ensures that the list will remain sorted after the insert.
+// TODO: clean up and refine the other two put methods after we remove copying over strategy
+func (t *OffsetTimeline) PutReferred(node wmb.WMB) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	// when inserting a referred WMB, 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.(wmb.WMB)
+		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 = wmb.WMB{
+					Watermark: node.Watermark,
+					Offset:    node.Offset,
 				}
 				return
 			}
@@ -151,7 +197,7 @@ func (t *OffsetTimeline) GetHeadOffset() int64 {
 	if t.watermarks.Len() == 0 {
 		return -1
 	}
-	return t.watermarks.Front().Value.(OffsetWatermark).offset
+	return t.watermarks.Front().Value.(wmb.WMB).Offset
 }
 
 // GetHeadWatermark returns the head watermark, which is the highest one.
@@ -161,33 +207,33 @@ func (t *OffsetTimeline) GetHeadWatermark() int64 {
 	if t.watermarks.Len() == 0 {
 		return 0
 	}
-	return t.watermarks.Front().Value.(OffsetWatermark).watermark
+	return t.watermarks.Front().Value.(wmb.WMB).Watermark
 }
 
-// GetHeadOffsetWatermark returns the largest offset with the largest watermark.
-func (t *OffsetTimeline) GetHeadOffsetWatermark() OffsetWatermark {
+// GetHeadWMB returns the largest offset with the largest watermark.
+func (t *OffsetTimeline) GetHeadWMB() wmb.WMB {
 	t.lock.RLock()
 	defer t.lock.RUnlock()
-	return t.watermarks.Front().Value.(OffsetWatermark)
+	return t.watermarks.Front().Value.(wmb.WMB)
 }
 
-// GetReferredWatermark returns the referred watermark that will be used to replace
-// the idle watermark value
-func (t *OffsetTimeline) GetReferredWatermark(idleWM int64) OffsetWatermark {
+// GetReferredWatermark returns a referred watermark, which is a validated watermark copied from other timelines,
+// to replace the idle watermark value
+func (t *OffsetTimeline) GetReferredWatermark(idleWM int64) wmb.WMB {
 	t.lock.RLock()
 	defer t.lock.RUnlock()
-	// find in the offset timeline where the OffsetWatermark has a watermark that is <= the idleWM
+	// find in the offset timeline where the WMB has an active watermark that is <= the idleWM
 	// because, when we replace the idleWM, we need to guarantee the referred watermark's value won't exceed
 	// the largest WM processed from Vn-1's idle processor (meaning the idle processor hasn't seen any data that
 	// is later than this idleWM)
 	for e := t.watermarks.Front(); e != nil; e = e.Next() {
-		if e.Value.(OffsetWatermark).watermark <= idleWM {
-			return e.Value.(OffsetWatermark)
+		if e.Value.(wmb.WMB).Watermark <= idleWM && !e.Value.(wmb.WMB).Idle {
+			return e.Value.(wmb.WMB)
 		}
 	}
-	return OffsetWatermark{
-		watermark: -1,
-		offset:    -1,
+	return wmb.WMB{
+		Watermark: -1,
+		Offset:    -1,
 	}
 }
 
@@ -198,8 +244,8 @@ func (t *OffsetTimeline) GetOffset(eventTime int64) int64 {
 	defer t.lock.RUnlock()
 
 	for e := t.watermarks.Front(); e != nil; e = e.Next() {
-		if eventTime >= e.Value.(OffsetWatermark).watermark {
-			return e.Value.(OffsetWatermark).offset
+		if eventTime >= e.Value.(wmb.WMB).Watermark {
+			return e.Value.(wmb.WMB).Offset
 		}
 	}
 
@@ -224,8 +270,8 @@ func (t *OffsetTimeline) GetEventTimeFromInt64(inputOffsetInt64 int64) int64 {
 	for e := t.watermarks.Front(); e != nil; e = e.Next() {
 		// get the event time has the closest offset to the input offset
 		// exclude the same offset because this offset may not finish processing yet
-		if e.Value.(OffsetWatermark).offset < inputOffsetInt64 {
-			eventTime = e.Value.(OffsetWatermark).watermark
+		if e.Value.(wmb.WMB).Offset < inputOffsetInt64 {
+			eventTime = e.Value.(wmb.WMB).Watermark
 			break
 		}
 	}
@@ -240,7 +286,11 @@ func (t *OffsetTimeline) Dump() string {
 	t.lock.RLock()
 	defer t.lock.RUnlock()
 	for e := t.watermarks.Front(); e != nil; e = e.Next() {
-		builder.WriteString(fmt.Sprintf("[%d:%d] -> ", e.Value.(OffsetWatermark).watermark, e.Value.(OffsetWatermark).offset))
+		if e.Value.(wmb.WMB).Idle {
+			builder.WriteString(fmt.Sprintf("[IDLE %d:%d] -> ", e.Value.(wmb.WMB).Watermark, e.Value.(wmb.WMB).Offset))
+		} else {
+			builder.WriteString(fmt.Sprintf("[%d:%d] -> ", e.Value.(wmb.WMB).Watermark, e.Value.(wmb.WMB).Offset))
+		}
 	}
 	if builder.Len() < 4 {
 		return ""
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index 2bc52e74..760ca99e 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -250,30 +250,34 @@ func (v *ProcessorManager) startTimeLineWatcher() {
 					continue
 				}
 				if otValue.Idle {
-					var processors = v.GetAllProcessors()
-					for processorName, _processor := range processors {
-						// skip the idle processor itself, only use other processors as reference
-						if processorName != value.Key() {
-							// in any other Vn-1 processor's offset timeline, we can replace the idle watermark
-							// with any watermark whose watermark is <= otValue.Watermark
-							referredWatermarkOffset := _processor.offsetTimeline.GetReferredWatermark(otValue.Watermark)
-							// if the referred watermark is empty, skip
-							if referredWatermarkOffset.watermark != -1 {
-								p.offsetTimeline.PutIdle(referredWatermarkOffset)
-								v.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", v.otWatcher.GetKVName()), zap.Int64("referredWatermark", referredWatermarkOffset.watermark), zap.Int64("offset", referredWatermarkOffset.offset))
-								break
+					if otValue.Offset != -1 {
+						// the idle watermark is not coming from a reduce vertex
+						// directly insert the idle watermark
+						p.offsetTimeline.PutIdle(otValue)
+						v.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", v.otWatcher.GetKVName()), zap.Int64("idleWatermark", otValue.Watermark), zap.Int64("idleOffset", otValue.Offset))
+					} else {
+						var processors = v.GetAllProcessors()
+						for processorName, _processor := range processors {
+							// skip the idle processor itself, only use other processors as reference
+							if processorName != value.Key() {
+								// in any other Vn-1 processor's offset timeline, we can replace the idle watermark
+								// with any watermark whose watermark is <= otValue.Watermark
+								referredWMB := _processor.offsetTimeline.GetReferredWatermark(otValue.Watermark)
+								// if the referred watermark is empty, skip
+								if referredWMB.Watermark != -1 {
+									p.offsetTimeline.PutReferred(referredWMB)
+									v.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", v.otWatcher.GetKVName()), zap.Int64("referredWatermark", referredWMB.Watermark), zap.Int64("referredOffset", referredWMB.Offset))
+									break
+								}
 							}
 						}
+						// if the break never happens, it's safe to do nothing ATM
+						// it could be the Vn-1's idle processor's speed is much slower
+						// than the others, or no data is flowing into this Vn's processor
 					}
-					// if the break never happens, it's safe to do nothing ATM
-					// it could be the Vn-1's idle processor's speed is much slower than the others
-					// or no data is flowing into this Vn's processor
 				} else {
 					// NOTE: currently, for source edges, the otValue.Idle is always false
-					p.offsetTimeline.Put(OffsetWatermark{
-						watermark: otValue.Watermark,
-						offset:    otValue.Offset,
-					})
+					p.offsetTimeline.Put(otValue)
 					v.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", v.otWatcher.GetKVName()), zap.Int64("watermark", otValue.Watermark), zap.Int64("offset", otValue.Offset))
 				}
 			case store.KVDelete:
diff --git a/pkg/watermark/fetch/source_fetcher.go b/pkg/watermark/fetch/source_fetcher.go
index 5a640cc8..a9160070 100644
--- a/pkg/watermark/fetch/source_fetcher.go
+++ b/pkg/watermark/fetch/source_fetcher.go
@@ -52,42 +52,48 @@ func NewSourceFetcher(ctx context.Context, sourceBufferName string, storeWatcher
 	}
 }
 
-// GetHeadWatermark returns the latest watermark of all the processors.
-func (e *sourceFetcher) GetHeadWatermark() wmb.Watermark {
-	var epoch int64 = math.MinInt64
+// GetWatermark returns the lowest of the latest Watermark of all the processors,
+// it ignores the input Offset.
+func (e *sourceFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
+	var epoch int64 = math.MaxInt64
 	for _, p := range e.processorManager.GetAllProcessors() {
 		if !p.IsActive() {
 			continue
 		}
-		if p.offsetTimeline.GetHeadWatermark() > epoch {
+		if p.offsetTimeline.GetHeadWatermark() < epoch {
 			epoch = p.offsetTimeline.GetHeadWatermark()
 		}
 	}
-	if epoch == math.MinInt64 {
-		// Use -1 as default watermark value to indicate there is no valid watermark yet.
-		return wmb.Watermark(time.UnixMilli(-1))
+	if epoch == math.MaxInt64 {
+		epoch = -1
 	}
 	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
-// GetWatermark returns the lowest of the latest watermark of all the processors,
-// it ignores the input offset.
-func (e *sourceFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
-	var epoch int64 = math.MaxInt64
+// GetHeadWatermark returns the latest watermark of all the processors.
+func (e *sourceFetcher) GetHeadWatermark() wmb.Watermark {
+	var epoch int64 = math.MinInt64
 	for _, p := range e.processorManager.GetAllProcessors() {
 		if !p.IsActive() {
 			continue
 		}
-		if p.offsetTimeline.GetHeadWatermark() < epoch {
+		if p.offsetTimeline.GetHeadWatermark() > epoch {
 			epoch = p.offsetTimeline.GetHeadWatermark()
 		}
 	}
-	if epoch == math.MaxInt64 {
-		epoch = -1
+	if epoch == math.MinInt64 {
+		// Use -1 as default watermark value to indicate there is no valid watermark yet.
+		return wmb.Watermark(time.UnixMilli(-1))
 	}
 	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
+// GetHeadWMB returns the latest idle WMB among all processors
+func (e *sourceFetcher) GetHeadWMB() wmb.WMB {
+	// TODO: what would this be...
+	return wmb.WMB{}
+}
+
 // Close function closes the watchers.
 func (e *sourceFetcher) Close() error {
 	e.log.Infof("Closing source watermark fetcher")
diff --git a/pkg/watermark/generic/noop.go b/pkg/watermark/generic/noop.go
index 572c2b42..636cc88b 100644
--- a/pkg/watermark/generic/noop.go
+++ b/pkg/watermark/generic/noop.go
@@ -47,7 +47,7 @@ func (n NoOpWMProgressor) PublishWatermark(_ wmb.Watermark, _ isb.Offset) {
 }
 
 // PublishIdleWatermark does a no-op idle watermark publish.
-func (n NoOpWMProgressor) PublishIdleWatermark(wmb.Watermark) {
+func (n NoOpWMProgressor) PublishIdleWatermark(wmb.Watermark, isb.Offset) {
 }
 
 // GetLatestWatermark returns the default watermark as the latest watermark.
@@ -60,6 +60,11 @@ func (n NoOpWMProgressor) GetHeadWatermark() wmb.Watermark {
 	return wmb.Watermark{}
 }
 
+// GetHeadWMB returns the default WMB.
+func (n NoOpWMProgressor) GetHeadWMB() wmb.WMB {
+	return wmb.WMB{}
+}
+
 // Close stops the no-op progressor.
 func (n NoOpWMProgressor) Close() error {
 	return nil
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index 7ff7c9bc..c568b09b 100644
--- a/pkg/watermark/publish/publisher.go
+++ b/pkg/watermark/publish/publisher.go
@@ -37,7 +37,7 @@ type Publisher interface {
 	// PublishWatermark publishes the watermark.
 	PublishWatermark(wmb.Watermark, isb.Offset)
 	// PublishIdleWatermark publishes the idle watermark.
-	PublishIdleWatermark(wm wmb.Watermark)
+	PublishIdleWatermark(wm wmb.Watermark, o isb.Offset)
 	// GetLatestWatermark returns the latest published watermark.
 	GetLatestWatermark() wmb.Watermark
 }
@@ -154,14 +154,23 @@ func (p *publish) validateWatermark(wm wmb.Watermark) (wmb.Watermark, bool) {
 }
 
 // PublishIdleWatermark publishes the idle watermark and will retry until it can succeed.
-func (p *publish) PublishIdleWatermark(wm wmb.Watermark) {
+// TODO: merge with PublishWatermark
+func (p *publish) PublishIdleWatermark(wm wmb.Watermark, offset isb.Offset) {
 	var key = p.entity.GetName()
 	validWM, skipWM := p.validateWatermark(wm)
 	if skipWM {
 		return
 	}
+	// build value
+	var seq int64
+	if p.opts.isSource || p.opts.isSink {
+		// for source and sink publisher, we don't care about the offset, also the sequence of the offset might not be integer.
+		seq = time.Now().UnixNano()
+	} else {
+		seq, _ = offset.Sequence()
+	}
 	var otValue = wmb.WMB{
-		Offset:    0,
+		Offset:    seq,
 		Watermark: validWM.UnixMilli(),
 		Idle:      true,
 	}
@@ -177,7 +186,7 @@ func (p *publish) PublishIdleWatermark(wm wmb.Watermark) {
 			// TODO: better exponential backoff
 			time.Sleep(time.Millisecond * 250)
 		} else {
-			p.log.Debugw("New idle watermark published", zap.String("HB", p.heartbeatStore.GetStoreName()), zap.String("OT", p.otStore.GetStoreName()), zap.String("key", key))
+			p.log.Debugw("New idle watermark published", zap.String("HB", p.heartbeatStore.GetStoreName()), zap.String("OT", p.otStore.GetStoreName()), zap.String("key", key), zap.Int64("offset", seq), zap.Int64("watermark", validWM.UnixMilli()))
 			break
 		}
 	}
diff --git a/pkg/watermark/wmb/wmb.go b/pkg/watermark/wmb/wmb.go
index f7e7d7bd..4c8a54c0 100644
--- a/pkg/watermark/wmb/wmb.go
+++ b/pkg/watermark/wmb/wmb.go
@@ -58,3 +58,57 @@ func DecodeToWMB(b []byte) (WMB, error) {
 	}
 	return v, nil
 }
+
+type WMBChecker struct {
+	counter int
+	max     int
+	w       WMB
+}
+
+// NewWMBChecker returns a WMBChecker to check if the wmb is idle.
+// If all the iterations get the same wmb, the wmb is considered as valid
+// and will be used to publish a wmb to pods of the next vertex.
+func NewWMBChecker(numOfIteration int) WMBChecker {
+	return WMBChecker{
+		counter: 0,
+		max:     numOfIteration,
+		w:       WMB{},
+	}
+}
+
+// ValidateHeadWMB checks if the head wmb is idle and is the same as the wmb from the previous iteration.
+// If all the iterations get the same wmb, returns true.
+func (c *WMBChecker) ValidateHeadWMB(w WMB) bool {
+	if !w.Idle {
+		// if wmb is not idle, skip and reset the counter
+		c.counter = 0
+		return false
+	}
+	// check the counter value
+	if c.counter == 0 {
+		c.counter++
+		// the wmb only writes once when counter is zero
+		c.w = w
+	} else if c.counter < c.max-1 {
+		c.counter++
+		if c.w == w {
+			// we get the same wmb, meaning the wmb is valid, continue
+		} else {
+			// else, start over
+			c.counter = 0
+		}
+	} else if c.counter >= c.max-1 {
+		c.counter = 0
+		if c.w == w {
+			// reach max iteration, if still get the same wmb,
+			// then the wmb is considered as valid, return ture
+			return true
+		}
+	}
+	return false
+}
+
+// GetCounter gets the current counter value for the WMBChecker, it's used in log and tests
+func (c *WMBChecker) GetCounter() int {
+	return c.counter
+}


Test patch:
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index 8e12a47e..bad829c8 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -51,28 +51,34 @@ const (
 var (
 	testStartTime       = time.Unix(1636470000, 0).UTC()
 	testSourceWatermark = time.Unix(1636460000, 0).UTC()
+	testWMBWatermark    = time.Unix(1636470000, 0).UTC()
 )
 
 type testForwardFetcher struct {
 	// for forward_test.go only
 }
 
-func (t testForwardFetcher) Close() error {
+func (t *testForwardFetcher) Close() error {
 	// won't be used
 	return nil
 }
 
 // GetWatermark uses current time as the watermark because we want to make sure
 // the test publisher is publishing watermark
-func (t testForwardFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
+func (t *testForwardFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
 	return wmb.Watermark(testSourceWatermark)
 }
 
-func (t testForwardFetcher) GetHeadWatermark() wmb.Watermark {
+func (t *testForwardFetcher) GetHeadWatermark() wmb.Watermark {
 	// won't be used
 	return wmb.Watermark{}
 }
 
+func (t *testForwardFetcher) GetHeadWMB() wmb.WMB {
+	// won't be used
+	return wmb.WMB{}
+}
+
 type myForwardTest struct {
 }
 
@@ -135,6 +141,382 @@ func TestNewInterStepDataForward(t *testing.T) {
 	<-stopped
 }
 
+type testWMBFetcher struct {
+	// for forward_test.go only
+	WMBTestSameHeadWMB bool // for testing same head wmb, if set true then WMBTestDiffHeadWMB must be false
+	sameCounter        int
+	sameLock           sync.RWMutex
+	WMBTestDiffHeadWMB bool // for testing different head wmb, if set true then WMBTestSameHeadWMB must be false
+	diffCounter        int
+	diffLock           sync.RWMutex
+}
+
+// RevertBoolValue set WMBTestSameHeadWMB and WMBTestDiffHeadWMB to opposite value
+func (t *testWMBFetcher) RevertBoolValue() {
+	t.sameLock.Lock()
+	defer t.sameLock.Unlock()
+	t.diffLock.Lock()
+	defer t.diffLock.Unlock()
+	t.WMBTestSameHeadWMB = !t.WMBTestSameHeadWMB
+	t.WMBTestDiffHeadWMB = !t.WMBTestDiffHeadWMB
+}
+
+func (t *testWMBFetcher) Close() error {
+	// won't be used
+	return nil
+}
+
+// GetWatermark uses current time as the watermark because we want to make sure
+// the test publisher is publishing watermark
+func (t *testWMBFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
+	return wmb.Watermark(testWMBWatermark)
+}
+
+func (t *testWMBFetcher) GetHeadWatermark() wmb.Watermark {
+	// won't be used
+	return wmb.Watermark{}
+}
+
+func (t *testWMBFetcher) GetHeadWMB() wmb.WMB {
+	t.sameLock.RLock()
+	defer t.sameLock.RUnlock()
+	t.diffLock.RLock()
+	defer t.diffLock.RUnlock()
+	if t.WMBTestSameHeadWMB {
+		t.sameCounter++
+		if t.sameCounter == 1 || t.sameCounter == 2 {
+			return wmb.WMB{
+				Idle:      true,
+				Offset:    100,
+				Watermark: 1636440000000,
+			}
+		}
+		if t.sameCounter == 3 || t.sameCounter == 4 {
+			return wmb.WMB{
+				Idle:      true,
+				Offset:    102,
+				Watermark: 1636480000000,
+			}
+		}
+	}
+	if t.WMBTestDiffHeadWMB {
+		t.diffCounter++
+		if t.diffCounter == 1 {
+			return wmb.WMB{
+				Idle:      true,
+				Offset:    100,
+				Watermark: 1636440000000,
+			}
+		}
+		if t.diffCounter == 2 {
+			return wmb.WMB{
+				Idle:      false,
+				Offset:    101,
+				Watermark: 1636450000000,
+			}
+		}
+	}
+	// won't be used
+	return wmb.WMB{}
+}
+
+func TestNewInterStepDataForwardIdleWatermark(t *testing.T) {
+	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, simplebuffer.WithReadTimeOut(time.Second)) // default read timeout is 1s
+	to1 := simplebuffer.NewInMemoryBuffer("to1", 10)
+	toSteps := map[string]isb.BufferWriter{
+		"to1": to1,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
+		PipelineName: "testPipeline",
+		AbstractVertex: dfv1.AbstractVertex{
+			Name: "testVertex",
+		},
+	}}
+
+	ctrlMessage := []isb.Message{{Header: isb.Header{Kind: isb.WMB}}}
+	writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime)
+
+	fetchWatermark := &testWMBFetcher{WMBTestSameHeadWMB: true}
+	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
+	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, myForwardTest{}, myForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(2), WithVertexType(dfv1.VertexTypeMapUDF))
+	assert.NoError(t, err)
+	assert.False(t, to1.IsFull())
+	assert.True(t, to1.IsEmpty())
+
+	stopped := f.Start()
+	// first batch: read message size is 1 with one ctrl message
+	// the ctrl message should be acked
+	// no message published to the next vertex
+	// so the timeline should be empty
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for fromStep.IsEmpty() {
+			select {
+			case <-ctx.Done():
+				if ctx.Err() == context.DeadlineExceeded {
+					err = ctx.Err()
+				}
+			default:
+				time.Sleep(1 * time.Millisecond)
+			}
+		}
+	}()
+	_, errs := fromStep.Write(ctx, ctrlMessage)
+	assert.Equal(t, make([]error, 1), errs)
+	wg.Wait()
+	if err != nil {
+		t.Fatal("expected the buffer not to be empty", err)
+	}
+	for !fromStep.IsEmpty() {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatal("expected the buffer to be empty", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+		}
+	}
+	otNil, _ := otStores["to1"].GetAllKeys(ctx)
+	assert.Nil(t, otNil)
+
+	// 2nd and 3rd batches: read message size is 0
+	// should send idle watermark
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
+		for otKeys1 == nil {
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			time.Sleep(time.Millisecond * 100)
+		}
+	}()
+	wg.Wait()
+
+	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
+	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
+	otDecode1, _ := wmb.DecodeToWMB(otValue1)
+	for otDecode1.Offset != 0 { // the first ctrl message written to isb. can't use idle because default idle=false
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatal("expected to have idle watermark in to1 timeline", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+			otDecode1, _ = wmb.DecodeToWMB(otValue1)
+		}
+	}
+	assert.Equal(t, wmb.WMB{
+		Idle:      true,
+		Offset:    0, // the first ctrl message written to isb
+		Watermark: 1636440000000,
+	}, otDecode1)
+	assert.Equal(t, isb.WMB, to1.GetMessages(1)[0].Kind)
+
+	// 4th batch: read message size = 1
+	// a new active watermark should be inserted
+	_, errs = fromStep.Write(ctx, writeMessages[:2])
+	assert.Equal(t, make([]error, 2), errs)
+
+	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+	otDecode1, _ = wmb.DecodeToWMB(otValue1)
+	for otDecode1.Idle {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatal("expected to have active watermark in to1 timeline", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+			otDecode1, _ = wmb.DecodeToWMB(otValue1)
+		}
+	}
+	assert.Equal(t, wmb.WMB{
+		Idle:      false,
+		Offset:    2, // the second message written to isb, read batch size is 2 so the offset is 0+2=2
+		Watermark: testWMBWatermark.UnixMilli(),
+	}, otDecode1)
+
+	// 5th & 6th batch: again idling but got diff head WMB
+	// so the head is still the same active watermark
+	// and no new ctrl message to the next vertex
+	f.fetchWatermark.(*testWMBFetcher).RevertBoolValue()
+	time.Sleep(2 * time.Second) // default read timeout is 1s
+	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+	otDecode1, _ = wmb.DecodeToWMB(otValue1)
+	assert.Equal(t, wmb.WMB{
+		Idle:      false,
+		Offset:    2, // the second message written to isb, read batch size is 2 so the offset is 0+2=2
+		Watermark: testWMBWatermark.UnixMilli(),
+	}, otDecode1)
+
+	var wantKind = []isb.MessageKind{
+		isb.WMB,
+		isb.Data,
+		isb.Data,
+		isb.Data, // this is data because it's the default value
+	}
+	var wantBody = []bool{
+		false,
+		true,
+		true,
+		false,
+	}
+	for idx, msg := range to1.GetMessages(4) {
+		assert.Equal(t, wantKind[idx], msg.Kind)
+		assert.Equal(t, wantBody[idx], len(msg.Body.Payload) > 0)
+	}
+
+	// stop will cancel the contexts and therefore the forwarder stops without waiting
+	f.Stop()
+
+	<-stopped
+}
+
+func TestNewInterStepDataForwardIdleWatermark_Reset(t *testing.T) {
+	fromStep := simplebuffer.NewInMemoryBuffer("from", 25, simplebuffer.WithReadTimeOut(time.Second)) // default read timeout is 1s
+	to1 := simplebuffer.NewInMemoryBuffer("to1", 10)
+	toSteps := map[string]isb.BufferWriter{
+		"to1": to1,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	vertex := &dfv1.Vertex{Spec: dfv1.VertexSpec{
+		PipelineName: "testPipeline",
+		AbstractVertex: dfv1.AbstractVertex{
+			Name: "testVertex",
+		},
+	}}
+
+	writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime)
+
+	fetchWatermark := &testWMBFetcher{WMBTestSameHeadWMB: true}
+	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
+	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, myForwardTest{}, myForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(2), WithVertexType(dfv1.VertexTypeMapUDF))
+	assert.NoError(t, err)
+	assert.False(t, to1.IsFull())
+	assert.True(t, to1.IsEmpty())
+
+	stopped := f.Start()
+	var wg sync.WaitGroup
+
+	// 1st and 2nd batches: read message size is 0
+	// should send idle watermark
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
+		for otKeys1 == nil {
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			time.Sleep(time.Millisecond * 100)
+		}
+	}()
+	wg.Wait()
+
+	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
+	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
+	otDecode1, _ := wmb.DecodeToWMB(otValue1)
+	for otDecode1.Offset != 0 { // the first ctrl message written to isb. can't use idle because default idle=false
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatal("expected to have idle watermark in to1 timeline", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+			otDecode1, _ = wmb.DecodeToWMB(otValue1)
+		}
+	}
+	assert.Equal(t, wmb.WMB{
+		Idle:      true,
+		Offset:    0, // the first ctrl message written to isb
+		Watermark: 1636440000000,
+	}, otDecode1)
+	assert.Equal(t, isb.WMB, to1.GetMessages(1)[0].Kind)
+
+	// 3rd batch: read message size = 1
+	// a new active watermark should be inserted
+	_, errs := fromStep.Write(ctx, writeMessages[:2])
+	assert.Equal(t, make([]error, 2), errs)
+
+	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+	otDecode1, _ = wmb.DecodeToWMB(otValue1)
+	for otDecode1.Idle {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatal("expected to have active watermark in to1 timeline", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+			otDecode1, _ = wmb.DecodeToWMB(otValue1)
+		}
+	}
+	assert.Equal(t, wmb.WMB{
+		Idle:      false,
+		Offset:    2, // the second message written to isb, read batch size is 2 so the offset is 0+2=2
+		Watermark: testWMBWatermark.UnixMilli(),
+	}, otDecode1)
+
+	// 5th & 6th batch: again idling should send a new ctrl message
+	otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+	otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+	otDecode1, _ = wmb.DecodeToWMB(otValue1)
+	for otDecode1.Offset != 3 { // the second ctrl message written to isb. can't use idle because default idle=false
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatal("expected to have idle watermark in to1 timeline", ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			otKeys1, _ = otStores["to1"].GetAllKeys(ctx)
+			otValue1, _ = otStores["to1"].GetValue(ctx, otKeys1[0])
+			otDecode1, _ = wmb.DecodeToWMB(otValue1)
+		}
+	}
+
+	var wantKind = []isb.MessageKind{
+		isb.WMB,
+		isb.Data,
+		isb.Data,
+		isb.WMB,
+	}
+	var wantBody = []bool{
+		false,
+		true,
+		true,
+		false,
+	}
+	for idx, msg := range to1.GetMessages(4) {
+		assert.Equal(t, wantKind[idx], msg.Kind)
+		assert.Equal(t, wantBody[idx], len(msg.Body.Payload) > 0)
+	}
+
+	// stop will cancel the contexts and therefore the forwarder stops without waiting
+	f.Stop()
+
+	<-stopped
+}
+
 // mySourceForwardTest tests source data transformer by updating message event time, and then verifying new event time and IsLate assignments.
 type mySourceForwardTest struct {
 }
@@ -200,7 +582,7 @@ func TestSourceInterStepDataForward(t *testing.T) {
 	defer cancel()
 
 	writeMessages := testutils.BuildTestWriteMessages(int64(20), testStartTime)
-	fetchWatermark := testForwardFetcher{}
+	fetchWatermark := &testForwardFetcher{}
 	_, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
 
 	// verify if source watermark publisher is not set, NewInterStepDataForward throws.
@@ -313,7 +695,7 @@ func TestNewInterStepDataForwardToOneStep(t *testing.T) {
 		},
 	}}
 
-	fetchWatermark := testForwardFetcher{}
+	fetchWatermark := &testForwardFetcher{}
 	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
 	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, myForwardTest{}, myForwardTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(2), WithVertexType(dfv1.VertexTypeMapUDF))
 	assert.NoError(t, err)
@@ -409,7 +791,7 @@ func TestNewInterStepDataForward_dropAll(t *testing.T) {
 		},
 	}}
 
-	fetchWatermark := testForwardFetcher{}
+	fetchWatermark := &testForwardFetcher{}
 	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
 	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, myForwardDropTest{}, myForwardDropTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(2), WithVertexType(dfv1.VertexTypeMapUDF))
 	assert.NoError(t, err)
@@ -501,7 +883,7 @@ func TestNewInterStepData_forwardToAll(t *testing.T) {
 			Name: "testVertex",
 		},
 	}}
-	fetchWatermark := testForwardFetcher{}
+	fetchWatermark := &testForwardFetcher{}
 	publishWatermark, otStores := buildPublisherMapAndOTStore(toSteps)
 	f, err := NewInterStepDataForward(vertex, fromStep, toSteps, myForwardToAllTest{}, myForwardToAllTest{}, fetchWatermark, publishWatermark, WithReadBatchSize(2), WithVertexType(dfv1.VertexTypeMapUDF))
 	assert.NoError(t, err)
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index 5df8de53..c2705246 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -78,7 +78,7 @@ func (e *EventTypeWMProgressor) PublishWatermark(watermark wmb.Watermark, offset
 	e.watermarks[offset.String()] = watermark
 }
 
-func (e *EventTypeWMProgressor) PublishIdleWatermark(wmb.Watermark) {
+func (e *EventTypeWMProgressor) PublishIdleWatermark(wmb.Watermark, isb.Offset) {
 	// TODO
 }
 
@@ -100,6 +100,10 @@ func (e *EventTypeWMProgressor) GetHeadWatermark() wmb.Watermark {
 	return wmb.Watermark{}
 }
 
+func (e *EventTypeWMProgressor) GetHeadWMB() wmb.WMB {
+	return wmb.WMB{}
+}
+
 // PayloadForTest is a dummy payload for testing.
 type PayloadForTest struct {
 	Key   string
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 69465abf..84899824 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -226,7 +226,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 					Idle:      false,
 				},
 				"buffer2": {
-					Offset:    0,
+					Offset:    -1,
 					Watermark: int64(119999),
 					Idle:      true,
 				},
@@ -269,12 +269,12 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			expected: []bool{true, true},
 			wmExpected: map[string]wmb.WMB{
 				"buffer1": {
-					Offset:    0,
+					Offset:    -1,
 					Watermark: int64(119999),
 					Idle:      true,
 				},
 				"buffer2": {
-					Offset:    0,
+					Offset:    -1,
 					Watermark: int64(119999),
 					Idle:      true,
 				},
diff --git a/pkg/watermark/fetch/edge_fetcher_test.go b/pkg/watermark/fetch/edge_fetcher_test.go
index 6eae0809..9ce3df13 100644
--- a/pkg/watermark/fetch/edge_fetcher_test.go
+++ b/pkg/watermark/fetch/edge_fetcher_test.go
@@ -40,29 +40,30 @@ func TestBuffer_GetWatermark(t *testing.T) {
 	// so use no op watcher for testing
 	hbWatcher := noop.NewKVOpWatch()
 	otWatcher := noop.NewKVOpWatch()
-	processorManager := NewProcessorManager(ctx, store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher))
+	storeWatcher := store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher)
+	processorManager := NewProcessorManager(ctx, storeWatcher)
 	var (
 		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5)
 		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5)
 		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5)
-		pod0Timeline = []OffsetWatermark{
-			{watermark: 11, offset: 9},
-			{watermark: 12, offset: 20},
-			{watermark: 13, offset: 21},
-			{watermark: 14, offset: 22},
-			{watermark: 17, offset: 28},
+		pod0Timeline = []wmb.WMB{
+			{Watermark: 11, Offset: 9},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 13, Offset: 21},
+			{Watermark: 14, Offset: 22},
+			{Watermark: 17, Offset: 28},
 		}
-		pod1Timeline = []OffsetWatermark{
-			{watermark: 8, offset: 13},
-			{watermark: 9, offset: 16},
-			{watermark: 10, offset: 18},
-			{watermark: 17, offset: 26},
+		pod1Timeline = []wmb.WMB{
+			{Watermark: 8, Offset: 13},
+			{Watermark: 9, Offset: 16},
+			{Watermark: 10, Offset: 18},
+			{Watermark: 17, Offset: 26},
 		}
-		pod2Timeline = []OffsetWatermark{
-			{watermark: 10, offset: 14},
-			{watermark: 12, offset: 17},
-			{watermark: 14, offset: 19},
-			{watermark: 17, offset: 24},
+		pod2Timeline = []wmb.WMB{
+			{Watermark: 10, Offset: 14},
+			{Watermark: 12, Offset: 17},
+			{Watermark: 14, Offset: 19},
+			{Watermark: 17, Offset: 24},
 		}
 	)
 
@@ -143,8 +144,208 @@ func TestBuffer_GetWatermark(t *testing.T) {
 			if got := b.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(tt.args.offset, 10) })); time.Time(got).In(location) != time.UnixMilli(tt.want).In(location) {
 				t.Errorf("GetWatermark() = %v, want %v", got, wmb.Watermark(time.UnixMilli(tt.want)))
 			}
-			// this will always be 14 because the timeline has been populated ahead of time
-			assert.Equal(t, time.Time(b.GetHeadWatermark()).In(location), time.UnixMilli(14).In(location))
+			// this will always be 17 because the timeline has been populated ahead of time
+			// GetHeadWatermark is only used in UI and test
+			assert.Equal(t, time.Time(b.GetHeadWatermark()).In(location), time.UnixMilli(17).In(location))
 		})
 	}
 }
+
+func Test_edgeFetcher_GetHeadWMB(t *testing.T) {
+	var (
+		ctx               = context.Background()
+		bufferName        = "testBuffer"
+		hbWatcher         = noop.NewKVOpWatch()
+		otWatcher         = noop.NewKVOpWatch()
+		storeWatcher      = store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher)
+		processorManager1 = NewProcessorManager(ctx, storeWatcher)
+		processorManager2 = NewProcessorManager(ctx, storeWatcher)
+		processorManager3 = NewProcessorManager(ctx, storeWatcher)
+		processorManager4 = NewProcessorManager(ctx, storeWatcher)
+	)
+
+	getHeadWMBTest1(ctx, processorManager1)
+	getHeadWMBTest2(ctx, processorManager2)
+	getHeadWMBTest3(ctx, processorManager3)
+	getHeadWMBTest4(ctx, processorManager4)
+
+	tests := []struct {
+		name             string
+		processorManager *ProcessorManager
+		want             wmb.WMB
+	}{
+		{
+			name:             "all pods idle and get an idle WMB",
+			processorManager: processorManager1,
+			want: wmb.WMB{
+				Idle:      true,
+				Offset:    28,
+				Watermark: 17,
+			},
+		},
+		{
+			name:             "some pods idle and skip an idle WMB",
+			processorManager: processorManager2,
+			want:             wmb.WMB{},
+		},
+		{
+			name:             "all pods not idle and skip an idle WMB",
+			processorManager: processorManager3,
+			want:             wmb.WMB{},
+		},
+		{
+			name:             "all pods empty timeline",
+			processorManager: processorManager3,
+			want:             wmb.WMB{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := &edgeFetcher{
+				ctx:              ctx,
+				bufferName:       bufferName,
+				storeWatcher:     storeWatcher,
+				processorManager: tt.processorManager,
+				log:              zaptest.NewLogger(t).Sugar(),
+			}
+			assert.Equalf(t, tt.want, e.GetHeadWMB(), "GetHeadWMB()")
+		})
+	}
+}
+
+func getHeadWMBTest1(ctx context.Context, processorManager1 *ProcessorManager) {
+	var (
+		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5)
+		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5)
+		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5)
+		pod0Timeline = []wmb.WMB{
+			{
+				Idle:      true,
+				Offset:    28,
+				Watermark: 17,
+			},
+		}
+		pod1Timeline = []wmb.WMB{
+			{
+				Idle:      true,
+				Offset:    26,
+				Watermark: 17,
+			},
+		}
+		pod2Timeline = []wmb.WMB{
+			{
+				Idle:      true,
+				Offset:    24,
+				Watermark: 17,
+			},
+		}
+	)
+
+	for _, watermark := range pod0Timeline {
+		testPod0.offsetTimeline.Put(watermark)
+	}
+	for _, watermark := range pod1Timeline {
+		testPod1.offsetTimeline.Put(watermark)
+	}
+	for _, watermark := range pod2Timeline {
+		testPod2.offsetTimeline.Put(watermark)
+	}
+	processorManager1.addProcessor("testPod0", testPod0)
+	processorManager1.addProcessor("testPod1", testPod1)
+	processorManager1.addProcessor("testPod2", testPod2)
+}
+
+func getHeadWMBTest2(ctx context.Context, processorManager1 *ProcessorManager) {
+	var (
+		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5)
+		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5)
+		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5)
+		pod0Timeline = []wmb.WMB{
+			{
+				Idle:      false,
+				Offset:    28,
+				Watermark: 17,
+			},
+		}
+		pod1Timeline = []wmb.WMB{
+			{
+				Idle:      true,
+				Offset:    26,
+				Watermark: 17,
+			},
+		}
+		pod2Timeline = []wmb.WMB{
+			{
+				Idle:      true,
+				Offset:    24,
+				Watermark: 17,
+			},
+		}
+	)
+
+	for _, watermark := range pod0Timeline {
+		testPod0.offsetTimeline.Put(watermark)
+	}
+	for _, watermark := range pod1Timeline {
+		testPod1.offsetTimeline.Put(watermark)
+	}
+	for _, watermark := range pod2Timeline {
+		testPod2.offsetTimeline.Put(watermark)
+	}
+	processorManager1.addProcessor("testPod0", testPod0)
+	processorManager1.addProcessor("testPod1", testPod1)
+	processorManager1.addProcessor("testPod2", testPod2)
+}
+
+func getHeadWMBTest3(ctx context.Context, processorManager1 *ProcessorManager) {
+	var (
+		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5)
+		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5)
+		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5)
+		pod0Timeline = []wmb.WMB{
+			{
+				Idle:      false,
+				Offset:    28,
+				Watermark: 17,
+			},
+		}
+		pod1Timeline = []wmb.WMB{
+			{
+				Idle:      false,
+				Offset:    26,
+				Watermark: 17,
+			},
+		}
+		pod2Timeline = []wmb.WMB{
+			{
+				Idle:      false,
+				Offset:    24,
+				Watermark: 17,
+			},
+		}
+	)
+
+	for _, watermark := range pod0Timeline {
+		testPod0.offsetTimeline.Put(watermark)
+	}
+	for _, watermark := range pod1Timeline {
+		testPod1.offsetTimeline.Put(watermark)
+	}
+	for _, watermark := range pod2Timeline {
+		testPod2.offsetTimeline.Put(watermark)
+	}
+	processorManager1.addProcessor("testPod0", testPod0)
+	processorManager1.addProcessor("testPod1", testPod1)
+	processorManager1.addProcessor("testPod2", testPod2)
+}
+
+func getHeadWMBTest4(ctx context.Context, processorManager1 *ProcessorManager) {
+	var (
+		testPod0 = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5)
+		testPod1 = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5)
+		testPod2 = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5)
+	)
+	processorManager1.addProcessor("testPod0", testPod0)
+	processorManager1.addProcessor("testPod1", testPod1)
+	processorManager1.addProcessor("testPod2", testPod2)
+}
diff --git a/pkg/watermark/fetch/offset_timeline_test.go b/pkg/watermark/fetch/offset_timeline_test.go
index 5549188e..47da223f 100644
--- a/pkg/watermark/fetch/offset_timeline_test.go
+++ b/pkg/watermark/fetch/offset_timeline_test.go
@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"github.com/numaproj/numaflow/pkg/isb"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -31,17 +32,17 @@ func TestTimeline_GetEventTime(t *testing.T) {
 		ctx            = context.Background()
 		emptyTimeline  = NewOffsetTimeline(ctx, 5)
 		testTimeline   = NewOffsetTimeline(ctx, 10)
-		testwatermarks = []OffsetWatermark{
-			{watermark: 10, offset: 9},
-			{watermark: 12, offset: 10},
-			{watermark: 12, offset: 20},
-			{watermark: 13, offset: 21},
-			{watermark: 15, offset: 24},
-			{watermark: 20, offset: 26},
-			{watermark: 23, offset: 27},
-			{watermark: 28, offset: 30},
-			{watermark: 29, offset: 35},
-			{watermark: 32, offset: 36},
+		testwatermarks = []wmb.WMB{
+			{Watermark: 10, Offset: 9},
+			{Watermark: 12, Offset: 10},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 13, Offset: 21},
+			{Watermark: 15, Offset: 24},
+			{Watermark: 20, Offset: 26},
+			{Watermark: 23, Offset: 27},
+			{Watermark: 28, Offset: 30},
+			{Watermark: 29, Offset: 35},
+			{Watermark: 32, Offset: 36},
 		}
 	)
 
@@ -128,16 +129,16 @@ func TestOffsetTimeline_GetOffset(t *testing.T) {
 	var (
 		ctx            = context.Background()
 		testTimeline   = NewOffsetTimeline(ctx, 10)
-		testwatermarks = []OffsetWatermark{
-			{watermark: 10, offset: 9},
-			{watermark: 12, offset: 20},
-			{watermark: 13, offset: 21},
-			{watermark: 15, offset: 24},
-			{watermark: 20, offset: 26},
-			{watermark: 23, offset: 27},
-			{watermark: 28, offset: 30},
-			{watermark: 29, offset: 35},
-			{watermark: 32, offset: 36},
+		testwatermarks = []wmb.WMB{
+			{Watermark: 10, Offset: 9},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 13, Offset: 21},
+			{Watermark: 15, Offset: 24},
+			{Watermark: 20, Offset: 26},
+			{Watermark: 23, Offset: 27},
+			{Watermark: 28, Offset: 30},
+			{Watermark: 29, Offset: 35},
+			{Watermark: 32, Offset: 36},
 		}
 	)
 
@@ -220,84 +221,133 @@ func TestOffsetTimeline_GetOffset(t *testing.T) {
 	}
 }
 
-func TestOffsetTimeline(t *testing.T) {
+func TestOffsetTimeline_PutReferred(t *testing.T) {
 	var (
 		ctx            = context.Background()
 		testTimeline   = NewOffsetTimeline(ctx, 10)
-		testwatermarks = []OffsetWatermark{
-			{watermark: 10, offset: 9},
-			{watermark: 12, offset: 20},
-			{watermark: 13, offset: 21},
-			{watermark: 15, offset: 24},
-			{watermark: 15, offset: 25}, // will overwrite the previous one
-			{watermark: 20, offset: 26},
-			{watermark: 23, offset: 27},
-			{watermark: 28, offset: 30},
-			{watermark: 29, offset: 35},
-			{watermark: 32, offset: 36},
+		testwatermarks = []wmb.WMB{
+			{Watermark: 10, Offset: 9},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 13, Offset: 21},
+			{Watermark: 15, Offset: 24},
+			{Watermark: 15, Offset: 25}, // will overwrite the previous one
+			{Watermark: 20, Offset: 26},
+			{Watermark: 23, Offset: 27},
+			{Watermark: 28, Offset: 30},
+			{Watermark: 29, Offset: 35},
+			{Watermark: 32, Offset: 36},
 		}
 	)
 
 	for _, watermark := range testwatermarks {
 		testTimeline.Put(watermark)
-		assert.Equal(t, watermark, testTimeline.GetHeadOffsetWatermark())
-		assert.Equal(t, watermark.watermark, testTimeline.GetHeadWatermark(), watermark.watermark)
-		assert.Equal(t, watermark.offset, testTimeline.GetHeadOffset())
+		assert.Equal(t, watermark, testTimeline.GetHeadWMB())
+		assert.Equal(t, watermark.Watermark, testTimeline.GetHeadWatermark(), watermark.Watermark)
+		assert.Equal(t, watermark.Offset, testTimeline.GetHeadOffset())
 	}
 	assert.Equal(t, "[32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20] -> [10:9] -> [-1:-1]", testTimeline.Dump())
 
-	testTimeline.Put(OffsetWatermark{watermark: 33, offset: 36})
+	testTimeline.Put(wmb.WMB{Watermark: 33, Offset: 36})
 	assert.Equal(t, "[33:36] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20] -> [10:9]", testTimeline.Dump())
 
-	testTimeline.Put(OffsetWatermark{watermark: 33, offset: 35})
+	testTimeline.Put(wmb.WMB{Watermark: 33, Offset: 35})
 	// should be ignored
 	assert.Equal(t, "[33:36] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20] -> [10:9]", testTimeline.Dump())
 
-	testTimeline.Put(OffsetWatermark{watermark: 30, offset: 33})
+	testTimeline.Put(wmb.WMB{Watermark: 30, Offset: 33})
 	// should be ignored
 	assert.Equal(t, "[33:36] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20] -> [10:9]", testTimeline.Dump())
 
-	testTimeline.Put(OffsetWatermark{watermark: 30, offset: 35})
+	testTimeline.Put(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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
+	testTimeline.PutReferred(wmb.WMB{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())
 
 }
 
+func TestOffsetTimeline_PutIdle(t *testing.T) {
+	var (
+		ctx          = context.Background()
+		testTimeline = NewOffsetTimeline(ctx, 10)
+		setUps       = []wmb.WMB{
+			{Idle: false, Watermark: 10, Offset: 9},
+			{Idle: false, Watermark: 12, Offset: 20},
+			{Idle: false, Watermark: 13, Offset: 21},
+			{Idle: false, Watermark: 15, Offset: 24},
+			{Idle: false, Watermark: 15, Offset: 25}, // will overwrite the previous one
+			{Idle: false, Watermark: 20, Offset: 26},
+			{Idle: false, Watermark: 23, Offset: 27},
+			{Idle: false, Watermark: 28, Offset: 30},
+			{Idle: false, Watermark: 29, Offset: 35},
+			{Idle: false, Watermark: 32, Offset: 36},
+		}
+	)
+
+	for _, watermark := range setUps {
+		testTimeline.Put(watermark)
+		assert.Equal(t, watermark, testTimeline.GetHeadWMB())
+		assert.Equal(t, watermark.Watermark, testTimeline.GetHeadWatermark(), watermark.Watermark)
+		assert.Equal(t, watermark.Offset, testTimeline.GetHeadOffset())
+	}
+	assert.Equal(t, "[32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20] -> [10:9] -> [-1:-1]", testTimeline.Dump())
+
+	testTimeline.PutIdle(wmb.WMB{Idle: true, Watermark: 33, Offset: 37}) // insert
+	assert.Equal(t, "[IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20] -> [10:9]", testTimeline.Dump())
+
+	testTimeline.Put(wmb.WMB{Idle: false, Watermark: 34, Offset: 38}) // insert
+	assert.Equal(t, "[34:38] -> [IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(wmb.WMB{Idle: true, Watermark: 34, Offset: 39}) // same watermark, update
+	assert.Equal(t, "[IDLE 34:39] -> [IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(wmb.WMB{Idle: true, Watermark: 36, Offset: 39}) // same offset, update
+	assert.Equal(t, "[IDLE 36:39] -> [IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(wmb.WMB{Idle: true, Watermark: 36, Offset: 38}) // smaller offset, ignore
+	assert.Equal(t, "[IDLE 36:39] -> [IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(wmb.WMB{Idle: true, Watermark: 35, Offset: 39}) // smaller watermark, ignore
+	assert.Equal(t, "[IDLE 36:39] -> [IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21] -> [12:20]", testTimeline.Dump())
+
+	testTimeline.PutIdle(wmb.WMB{Idle: true, Watermark: 37, Offset: 40}) // larger offset, insert
+	assert.Equal(t, "[IDLE 37:40] -> [IDLE 36:39] -> [IDLE 33:37] -> [32:36] -> [29:35] -> [28:30] -> [23:27] -> [20:26] -> [15:25] -> [13:21]", testTimeline.Dump())
+
+}
+
 func TestOffsetTimeline_GetReferredWatermark(t *testing.T) {
 	var (
 		ctx            = context.Background()
 		testTimeline   = NewOffsetTimeline(ctx, 10)
-		testWatermarks = [10]OffsetWatermark{
-			{watermark: 10, offset: 9},
-			{watermark: 12, offset: 20},
-			{watermark: 13, offset: 21},
-			{watermark: 15, offset: 24},
-			{watermark: 15, offset: 25}, // will overwrite the previous one
-			{watermark: 20, offset: 26},
-			{watermark: 23, offset: 27},
-			{watermark: 28, offset: 30},
-			{watermark: 29, offset: 35},
-			{watermark: 32, offset: 36},
+		testWatermarks = [10]wmb.WMB{
+			{Watermark: 10, Offset: 9},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 13, Offset: 21},
+			{Watermark: 15, Offset: 24},
+			{Watermark: 15, Offset: 25}, // will overwrite the previous one
+			{Watermark: 20, Offset: 26},
+			{Watermark: 23, Offset: 27},
+			{Watermark: 28, Offset: 30},
+			{Watermark: 29, Offset: 35},
+			{Watermark: 32, Offset: 36},
 		}
 		referredEventTimes = [10]int64{
 			12,
@@ -311,17 +361,17 @@ func TestOffsetTimeline_GetReferredWatermark(t *testing.T) {
 			30,
 			32,
 		}
-		expectedReferredWatermarks = [10]OffsetWatermark{
-			{watermark: 10, offset: 9},
-			{watermark: 12, offset: 20},
-			{watermark: 12, offset: 20},
-			{watermark: 13, offset: 21},
-			{watermark: 15, offset: 25},
-			{watermark: 15, offset: 25},
-			{watermark: 20, offset: 26},
-			{watermark: 28, offset: 30},
-			{watermark: 29, offset: 35},
-			{watermark: 32, offset: 36},
+		expectedReferredWatermarks = [10]wmb.WMB{
+			{Watermark: 10, Offset: 9},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 12, Offset: 20},
+			{Watermark: 13, Offset: 21},
+			{Watermark: 15, Offset: 25},
+			{Watermark: 15, Offset: 25},
+			{Watermark: 20, Offset: 26},
+			{Watermark: 28, Offset: 30},
+			{Watermark: 29, Offset: 35},
+			{Watermark: 32, Offset: 36},
 		}
 	)
 
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index b4ed3ac3..64330822 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -311,8 +311,8 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		}
 	}
 
-	// publish an idle watermark
-	otValueByte, err = otValueToBytes(0, epoch+600, true)
+	// publish an idle watermark: simulate reduce
+	otValueByte, err = otValueToBytes(-1, epoch+600, true)
 	assert.NoError(t, err)
 	err = otStore.PutKV(ctx, "p1", otValueByte)
 	assert.NoError(t, err)
@@ -330,6 +330,25 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		}
 	}
 
+	// publish an idle watermark: simulate map
+	otValueByte, err = otValueToBytes(106, epoch+600, true)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
+	assert.NoError(t, err)
+
+	// p1 should get the head offset watermark from p2
+	for allProcessors["p1"].offsetTimeline.Dump() != "[IDLE 1651161660600:106] -> [1651161660500:105] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1]" {
+		select {
+		case <-ctx.Done():
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Fatalf("[IDLE 1651161660600:106] -> [1651161660500:105] -> [-1:-1] -> ..., got %s: %s", allProcessors["p1"].offsetTimeline.Dump(), ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+
 	wg.Wait()
 	cancel()
 }
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index e945c7ce..fe7ab258 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -98,7 +98,8 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	assert.Equal(t, wmb.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), head.String())
 
 	// simulate the next read batch where the watermark is now $epoch
-	p.PublishIdleWatermark(wmb.Watermark(time.UnixMilli(epoch).In(location)))
+	// set offset to -1 to simulate reduce vertex publish idle watermark
+	p.PublishIdleWatermark(wmb.Watermark(time.UnixMilli(epoch).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(-1) }))
 	keys, err = p.otStore.GetAllKeys(p.ctx)
 	assert.NoError(t, err)
 	assert.Equal(t, []string{"publisherTestPod1"}, keys)
@@ -107,6 +108,8 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	otDecode, err := wmb.DecodeToWMB(otValue)
 	assert.NoError(t, err)
 	assert.True(t, otDecode.Idle)
+	assert.Equal(t, otDecode.Watermark, epoch)
+	assert.Equal(t, otDecode.Offset, int64(-1))
 
 	_ = p.Close()
 
diff --git a/pkg/watermark/wmb/wmb_test.go b/pkg/watermark/wmb/wmb_test.go
index bef3d56c..ea82da9d 100644
--- a/pkg/watermark/wmb/wmb_test.go
+++ b/pkg/watermark/wmb/wmb_test.go
@@ -19,8 +19,11 @@ package wmb
 import (
 	"bytes"
 	"encoding/binary"
+	"fmt"
 	"reflect"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestDecodeToOTValue(t *testing.T) {
@@ -198,3 +201,115 @@ func TestOTValue_EncodeToBytes(t *testing.T) {
 		})
 	}
 }
+
+func TestWMBChecker_ValidateHeadWMB(t *testing.T) {
+	var (
+		c     = NewWMBChecker(2)
+		tests = []struct {
+			name        string
+			wmbList     []WMB
+			wantCounter []int
+			want        bool
+		}{
+			{
+				name: "good",
+				wmbList: []WMB{
+					{
+						Idle:      true,
+						Offset:    0,
+						Watermark: 1000,
+					},
+					{
+						Idle:      true,
+						Offset:    0,
+						Watermark: 1000,
+					},
+				},
+				wantCounter: []int{
+					1, 0,
+				},
+				want: true,
+			},
+			{
+				name: "diff_head_wmb",
+				wmbList: []WMB{
+					{
+						Idle:      true,
+						Offset:    0,
+						Watermark: 1000,
+					},
+					{
+						Idle:      true,
+						Offset:    2, // diff head wmb, will return false
+						Watermark: 3000,
+					},
+				},
+				wantCounter: []int{
+					1, 0,
+				},
+				want: false,
+			},
+			{
+				name: "active_head_wmb_2",
+				wmbList: []WMB{
+					{
+						Idle:      true,
+						Offset:    0,
+						Watermark: 1000,
+					},
+					{
+						Idle:      false, // not idle, will return false
+						Offset:    1,
+						Watermark: 2000,
+					},
+				},
+				wantCounter: []int{
+					1, 0,
+				},
+				want: false,
+			},
+			{
+				name: "active_head_wmb_1",
+				wmbList: []WMB{
+					{
+						Idle:      false, // not idle, will return false
+						Offset:    2,
+						Watermark: 2000,
+					},
+				},
+				wantCounter: []int{
+					0,
+				},
+				want: false,
+			},
+			{
+				name: "good_check_again",
+				wmbList: []WMB{
+					{
+						Idle:      true,
+						Offset:    3,
+						Watermark: 4000,
+					},
+					{
+						Idle:      true,
+						Offset:    3,
+						Watermark: 4000,
+					},
+				},
+				wantCounter: []int{
+					1, 0,
+				},
+				want: true,
+			},
+		}
+	)
+	for _, test := range tests {
+		var result bool
+		for i, w := range test.wmbList {
+			result = c.ValidateHeadWMB(w)
+			assert.Equal(t, test.wantCounter[i], c.GetCounter(), fmt.Sprintf("test [%s] failed: want %d, got %d", test.name, test.wantCounter[i], c.GetCounter()))
+		}
+		assert.Equal(t, test.want, result, fmt.Sprintf("test [%s] failed: want %t, got %t", test.name, test.want, result))
+	}
+
+}