Repository: numaproj/numaflow
Pull request #575 (2023-02-28 23:39:09)

fix: Idle watermark fix for read batch size > 0 and partial idle outgoing edges
In the case when a pod doesn't publish to all the out edges, we send an idle watermark whose watermark is assigned as the read batch's watermark.

Note that this solution is when the read in batch size is > 0. Meaning the reduce vertex is receiving incoming messages.


Patch:
diff --git a/.github/workflows/nightly-build.yml b/.github/workflows/nightly-build.yml
new file mode 100644
index 00000000..4ee6f61d
--- /dev/null
+++ b/.github/workflows/nightly-build.yml
@@ -0,0 +1,88 @@
+name: nightly-build
+
+on:
+  schedule:
+    - cron: '55 23 * * *'
+  push:
+    branches:
+      - dev-nightly*
+
+permissions:
+  contents: read
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  build-binaries:
+    runs-on: ubuntu-20.04
+    if: github.repository == 'numaproj/numaflow'
+    name: Build binaries
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+
+      - name: Setup Go
+        uses: actions/setup-go@v3.1.0
+        with:
+          go-version: 1.19
+
+      - name: Build binaries
+        run: |
+          make build
+          chmod -R +x dist
+      - name: Make checksums
+        run: make checksums
+      - name: store artifacts
+        uses: actions/upload-artifact@v3
+        with:
+          name: binaries
+          path: dist
+
+  build-push-linux-multi:
+    name: Build & push linux/amd64 and linux/arm64
+    needs: [ build-binaries ]
+    runs-on: ubuntu-20.04
+    if: github.repository == 'numaproj/numaflow'
+    strategy:
+      matrix:
+        target: [ numaflow ]
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setup Node
+        uses: actions/setup-node@v3
+        with:
+          node-version: "16"
+      - name: Setup Node-Cache
+        uses: actions/cache@v3
+        with:
+          path: ui/node_modules
+          key: ${{ runner.os }}-node-dep-v1-${{ hashFiles('**/yarn.lock') }}
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v2
+
+      - name: Download binaries
+        uses: actions/download-artifact@v3
+        with:
+          name: binaries
+          path: dist/
+
+      - name: Registry Login
+        uses: docker/login-action@v2
+        with:
+          registry: quay.io
+          username: ${{ secrets.QUAYIO_USERNAME }}
+          password: ${{ secrets.QUAYIO_PASSWORD }}
+
+      - name: Set Version
+        id: version
+        run: |
+          tag=$(date +"%Y%m%d")
+          echo "VERSION=nightly-$tag" >> $GITHUB_OUTPUT
+
+      - name: Container build and push with arm64/amd64
+        run: |
+          IMAGE_NAMESPACE=${{ secrets.QUAYIO_ORG }} VERSION=${{ steps.version.outputs.VERSION }} DOCKER_PUSH=true DOCKER_BUILD_ARGS="--label \"quay.expires-after=30d\"" make image-multi
+
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 606db310..68be289a 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -6,7 +6,7 @@ on:
       - 'v*'
     branches:
       - main
-      - dev-*
+      - dev-release*
 
 defaults:
   run:
@@ -81,7 +81,7 @@ jobs:
           if [ $tag = "main" ]; then
             tag="latest"
           fi
-          echo "::set-output name=VERSION::$tag"
+          echo "VERSION=$tag" >> $GITHUB_OUTPUT
 
       - name: Container build and push with arm64/amd64
         run: |
diff --git a/Makefile b/Makefile
index 30d82ddd..3f44324b 100644
--- a/Makefile
+++ b/Makefile
@@ -12,6 +12,7 @@ GIT_TAG=$(shell if [ -z "`git status --porcelain`" ]; then git describe --exact-
 GIT_TREE_STATE=$(shell if [ -z "`git status --porcelain`" ]; then echo "clean" ; else echo "dirty"; fi)
 
 DOCKER_PUSH?=false
+DOCKER_BUILD_ARGS?=
 IMAGE_NAMESPACE?=quay.io/numaproj
 VERSION?=latest
 BASE_VERSION:=latest
@@ -147,14 +148,14 @@ ui-test: ui-build
 
 .PHONY: image
 image: clean ui-build dist/$(BINARY_NAME)-linux-amd64
-	DOCKER_BUILDKIT=1 $(DOCKER) build --build-arg "ARCH=amd64" --build-arg "BASE_IMAGE=$(DEV_BASE_IMAGE)" -t $(IMAGE_NAMESPACE)/$(BINARY_NAME):$(VERSION) --target $(BINARY_NAME) -f $(DOCKERFILE) .
+	DOCKER_BUILDKIT=1 $(DOCKER) build --build-arg "ARCH=amd64" --build-arg "BASE_IMAGE=$(DEV_BASE_IMAGE)" $(DOCKER_BUILD_ARGS) -t $(IMAGE_NAMESPACE)/$(BINARY_NAME):$(VERSION) --target $(BINARY_NAME) -f $(DOCKERFILE) .
 	@if [ "$(DOCKER_PUSH)" = "true" ]; then $(DOCKER) push $(IMAGE_NAMESPACE)/$(BINARY_NAME):$(VERSION); fi
 ifdef IMAGE_IMPORT_CMD
 	$(IMAGE_IMPORT_CMD) $(IMAGE_NAMESPACE)/$(BINARY_NAME):$(VERSION)
 endif
 
 image-multi: ui-build set-qemu dist/$(BINARY_NAME)-linux-arm64.gz dist/$(BINARY_NAME)-linux-amd64.gz
-	$(DOCKER) buildx build --sbom=false --provenance=false --build-arg "BASE_IMAGE=$(RELEASE_BASE_IMAGE)" --tag $(IMAGE_NAMESPACE)/$(BINARY_NAME):$(VERSION) --target $(BINARY_NAME) --platform linux/amd64,linux/arm64 --file ./Dockerfile ${PUSH_OPTION} .
+	$(DOCKER) buildx build --sbom=false --provenance=false --build-arg "BASE_IMAGE=$(RELEASE_BASE_IMAGE)" $(DOCKER_BUILD_ARGS) -t $(IMAGE_NAMESPACE)/$(BINARY_NAME):$(VERSION) --target $(BINARY_NAME) --platform linux/amd64,linux/arm64 --file $(DOCKERFILE) ${PUSH_OPTION} .
 
 set-qemu:
 	$(DOCKER) pull tonistiigi/binfmt:latest
diff --git a/docs/operations/releases.md b/docs/operations/releases.md
index e2c2138a..f7dc631a 100644
--- a/docs/operations/releases.md
+++ b/docs/operations/releases.md
@@ -13,3 +13,9 @@ There's a `stable` tag, pointing to a latest stable release, usually it is the l
 ## Release Cycle
 
 TBD as Numaflow is under active development.
+
+## Nightly Build
+
+If you want to try out the new features on `main` branch, Numaflow provides nightly build images from `main`, the images are available in the format of `quay.io/numaproj/numaflow:nightly-yyyyMMdd`.
+
+Nightly build images expire in 30 days.
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index 1f8e0815..8bc58fb5 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -344,7 +344,8 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// batch processing cycle, send an idle watermark
 		for bufferName := range isdf.publishWatermark {
 			if !activeWatermarkBuffers[bufferName] {
-				isdf.publishWatermark[bufferName].PublishIdleWatermark()
+				// use the watermark of the current read batch for the idle watermark
+				isdf.publishWatermark[bufferName].PublishIdleWatermark(processorWM)
 			}
 		}
 	}
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index ff164e04..587d97be 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -265,7 +265,8 @@ func (p *ProcessAndForward) publishWM(wm processor.Watermark, writeOffsets map[s
 		// batch processing cycle, send an idle watermark
 		for bufferName := range p.publishWatermark {
 			if !activeWatermarkBuffers[bufferName] {
-				p.publishWatermark[bufferName].PublishIdleWatermark()
+				// use the watermark of the current read batch for the idle watermark
+				p.publishWatermark[bufferName].PublishIdleWatermark(wm)
 			}
 		}
 	}
diff --git a/pkg/watermark/fetch/offset_timeline.go b/pkg/watermark/fetch/offset_timeline.go
index e3cf875f..ad355cc9 100644
--- a/pkg/watermark/fetch/offset_timeline.go
+++ b/pkg/watermark/fetch/offset_timeline.go
@@ -171,6 +171,26 @@ func (t *OffsetTimeline) GetHeadOffsetWatermark() OffsetWatermark {
 	return t.watermarks.Front().Value.(OffsetWatermark)
 }
 
+// GetReferredWatermark returns the referred watermark that will be used to replace
+// the idle watermark value
+func (t *OffsetTimeline) GetReferredWatermark(idleWM int64) OffsetWatermark {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+	// find in the offset timeline where the OffsetWatermark has a 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)
+		}
+	}
+	return OffsetWatermark{
+		watermark: -1,
+		offset:    -1,
+	}
+}
+
 // GetOffset will return the offset for the given event-time.
 // TODO(jyu6): will make Watermark an interface make it easy to pass an Offset and return a Watermark?
 func (t *OffsetTimeline) GetOffset(eventTime int64) int64 {
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index 41a21178..24cfa24d 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -252,26 +252,30 @@ func (v *ProcessorManager) startTimeLineWatcher() {
 				if otValue.Idle {
 					var processors = v.GetAllProcessors()
 					for processorName, processor := range processors {
-						// skip the processor itself, only use other processors as reference
+						// skip the idle processor itself, only use other processors as reference
 						if processorName != value.Key() {
-							// any other Vn-1 processor's non-empty head offsetWatermark can replace the idle watermark
-							// if all tail offsetWatermarks are empty, then it means there's no data flowing into
-							// this Vn processor, so it's safe to do nothing
-							watermarkOffset := processor.offsetTimeline.GetHeadOffsetWatermark()
-							if watermarkOffset.offset != -1 {
-								p.offsetTimeline.PutIdle(watermarkOffset)
+							// 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 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,
 					})
+					v.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", v.otWatcher.GetKVName()), zap.Int64("watermark", otValue.Watermark), zap.Int64("offset", otValue.Offset))
 				}
-				v.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", v.otWatcher.GetKVName()), zap.Int64("watermark", otValue.Watermark), zap.Int64("offset", otValue.Offset))
 			case store.KVDelete:
 				// we do not care about Delete events because the timeline bucket is meant to grow and the TTL will
 				// naturally trim the KV store.
diff --git a/pkg/watermark/generic/noop.go b/pkg/watermark/generic/noop.go
index 7913ef92..53bbf356 100644
--- a/pkg/watermark/generic/noop.go
+++ b/pkg/watermark/generic/noop.go
@@ -47,7 +47,7 @@ func (n NoOpWMProgressor) PublishWatermark(_ processor.Watermark, _ isb.Offset)
 }
 
 // PublishIdleWatermark does a no-op idle watermark publish.
-func (n NoOpWMProgressor) PublishIdleWatermark() {
+func (n NoOpWMProgressor) PublishIdleWatermark(processor.Watermark) {
 }
 
 // GetLatestWatermark returns the default watermark as the latest watermark.
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index c3c050ec..ddd656eb 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(processor.Watermark, isb.Offset)
 	// PublishIdleWatermark publishes the idle watermark.
-	PublishIdleWatermark()
+	PublishIdleWatermark(wm processor.Watermark)
 	// GetLatestWatermark returns the latest published watermark.
 	GetLatestWatermark() processor.Watermark
 }
@@ -96,18 +96,8 @@ func (p *publish) initialSetup() {
 // PublishWatermark publishes watermark and will retry until it can succeed. It will not publish if the new-watermark
 // is less than the current head watermark.
 func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
-	if p.opts.isSource && p.opts.delay.Nanoseconds() > 0 && !time.Time(wm).IsZero() {
-		wm = processor.Watermark(time.Time(wm).Add(-p.opts.delay))
-	}
-	// update p.headWatermark only if wm > p.headWatermark
-	if wm.After(time.Time(p.headWatermark)) {
-		p.log.Debugw("New watermark is updated for the head watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
-		p.headWatermark = wm
-	} else if wm.Before(time.Time(p.headWatermark)) {
-		p.log.Warnw("Skip publishing the new watermark because it's older than the current watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
-		return
-	} else {
-		p.log.Debugw("Skip publishing the new watermark because it's the same as the current watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
+	validWM, skipWM := p.validateWatermark(wm)
+	if skipWM {
 		return
 	}
 
@@ -123,7 +113,7 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 	}
 	var otValue = ot.Value{
 		Offset:    seq,
-		Watermark: wm.UnixMilli(),
+		Watermark: validWM.UnixMilli(),
 	}
 	value, err := otValue.EncodeToBytes()
 	if err != nil {
@@ -137,18 +127,42 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 			// TODO: better exponential backoff
 			time.Sleep(time.Millisecond * 250)
 		} else {
-			p.log.Debugw("New watermark published with offset", zap.Int64("head", p.headWatermark.UnixMilli()), zap.Int64("new", wm.UnixMilli()), zap.Int64("offset", seq))
+			p.log.Debugw("New watermark published with offset", zap.Int64("head", p.headWatermark.UnixMilli()), zap.Int64("new", validWM.UnixMilli()), zap.Int64("offset", seq))
 			break
 		}
 	}
 }
 
+// validateWatermark checks if the new watermark is greater than the head watermark, return true if yes,
+// otherwise, return false
+func (p *publish) validateWatermark(wm processor.Watermark) (processor.Watermark, bool) {
+	if p.opts.isSource && p.opts.delay.Nanoseconds() > 0 && !time.Time(wm).IsZero() {
+		wm = processor.Watermark(time.Time(wm).Add(-p.opts.delay))
+	}
+	// update p.headWatermark only if wm > p.headWatermark
+	if wm.After(time.Time(p.headWatermark)) {
+		p.log.Debugw("New watermark is updated for the head watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
+		p.headWatermark = wm
+	} else if wm.Before(time.Time(p.headWatermark)) {
+		p.log.Warnw("Skip publishing the new watermark because it's older than the current watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
+		return processor.Watermark{}, true
+	} else {
+		p.log.Debugw("Skip publishing the new watermark because it's the same as the current watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
+		return processor.Watermark{}, true
+	}
+	return wm, false
+}
+
 // PublishIdleWatermark publishes the idle watermark and will retry until it can succeed.
-func (p *publish) PublishIdleWatermark() {
+func (p *publish) PublishIdleWatermark(wm processor.Watermark) {
 	var key = p.entity.GetName()
+	validWM, skipWM := p.validateWatermark(wm)
+	if skipWM {
+		return
+	}
 	var otValue = ot.Value{
 		Offset:    0,
-		Watermark: 0,
+		Watermark: validWM.UnixMilli(),
 		Idle:      true,
 	}
 	value, err := otValue.EncodeToBytes()
diff --git a/ui/src/components/pipeline/graph/CustomEdge.css b/ui/src/components/pipeline/graph/CustomEdge.css
index 24d78a32..c7762aba 100644
--- a/ui/src/components/pipeline/graph/CustomEdge.css
+++ b/ui/src/components/pipeline/graph/CustomEdge.css
@@ -4,15 +4,12 @@
 
 .edge-watermark {
     width: fit-content;
-    height: 12px;
-    background: #ccd6dd;
-    border: 1px solid #8fa4b1;
+    background: white;
     position: absolute;
-    border-radius: 5px;
+    border-radius: 2px;
     padding: 0px 5px;
     font-size: 0.8em;
     text-transform: lowercase;
-    margin-top: -20px;
     text-align: center;
     font-family: "IBM Plex Sans", sans-serif;
     pointer-events: all;
@@ -20,15 +17,12 @@
 
 .edge-backpressure {
     width: fit-content;
-    height: 12px;
-    background: #ccd6dd;
-    border: 1px solid #8fa4b1;
+    background: white;
     position: absolute;
-    border-radius: 5px;
+    border-radius: 2px;
     padding: 0px 5px;
     font-size: 0.8em;
     text-transform: lowercase;
-    margin-top: 8px;
     text-align: center;
     font-family: "IBM Plex Sans", sans-serif;
     pointer-events: all;
diff --git a/ui/src/components/pipeline/graph/CustomEdge.tsx b/ui/src/components/pipeline/graph/CustomEdge.tsx
index f5fb67f8..a2e4de88 100644
--- a/ui/src/components/pipeline/graph/CustomEdge.tsx
+++ b/ui/src/components/pipeline/graph/CustomEdge.tsx
@@ -31,6 +31,15 @@ const CustomEdge: FC<EdgeProps> = ({
             fontWeight: 700,
         };
     }
+    let wmStyle;
+    let pendingStyle;
+    if (sourceY === targetY) {
+        wmStyle = { marginTop: "-12px" }
+        pendingStyle = { marginTop: "1px" }
+    } else {
+        wmStyle = { right: "1px" }
+        pendingStyle = { left: "1px" }
+    }
 
     return (
         <>
@@ -44,10 +53,9 @@ const CustomEdge: FC<EdgeProps> = ({
                         fontFamily: "IBM Plex Sans",
                         fontWeight: 400,
                         fontSize: "0.6rem",
-                        width: "40px",
+                        width: "fit-content",
                         display: "flex",
                         justifyContent: "center",
-
                     }}
                     className="nodrag nopan"
                 >
@@ -64,7 +72,7 @@ const CustomEdge: FC<EdgeProps> = ({
                             arrow
                             placement={"top"}
                         >
-                            <div className={"edge-watermark"}>
+                            <div className={"edge-watermark"} style={wmStyle}>
                                 {Math.min.apply(null, data.edgeWatermark.watermarks)}
                             </div>
                         </Tooltip>
@@ -78,7 +86,7 @@ const CustomEdge: FC<EdgeProps> = ({
                         arrow
                         placement={"bottom"}
                     >
-                        <div className={"edge-backpressure"}>
+                        <div className={"edge-backpressure"} style={pendingStyle}>
                             {data.backpressureLabel}
                         </div>
                     </Tooltip>


Test patch:
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index e5de4a92..fec028c1 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -77,7 +77,7 @@ func (e *EventTypeWMProgressor) PublishWatermark(watermark processor.Watermark,
 	e.watermarks[offset.String()] = watermark
 }
 
-func (e *EventTypeWMProgressor) PublishIdleWatermark() {
+func (e *EventTypeWMProgressor) PublishIdleWatermark(processor.Watermark) {
 	// TODO
 }
 
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 0c0fbce4..8180e88e 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -212,12 +212,12 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			wmExpected: map[string]ot.Value{
 				"buffer1": {
 					Offset:    0,
-					Watermark: 60000,
+					Watermark: int64(60000),
 					Idle:      false,
 				},
 				"buffer2": {
 					Offset:    0,
-					Watermark: 0,
+					Watermark: int64(60000),
 					Idle:      true,
 				},
 			},
@@ -236,12 +236,12 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			wmExpected: map[string]ot.Value{
 				"buffer1": {
 					Offset:    0,
-					Watermark: 60000,
+					Watermark: int64(60000),
 					Idle:      false,
 				},
 				"buffer2": {
 					Offset:    0,
-					Watermark: 60000,
+					Watermark: int64(60000),
 					Idle:      false,
 				},
 			},
@@ -260,12 +260,12 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			wmExpected: map[string]ot.Value{
 				"buffer1": {
 					Offset:    0,
-					Watermark: 0,
+					Watermark: int64(60000),
 					Idle:      true,
 				},
 				"buffer2": {
 					Offset:    0,
-					Watermark: 0,
+					Watermark: int64(60000),
 					Idle:      true,
 				},
 			},
diff --git a/pkg/watermark/fetch/offset_timeline_test.go b/pkg/watermark/fetch/offset_timeline_test.go
index 8c6fbd66..5549188e 100644
--- a/pkg/watermark/fetch/offset_timeline_test.go
+++ b/pkg/watermark/fetch/offset_timeline_test.go
@@ -18,6 +18,7 @@ package fetch
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"testing"
 
@@ -281,3 +282,52 @@ func TestOffsetTimeline(t *testing.T) {
 	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_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},
+		}
+		referredEventTimes = [10]int64{
+			12,
+			12,
+			12,
+			13,
+			16,
+			19,
+			20,
+			28,
+			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},
+		}
+	)
+
+	for i, watermark := range testWatermarks {
+		testTimeline.Put(watermark)
+		assert.Equal(t, expectedReferredWatermarks[i], testTimeline.GetReferredWatermark(referredEventTimes[i]), fmt.Sprintf("test %d", i))
+	}
+	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())
+}
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index 2708ff86..67ed1873 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -312,7 +312,7 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	}
 
 	// publish an idle watermark
-	otValueByte, err = otValueToBytes(0, 0, true)
+	otValueByte, err = otValueToBytes(0, epoch+600, true)
 	assert.NoError(t, err)
 	err = otStore.PutKV(ctx, "p1", otValueByte)
 	assert.NoError(t, err)
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index 0a6ae629..cf534530 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -96,7 +96,8 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	head := p.GetLatestWatermark()
 	assert.Equal(t, processor.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), head.String())
 
-	p.PublishIdleWatermark()
+	// simulate the next read batch where the watermark is now $epoch
+	p.PublishIdleWatermark(processor.Watermark(time.UnixMilli(epoch).In(location)))
 	keys, err = p.otStore.GetAllKeys(p.ctx)
 	assert.NoError(t, err)
 	assert.Equal(t, []string{"publisherTestPod1"}, keys)