Repository: numaproj/numaflow
Pull request #614 (2023-03-16 14:05:49)

chore: fix publish watermark inside tickgen


Patch:
diff --git a/Makefile b/Makefile
index 3f44324b..81ba344e 100644
--- a/Makefile
+++ b/Makefile
@@ -193,6 +193,7 @@ manifests: crds
 	kubectl kustomize config/cluster-install > config/install.yaml
 	kubectl kustomize config/namespace-install > config/namespace-install.yaml
 	kubectl kustomize config/advanced-install/namespaced-controller > config/advanced-install/namespaced-controller-wo-crds.yaml
+	kubectl kustomize config/advanced-install/namespaced-numaflow-server > config/advanced-install/namespaced-numaflow-server.yaml
 	kubectl kustomize config/advanced-install/numaflow-server > config/advanced-install/numaflow-server.yaml
 	kubectl kustomize config/advanced-install/minimal-crds > config/advanced-install/minimal-crds.yaml
 
diff --git a/config/advanced-install/namespaced-controller-wo-crds.yaml b/config/advanced-install/namespaced-controller-wo-crds.yaml
index db4c2d05..f51cd9a5 100644
--- a/config/advanced-install/namespaced-controller-wo-crds.yaml
+++ b/config/advanced-install/namespaced-controller-wo-crds.yaml
@@ -191,7 +191,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.9.8
+          natsImage: nats:2.9.15
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -235,6 +235,11 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
+        - version: 2.9.15
+          natsImage: nats:2.9.15
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
 kind: ConfigMap
 metadata:
   name: numaflow-controller-config
diff --git a/config/advanced-install/namespaced-numaflow-server.yaml b/config/advanced-install/namespaced-numaflow-server.yaml
new file mode 100644
index 00000000..5d8e306f
--- /dev/null
+++ b/config/advanced-install/namespaced-numaflow-server.yaml
@@ -0,0 +1,128 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: numaflow-server-sa
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: numaflow-server-role
+rules:
+- apiGroups:
+  - numaflow.numaproj.io
+  resources:
+  - interstepbufferservices
+  - interstepbufferservices/finalizers
+  - interstepbufferservices/status
+  - pipelines
+  - pipelines/finalizers
+  - pipelines/status
+  - vertices
+  - vertices/finalizers
+  - vertices/status
+  - vertices/scale
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  - pods/exec
+  - pods/log
+  - configmaps
+  - services
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - apps
+  resources:
+  - deployments
+  - statefulsets
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - metrics.k8s.io
+  resources:
+  - pods
+  verbs:
+  - get
+  - list
+  - watch
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: numaflow-server-binding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: numaflow-server-role
+subjects:
+- kind: ServiceAccount
+  name: numaflow-server-sa
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: numaflow-server
+spec:
+  ports:
+  - port: 8443
+    targetPort: 8443
+  selector:
+    app.kubernetes.io/component: numaflow-ux
+    app.kubernetes.io/part-of: numaflow
+  type: ClusterIP
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: numaflow-server
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: numaflow-ux
+      app.kubernetes.io/part-of: numaflow
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/component: numaflow-ux
+        app.kubernetes.io/part-of: numaflow
+    spec:
+      containers:
+      - args:
+        - server
+        - --namespaced
+        env:
+        - name: NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        image: quay.io/numaproj/numaflow:latest
+        imagePullPolicy: Always
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8443
+            scheme: HTTPS
+          initialDelaySeconds: 3
+          periodSeconds: 3
+        name: main
+        resources:
+          limits:
+            cpu: 500m
+            memory: 1024Mi
+          requests:
+            cpu: 100m
+            memory: 200Mi
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 9737
+      serviceAccountName: numaflow-server-sa
diff --git a/config/advanced-install/namespaced-numaflow-server/kustomization.yaml b/config/advanced-install/namespaced-numaflow-server/kustomization.yaml
new file mode 100644
index 00000000..92dfbc2e
--- /dev/null
+++ b/config/advanced-install/namespaced-numaflow-server/kustomization.yaml
@@ -0,0 +1,23 @@
+#
+# Controller only with namespaced install, witout CRD
+#
+
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - ../../base/numaflow-server
+  - ../../namespace-install/rbac/numaflow-server
+
+images:
+  - name: quay.io/numaproj/numaflow
+    newTag: latest
+
+patches:
+  - patch: |-
+      - op: add
+        path: /spec/template/spec/containers/0/args/-
+        value: --namespaced
+    target:
+      kind: Deployment
+      name: numaflow-server
diff --git a/config/base/controller-manager/numaflow-controller-config.yaml b/config/base/controller-manager/numaflow-controller-config.yaml
index d8a45289..6ea15406 100644
--- a/config/base/controller-manager/numaflow-controller-config.yaml
+++ b/config/base/controller-manager/numaflow-controller-config.yaml
@@ -78,7 +78,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.9.8
+          natsImage: nats:2.9.15
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -122,3 +122,8 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
+        - version: 2.9.15
+          natsImage: nats:2.9.15
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
diff --git a/config/install.yaml b/config/install.yaml
index 757a4132..ffd85c13 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -11716,7 +11716,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.9.8
+          natsImage: nats:2.9.15
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -11760,6 +11760,11 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
+        - version: 2.9.15
+          natsImage: nats:2.9.15
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
 kind: ConfigMap
 metadata:
   name: numaflow-controller-config
diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
index a6eaa96e..c0b2f27b 100644
--- a/config/namespace-install.yaml
+++ b/config/namespace-install.yaml
@@ -11627,7 +11627,7 @@ data:
             replicas: 3
         versions:
         - version: latest
-          natsImage: nats:2.9.8
+          natsImage: nats:2.9.15
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
@@ -11671,6 +11671,11 @@ data:
           metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
           configReloaderImage: natsio/nats-server-config-reloader:0.7.0
           startCommand: /nats-server
+        - version: 2.9.15
+          natsImage: nats:2.9.15
+          metricsExporterImage: natsio/prometheus-nats-exporter:0.9.1
+          configReloaderImage: natsio/nats-server-config-reloader:0.7.0
+          startCommand: /nats-server
 kind: ConfigMap
 metadata:
   name: numaflow-controller-config
diff --git a/go.mod b/go.mod
index ce46d48e..e0fb57ed 100644
--- a/go.mod
+++ b/go.mod
@@ -26,7 +26,7 @@ require (
 	github.com/imdario/mergo v0.3.13
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe
 	github.com/nats-io/nats-server/v2 v2.7.5-0.20220415000625-a6b62f61a703
-	github.com/nats-io/nats.go v1.21.0
+	github.com/nats-io/nats.go v1.24.0
 	github.com/numaproj/numaflow-go v0.3.4
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.32.1
@@ -164,7 +164,7 @@ require (
 	github.com/yudai/gojsondiff v1.0.0 // indirect
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // indirect
 	go.mongodb.org/mongo-driver v1.7.3 // indirect
-	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
+	golang.org/x/crypto v0.5.0 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/net v0.7.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
diff --git a/go.sum b/go.sum
index 7f7a3afd..596d9fa2 100644
--- a/go.sum
+++ b/go.sum
@@ -689,8 +689,8 @@ github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a h1:lem6QCvxR0Y28g
 github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
 github.com/nats-io/nats-server/v2 v2.7.5-0.20220415000625-a6b62f61a703 h1:d8siT+8VQ68hfqPqYZvpMrHIihlMVW3gGy+o2hRDCyg=
 github.com/nats-io/nats-server/v2 v2.7.5-0.20220415000625-a6b62f61a703/go.mod h1:5vic7C58BFEVltiZhs7Kq81q2WcEPhJPsmNv1FOrdv0=
-github.com/nats-io/nats.go v1.21.0 h1:kQiWyQMMMIPjDR7NanrLhTnRUxWgU04yrzmYdq9JxCU=
-github.com/nats-io/nats.go v1.21.0/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
+github.com/nats-io/nats.go v1.24.0 h1:CRiD8L5GOQu/DcfkmgBcTTIQORMwizF+rPk6T0RaHVQ=
+github.com/nats-io/nats.go v1.24.0/go.mod h1:dVQF+BK3SzUZpwyzHedXsvH3EO38aVKuOPkkHlv5hXA=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -972,8 +972,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210920023735-84f357641f63/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
-golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
+golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
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/reconciler/pipeline/controller.go b/pkg/reconciler/pipeline/controller.go
index 53ad26b8..149b774b 100644
--- a/pkg/reconciler/pipeline/controller.go
+++ b/pkg/reconciler/pipeline/controller.go
@@ -435,6 +435,12 @@ func buildVertices(pl *dfv1.Pipeline) map[string]dfv1.Vertex {
 		replicas := int32(1)
 		if pl.Status.Phase == dfv1.PipelinePhasePaused {
 			replicas = int32(0)
+		} else if v.UDF != nil && v.UDF.GroupBy != nil {
+			for _, edge := range fromEdges {
+				if edge.Parallelism != nil {
+					replicas = *edge.Parallelism
+				}
+			}
 		} else {
 			x := vCopy.Scale
 			if x.Min != nil && *x.Min > 1 && replicas < *x.Min {
@@ -444,6 +450,7 @@ func buildVertices(pl *dfv1.Pipeline) map[string]dfv1.Vertex {
 				replicas = *x.Max
 			}
 		}
+
 		spec := dfv1.VertexSpec{
 			AbstractVertex:             *vCopy,
 			PipelineName:               pl.Name,
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/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 7f1ea122..6a8fcd9b 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -275,10 +275,8 @@ func (mg *memgen) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	if len(msgs) <= 0 {
 		return
 	}
-	// use the first event time as watermark to make it conservative
-	nanos, _ := msgs[0].ReadOffset.Sequence()
-	// remove the nanosecond precision
-	mg.sourcePublishWM.PublishWatermark(wmb.Watermark(time.Unix(0, nanos)), nil) // Source publisher does not care about the offset
+	// use the first event time of the message as watermark to make it conservative
+	mg.sourcePublishWM.PublishWatermark(wmb.Watermark(msgs[0].EventTime), nil) // Source publisher does not care about the offset
 }
 
 // Ack acknowledges an array of offset.
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))
+	}
+
+}