Repository: numaproj/numaflow (go)

Issue:
Reduce JetStream connections/consumers used by watermark fetcher
# Summary

Today, during watermark propagation, each replica of Vertex N starts a watermark fetcher to watch OT data from Vertex (N-1), depends on how many replicas Vertex (N-1) has, the watermark fetcher runs same number of JetStream K/V watchers, thus same numbers of JetStream connections and consumers. For example, Vertex N has 10 Pods, vertex (N-1) has 5 Pods, then the consumers to OT bucket of V (N-1) is `10 * 5 = 50`. For a Source Vertex OT bucket, the consumer number is always square of replica number. If it's a Kafka source, then it will be square of partitions - a `36` partition Kafka topic source will have `1296` consumers. A `100` replica source vertex will have `10, 000` consumers, which might overwhelm the JetStream server.

We need to enhance the watermark fetcher logic to reduce the connections/consumers to JetStream.

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
feat: watermark otwatcher enhancement
Fixes #345 

# Test
```shell
~ # js s report #in(1) cat(1) out(1)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 1        │ 125 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 5        │ 615 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 2         │ 5        │ 720 B   │ 0    │ 34      │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 2         │ 10       │ 1.4 KiB │ 0    │ 4       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 1         │ 10       │ 1.4 KiB │ 0    │ 4       │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 2         │ 25       │ 3.0 KiB │ 0    │ 89      │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 2         │ 50       │ 5.9 KiB │ 0    │ 13      │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 1         │ 50       │ 6.0 KiB │ 0    │ 20      │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 2,165    │ 471 KiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 2,165    │ 499 KiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # js s report #in(3) cat(1) out(1)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 1        │ 125 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 5        │ 615 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 2         │ 5        │ 720 B   │ 0    │ 129     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 2         │ 10       │ 1.4 KiB │ 0    │ 98      │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 3         │ 10       │ 1.4 KiB │ 0    │ 98      │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 2         │ 25       │ 3.0 KiB │ 0    │ 473     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 2         │ 50       │ 5.9 KiB │ 0    │ 403     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 3         │ 50       │ 6.0 KiB │ 0    │ 495     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 4,560    │ 995 KiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 4,560    │ 1.0 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # js s report #in(3) cat(4) out(1)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 1        │ 125 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 5        │ 615 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 2         │ 5        │ 720 B   │ 0    │ 132     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 5         │ 10       │ 1.4 KiB │ 0    │ 110     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 3         │ 10       │ 1.4 KiB │ 0    │ 110     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 2         │ 25       │ 3.0 KiB │ 0    │ 497     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 5         │ 50       │ 5.9 KiB │ 0    │ 455     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 3         │ 50       │ 6.0 KiB │ 0    │ 549     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 4,830    │ 1.0 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 4,830    │ 1.1 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # js s report #in(3) cat(4) out(2)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 2        │ 250 B   │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 3         │ 5        │ 645 B   │ 0    │ 179     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 10       │ 1.2 KiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 5         │ 10       │ 1.4 KiB │ 0    │ 145     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 3         │ 10       │ 1.4 KiB │ 0    │ 145     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 3         │ 25       │ 3.0 KiB │ 0    │ 729     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 5         │ 50       │ 5.9 KiB │ 0    │ 617     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 3         │ 50       │ 6.0 KiB │ 0    │ 723     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 5,700    │ 1.2 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 5,700    │ 1.3 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # js s report #in(1) cat(1) out(1)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 2        │ 275 B   │ 0    │ 2       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 2         │ 5        │ 720 B   │ 0    │ 212     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 10       │ 1.2 KiB │ 0    │ 20      │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 2         │ 10       │ 1.4 KiB │ 0    │ 173     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 1         │ 10       │ 1.4 KiB │ 0    │ 173     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 2         │ 25       │ 3.0 KiB │ 0    │ 977     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 2         │ 50       │ 5.9 KiB │ 0    │ 746     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 1         │ 50       │ 6.0 KiB │ 0    │ 854     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 6,356    │ 1.4 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 6,360    │ 1.4 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # js s report #in(0) cat(0) out(0)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 2        │ 300 B   │ 0    │ 7       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 1         │ 5        │ 745 B   │ 0    │ 216     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 10       │ 1.2 KiB │ 0    │ 28      │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 1         │ 10       │ 1.4 KiB │ 0    │ 176     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 0         │ 10       │ 1.5 KiB │ 0    │ 176     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 1         │ 25       │ 3.0 KiB │ 0    │ 985     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 1         │ 50       │ 5.9 KiB │ 0    │ 754     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 0         │ 50       │ 6.0 KiB │ 0    │ 866     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 6,416    │ 1.4 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 6,420    │ 1.4 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # js s report #in(1) cat(1) out(1)
Obtaining Stream stats

╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                                                                              Stream Report                                                                                               │
├─────────────────────────────────────────────────────────────────┬─────────┬───────────┬───────────┬──────────┬─────────┬──────┬─────────┬────────────────────────────────────────────────────────────────┤
│ Stream                                                          │ Storage │ Placement │ Consumers │ Messages │ Bytes   │ Lost │ Deleted │ Replicas                                                       │
├─────────────────────────────────────────────────────────────────┼─────────┼───────────┼───────────┼──────────┼─────────┼──────┼─────────┼────────────────────────────────────────────────────────────────┤
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_PROCESSORS  │ File    │           │ 1         │ 2        │ 275 B   │ 0    │ 13      │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_PROCESSORS   │ File    │           │ 2         │ 5        │ 720 B   │ 0    │ 221     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-out_SINK_OT          │ File    │           │ 1         │ 10       │ 1.2 KiB │ 0    │ 40      │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_PROCESSORS    │ File    │           │ 2         │ 10       │ 1.4 KiB │ 0    │ 180     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_PROCESSORS │ File    │           │ 1         │ 10       │ 1.4 KiB │ 0    │ 180     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ KV_simple-pipeline-default-simple-pipeline-cat-out_OT           │ File    │           │ 2         │ 25       │ 3.0 KiB │ 0    │ 997     │ isbsvc-default-js-0*, isbsvc-default-js-1, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in-cat_OT            │ File    │           │ 2         │ 50       │ 5.9 KiB │ 0    │ 767     │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ KV_simple-pipeline-default-simple-pipeline-in_SOURCE_OT         │ File    │           │ 1         │ 50       │ 6.0 KiB │ 0    │ 886     │ isbsvc-default-js-0, isbsvc-default-js-1, isbsvc-default-js-2* │
│ simple-pipeline-default-simple-pipeline-cat-out                 │ File    │           │ 1         │ 6,525    │ 1.4 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
│ simple-pipeline-default-simple-pipeline-in-cat                  │ File    │           │ 1         │ 6,525    │ 1.5 MiB │ 0    │ 0       │ isbsvc-default-js-0, isbsvc-default-js-1*, isbsvc-default-js-2 │
╰─────────────────────────────────────────────────────────────────┴─────────┴───────────┴───────────┴──────────┴─────────┴──────┴─────────┴────────────────────────────────────────────────────────────────╯

~ # 
```


Patch:
diff --git a/pkg/watermark/fetch/edge_fetcher.go b/pkg/watermark/fetch/edge_fetcher.go
index 1b545eef..7c63d75e 100644
--- a/pkg/watermark/fetch/edge_fetcher.go
+++ b/pkg/watermark/fetch/edge_fetcher.go
@@ -104,7 +104,7 @@ func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) processor.Watermark {
 		}
 		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.GetID())
+			e.processorManager.DeleteProcessor(p.entity.GetName())
 		}
 	}
 	// if the offset is smaller than every offset in the timeline, set the value to be -1
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index 9028d777..00bdb4cf 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -24,7 +24,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
@@ -69,6 +71,7 @@ func NewProcessorManager(ctx context.Context, watermarkStoreWatcher store.Waterm
 	}
 	go v.startRefreshingProcessors()
 	go v.startHeatBeatWatcher()
+	go v.startTimeLineWatcher()
 	return v
 }
 
@@ -141,7 +144,6 @@ func (v *ProcessorManager) refreshingProcessors() {
 			// TODO: how to delete the pod from the heartbeat store?
 			v.log.Infow("Processor has been inactive for 10 heartbeats, deleting...", zap.String("key", pName), zap.String(pName, p.String()))
 			p.setStatus(_deleted)
-			p.stopTimeLineWatcher()
 			v.heartbeat.Delete(pName)
 		} else if time.Now().Unix()-pTime > v.opts.podHeartbeatRate {
 			// if the pod's last heartbeat is greater than podHeartbeatRate
@@ -177,7 +179,7 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 					// The fromProcessor may have been deleted
 					// TODO: make capacity configurable
 					var entity = processor.NewProcessorEntity(value.Key())
-					var fromProcessor = NewProcessorToFetch(v.ctx, entity, 10, v.otWatcher)
+					var fromProcessor = NewProcessorToFetch(v.ctx, entity, 10)
 					v.addProcessor(value.Key(), fromProcessor)
 					v.log.Infow("v.AddProcessor successfully added a new fromProcessor", zap.String("fromProcessor", value.Key()))
 				} else { // else just make a note that this processor is still active
@@ -200,7 +202,6 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 				} else {
 					v.log.Infow("Deleting", zap.String("key", value.Key()), zap.String(value.Key(), p.String()))
 					p.setStatus(_deleted)
-					p.stopTimeLineWatcher()
 					v.heartbeat.Delete(value.Key())
 				}
 			case store.KVPurge:
@@ -212,3 +213,53 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 
 	}
 }
+
+func (v *ProcessorManager) startTimeLineWatcher() {
+	watchCh, stopped := v.otWatcher.Watch(v.ctx)
+
+	for {
+		select {
+		case <-stopped:
+			return
+		case value := <-watchCh:
+			if value == nil {
+				continue
+			}
+			switch value.Operation() {
+			case store.KVPut:
+				p := v.GetProcessor(value.Key())
+				_ = wait.ExponentialBackoffWithContext(v.ctx, wait.Backoff{
+					// default heartbeat rate is set to 5 seconds, so retry every "duration * factor + [0, jitter]" interval for 5 times
+					Duration: 1 * time.Second,
+					Factor:   1,
+					Jitter:   0.1,
+					Steps:    5,
+				}, func() (done bool, err error) {
+					if p = v.GetProcessor(value.Key()); p == nil {
+						return false, nil
+					}
+					return true, nil
+				})
+				if p == nil {
+					v.log.Errorw("Unable to find the processor", zap.String("processorEntity", value.Key()))
+					continue
+				}
+				otValue, err := ot.DecodeToOTValue(value.Value())
+				if err != nil {
+					v.log.Errorw("Unable to decode the value", zap.String("processorEntity", p.entity.GetName()), zap.Error(err))
+					continue
+				}
+				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))
+			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.
+			case store.KVPurge:
+				// skip
+			}
+		}
+	}
+}
diff --git a/pkg/watermark/fetch/processor_to_fetch.go b/pkg/watermark/fetch/processor_to_fetch.go
index b8346839..4187c6ce 100644
--- a/pkg/watermark/fetch/processor_to_fetch.go
+++ b/pkg/watermark/fetch/processor_to_fetch.go
@@ -21,12 +21,10 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"go.uber.org/zap"
 
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
-	"github.com/numaproj/numaflow/pkg/watermark/store"
 )
 
 type status int
@@ -52,32 +50,26 @@ func (s status) String() string {
 // ProcessorToFetch is the smallest unit of entity (from which we fetch data) that does inorder processing or contains inorder data.
 type ProcessorToFetch struct {
 	ctx            context.Context
-	cancel         context.CancelFunc
 	entity         processor.ProcessorEntitier
 	status         status
 	offsetTimeline *OffsetTimeline
-	otWatcher      store.WatermarkKVWatcher
 	lock           sync.RWMutex
 	log            *zap.SugaredLogger
 }
 
 func (p *ProcessorToFetch) String() string {
-	return fmt.Sprintf("%s status:%v, timeline: %s", p.entity.GetID(), p.getStatus(), p.offsetTimeline.Dump())
+	return fmt.Sprintf("%s status:%v, timeline: %s", p.entity.GetName(), p.getStatus(), p.offsetTimeline.Dump())
 }
 
 // NewProcessorToFetch creates ProcessorToFetch.
-func NewProcessorToFetch(ctx context.Context, processor processor.ProcessorEntitier, capacity int, watcher store.WatermarkKVWatcher) *ProcessorToFetch {
-	ctx, cancel := context.WithCancel(ctx)
+func NewProcessorToFetch(ctx context.Context, processor processor.ProcessorEntitier, capacity int) *ProcessorToFetch {
 	p := &ProcessorToFetch{
 		ctx:            ctx,
-		cancel:         cancel,
 		entity:         processor,
 		status:         _active,
 		offsetTimeline: NewOffsetTimeline(ctx, capacity),
-		otWatcher:      watcher,
 		log:            logging.FromContext(ctx),
 	}
-	go p.startTimeLineWatcher()
 	return p
 }
 
@@ -113,46 +105,3 @@ func (p *ProcessorToFetch) IsDeleted() bool {
 	defer p.lock.RUnlock()
 	return p.status == _deleted
 }
-
-func (p *ProcessorToFetch) stopTimeLineWatcher() {
-	p.cancel()
-}
-
-func (p *ProcessorToFetch) startTimeLineWatcher() {
-	watchCh, stopped := p.otWatcher.Watch(p.ctx)
-
-	for {
-		select {
-		case <-stopped:
-			// no need to close ot watcher here because the ot watcher is shared for the given vertex
-			// the parent ctx will close the ot watcher
-			return
-		case value := <-watchCh:
-			// TODO: why will value will be nil?
-			if value == nil {
-				continue
-			}
-			switch value.Operation() {
-			case store.KVPut:
-				if value.Key() != p.entity.BuildOTWatcherKey() {
-					continue
-				}
-				otValue, err := ot.DecodeToOTValue(value.Value())
-				if err != nil {
-					p.log.Errorw("Unable to decode the value", zap.String("processorEntity", p.entity.GetID()), zap.Error(err))
-					continue
-				}
-				p.offsetTimeline.Put(OffsetWatermark{
-					watermark: otValue.Watermark,
-					offset:    otValue.Offset,
-				})
-				p.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", p.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.
-			case store.KVPurge:
-				// skip
-			}
-		}
-	}
-}
diff --git a/pkg/watermark/processor/entity.go b/pkg/watermark/processor/entity.go
index 1c41cc15..7dd786e9 100644
--- a/pkg/watermark/processor/entity.go
+++ b/pkg/watermark/processor/entity.go
@@ -49,8 +49,7 @@ func (w Watermark) Before(t time.Time) bool {
 // ProcessorEntitier defines what can be a processor. The Processor is the smallest unit where the watermark will
 // monotonically increase.
 type ProcessorEntitier interface {
-	GetID() string
-	BuildOTWatcherKey() string
+	GetName() string
 }
 
 // processorEntity implements ProcessorEntitier.
@@ -68,12 +67,7 @@ func NewProcessorEntity(name string) ProcessorEntitier {
 	}
 }
 
-// GetID returns the ID of the processor.
-func (p *processorEntity) GetID() string {
+// GetName returns the ID of the processor.
+func (p *processorEntity) GetName() string {
 	return p.name
 }
-
-// BuildOTWatcherKey builds the offset-timeline key name
-func (p *processorEntity) BuildOTWatcherKey() string {
-	return p.GetID()
-}
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index 319f63a4..c75e26d5 100644
--- a/pkg/watermark/publish/publisher.go
+++ b/pkg/watermark/publish/publisher.go
@@ -54,7 +54,7 @@ type publish struct {
 
 // NewPublish returns `Publish`.
 func NewPublish(ctx context.Context, processorEntity processor.ProcessorEntitier, watermarkStores store.WatermarkStorer, inputOpts ...PublishOption) Publisher {
-	log := logging.FromContext(ctx).With("entityID", processorEntity.GetID()).
+	log := logging.FromContext(ctx).With("entityID", processorEntity.GetName()).
 		With("otStore", watermarkStores.OffsetTimelineStore().GetStoreName()).
 		With("hbStore", watermarkStores.HeartbeatStore().GetStoreName())
 	log.Info("Creating a new watermark publisher")
@@ -108,7 +108,7 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 		return
 	}
 
-	var key = p.entity.BuildOTWatcherKey()
+	var key = p.entity.GetName()
 
 	// build value
 	var seq int64
@@ -145,16 +145,16 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 func (p *publish) loadLatestFromStore() processor.Watermark {
 	var (
 		timeWatermark = time.UnixMilli(-1)
-		key           = p.entity.BuildOTWatcherKey()
+		key           = p.entity.GetName()
 	)
 	byteValue, err := p.otStore.GetValue(p.ctx, key)
 	if err != nil {
-		p.log.Errorw("Unable to load latest watermark from ot store (failed to get value from ot store)", zap.String("OT", p.otStore.GetStoreName()), zap.String("processorEntity", p.entity.GetID()), zap.Error(err))
+		p.log.Errorw("Unable to load latest watermark from ot store (failed to get value from ot store)", zap.String("OT", p.otStore.GetStoreName()), zap.String("processorEntity", p.entity.GetName()), zap.Error(err))
 		return processor.Watermark(timeWatermark)
 	}
 	otValue, err := ot.DecodeToOTValue(byteValue)
 	if err != nil {
-		p.log.Errorw("Unable to load latest watermark from ot store (failed to decode ot value)", zap.String("OT", p.otStore.GetStoreName()), zap.String("processorEntity", p.entity.GetID()), zap.Error(err))
+		p.log.Errorw("Unable to load latest watermark from ot store (failed to decode ot value)", zap.String("OT", p.otStore.GetStoreName()), zap.String("processorEntity", p.entity.GetName()), zap.Error(err))
 		return processor.Watermark(timeWatermark)
 	}
 	timeWatermark = time.UnixMilli(otValue.Watermark)
@@ -175,7 +175,7 @@ func (p *publish) publishHeartbeat() {
 		case <-p.ctx.Done():
 			return
 		case <-ticker.C:
-			err := p.heartbeatStore.PutKV(p.ctx, p.entity.GetID(), []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err := p.heartbeatStore.PutKV(p.ctx, p.entity.GetName(), []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			if err != nil {
 				p.log.Errorw("Put to bucket failed", zap.String("bucket", p.heartbeatStore.GetStoreName()), zap.Error(err))
 			}
@@ -199,8 +199,8 @@ func (p *publish) Close() error {
 	//   - remove itself from heartbeat bucket
 
 	// clean up heartbeat bucket
-	if err := p.heartbeatStore.DeleteKey(p.ctx, p.entity.GetID()); err != nil {
-		p.log.Errorw("Failed to delete the key in the heartbeat bucket", zap.String("bucket", p.heartbeatStore.GetStoreName()), zap.String("key", p.entity.GetID()), zap.Error(err))
+	if err := p.heartbeatStore.DeleteKey(p.ctx, p.entity.GetName()); err != nil {
+		p.log.Errorw("Failed to delete the key in the heartbeat bucket", zap.String("bucket", p.heartbeatStore.GetStoreName()), zap.String("key", p.entity.GetName()), zap.Error(err))
 		return err
 	}
 	return nil
diff --git a/pkg/watermark/store/inmem/kv_store.go b/pkg/watermark/store/inmem/kv_store.go
index cbbd538a..2e6b13ae 100644
--- a/pkg/watermark/store/inmem/kv_store.go
+++ b/pkg/watermark/store/inmem/kv_store.go
@@ -141,5 +141,7 @@ func (kv *inMemStore) PutKV(_ context.Context, k string, v []byte) error {
 
 // Close closes the channel connection and clean up the bucket.
 func (kv *inMemStore) Close() {
+	kv.kvLock.Lock()
+	defer kv.kvLock.Unlock()
 	close(kv.kvEntryCh)
 }


Test patch:
diff --git a/pkg/reduce/reduce_test.go b/pkg/reduce/reduce_test.go
index a10ec907..a4324803 100644
--- a/pkg/reduce/reduce_test.go
+++ b/pkg/reduce/reduce_test.go
@@ -252,7 +252,7 @@ func TestReduceDataForward_Count(t *testing.T) {
 	// create from buffers
 	fromBuffer := simplebuffer.NewInMemoryBuffer(fromBufferName, fromBufferSize)
 
-	//create to buffers
+	// create to buffers
 	buffer := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
 	toBuffer := map[string]isb.BufferWriter{
 		toBufferName: buffer,
@@ -325,7 +325,7 @@ func TestReduceDataForward_Sum(t *testing.T) {
 	// create from buffers
 	fromBuffer := simplebuffer.NewInMemoryBuffer(fromBufferName, fromBufferSize)
 
-	//create to buffers
+	// create to buffers
 	buffer := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
 	toBuffer := map[string]isb.BufferWriter{
 		toBufferName: buffer,
@@ -398,7 +398,7 @@ func TestReduceDataForward_Max(t *testing.T) {
 	// create from buffers
 	fromBuffer := simplebuffer.NewInMemoryBuffer(fromBufferName, fromBufferSize)
 
-	//create to buffers
+	// create to buffers
 	buffer := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
 	toBuffer := map[string]isb.BufferWriter{
 		toBufferName: buffer,
@@ -471,7 +471,7 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 	// create from buffers
 	fromBuffer := simplebuffer.NewInMemoryBuffer(fromBufferName, fromBufferSize)
 
-	//create to buffers
+	// create to buffers
 	buffer := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
 	toBuffer := map[string]isb.BufferWriter{
 		toBufferName: buffer,
diff --git a/pkg/watermark/fetch/edge_fetcher_test.go b/pkg/watermark/fetch/edge_fetcher_test.go
index 9a4f6922..a3307d9d 100644
--- a/pkg/watermark/fetch/edge_fetcher_test.go
+++ b/pkg/watermark/fetch/edge_fetcher_test.go
@@ -40,9 +40,9 @@ func TestBuffer_GetWatermark(t *testing.T) {
 	otWatcher := noop.NewKVOpWatch()
 	processorManager := NewProcessorManager(ctx, store.BuildWatermarkStoreWatcher(hbWatcher, otWatcher))
 	var (
-		testPod0     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5, otWatcher)
-		testPod1     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod2"), 5, otWatcher)
-		testPod2     = NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod3"), 5, otWatcher)
+		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},
diff --git a/pkg/watermark/fetch/processor_manager_inmem_test.go b/pkg/watermark/fetch/processor_manager_inmem_test.go
index 240c1a29..28977f0f 100644
--- a/pkg/watermark/fetch/processor_manager_inmem_test.go
+++ b/pkg/watermark/fetch/processor_manager_inmem_test.go
@@ -182,27 +182,13 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	assert.True(t, allProcessors["p1"].IsActive())
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" has been deleted from vertex.Processors
-	// so "p1" will be considered as a new processors and a new offsetTimeline watcher for "p1" will be created
+	// so "p1" will be considered as a new processors with a new default offset timeline
 	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
 	p1 := testBuffer.processorManager.GetProcessor("p1")
-
 	assert.NotNil(t, p1)
 	assert.True(t, p1.IsActive())
 	assert.NotNil(t, p1.offsetTimeline)
-	// wait till the offsetTimeline has been populated
-	newP1head := p1.offsetTimeline.GetHeadOffset()
-	for newP1head == -1 {
-		select {
-		case <-ctx.Done():
-			t.Fatalf("expected head offset to not be equal to -1, %s", ctx.Err())
-		default:
-			time.Sleep(1 * time.Millisecond)
-			newP1head = p1.offsetTimeline.GetHeadOffset()
-		}
-	}
-	// because we keep the kv updates history in the in mem watermark
-	// the new watcher will read all the history data to create this new offsetTimeline
-	assert.Equal(t, int64(100), p1.offsetTimeline.GetHeadOffset())
+	assert.Equal(t, int64(-1), p1.offsetTimeline.GetHeadOffset())
 
 	// publish a new watermark 101
 	otValueByte, err = otValueToBytes(testOffset+1, epoch)
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index c75b3385..01ddb509 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -243,25 +243,13 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	assert.True(t, allProcessors["p1"].IsActive())
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" has been deleted from vertex.Processors
-	// so "p1" will be considered as a new processors and a new offsetTimeline watcher for "p1" will be created
+	// so "p1" will be considered as a new processors with a new default offset timeline
 	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+1, 10) }))
 	p1 := testBuffer.processorManager.GetProcessor("p1")
 	assert.NotNil(t, p1)
 	assert.True(t, p1.IsActive())
 	assert.NotNil(t, p1.offsetTimeline)
-	// wait till the offsetTimeline has been populated
-	newP1head := p1.offsetTimeline.GetHeadOffset()
-	// because the bucket hasn't been cleaned up, the new watcher will read all the history data to create this new offsetTimeline
-	for newP1head != 102 {
-		fmt.Println(p1.offsetTimeline.Dump())
-		select {
-		case <-ctx.Done():
-			t.Fatalf("expected head offset to be 102, %s", ctx.Err())
-		default:
-			time.Sleep(1 * time.Millisecond)
-			newP1head = p1.offsetTimeline.GetHeadOffset()
-		}
-	}
+	assert.Equal(t, int64(-1), p1.offsetTimeline.GetHeadOffset())
 
 	// publish a new watermark 103
 	otValueByte, err = otValueToBytes(testOffset+3, epoch)
@@ -310,11 +298,11 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	// added 103 in the previous steps for p1, so the head should be 103 after resume
 	assert.Equal(t, int64(103), p1.offsetTimeline.GetHeadOffset())
 
-	for allProcessors["p1"].offsetTimeline.Dump() != "[1651161660000:103] -> [1651161600300:102] -> [1651161600200:101] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1] -> [-1:-1]" {
+	for allProcessors["p1"].offsetTimeline.Dump() != "[1651161660000:103] -> [-1:-1] -> [-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("expected p1 has the offset timeline [1651161660000:103] -> [1651161600300:102] -> [1651161600200:101] -> [-1:-1]..., got %s: %s", allProcessors["p1"].offsetTimeline.Dump(), ctx.Err())
+				t.Fatalf("expected p1 has the offset timeline [1651161660000:103] -> [-1:-1] -> [-1:-1] -> [-1:-1]..., got %s: %s", allProcessors["p1"].offsetTimeline.Dump(), ctx.Err())
 			}
 		default:
 			time.Sleep(1 * time.Millisecond)
diff --git a/pkg/watermark/fetch/processor_to_fetch_test.go b/pkg/watermark/fetch/processor_to_fetch_test.go
index c2887609..6ec32e02 100644
--- a/pkg/watermark/fetch/processor_to_fetch_test.go
+++ b/pkg/watermark/fetch/processor_to_fetch_test.go
@@ -22,7 +22,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
@@ -30,7 +29,7 @@ import (
 
 func TestFromProcessor_setStatus(t *testing.T) {
 	var ctx = context.Background()
-	p := NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5, noop.NewKVOpWatch())
+	p := NewProcessorToFetch(ctx, processor.NewProcessorEntity("testPod1"), 5)
 	p.setStatus(_inactive)
 	assert.Equal(t, _inactive, p.status)
 }
diff --git a/pkg/watermark/processor/entity_test.go b/pkg/watermark/processor/entity_test.go
index 5d1934d6..0443b3d1 100644
--- a/pkg/watermark/processor/entity_test.go
+++ b/pkg/watermark/processor/entity_test.go
@@ -26,7 +26,7 @@ import (
 
 func TestEntity(t *testing.T) {
 	e := NewProcessorEntity("pod0")
-	assert.Equal(t, "pod0", e.GetID())
+	assert.Equal(t, "pod0", e.GetName())
 }
 
 func ExampleWatermark_String() {
diff --git a/pkg/watermark/publish/publisher_inmem_test.go b/pkg/watermark/publish/publisher_inmem_test.go
index f2699c66..1a580d1f 100644
--- a/pkg/watermark/publish/publisher_inmem_test.go
+++ b/pkg/watermark/publish/publisher_inmem_test.go
@@ -70,7 +70,7 @@ func TestPublisherWithSharedOTBuckets_InMem(t *testing.T) {
 
 	_ = p.Close()
 
-	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetID())
+	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetName())
 	assert.Equal(t, fmt.Errorf("key publisherTestPod1 not found"), err)
 
 }
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index 65824a6c..128807c9 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -94,6 +94,6 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 
 	_ = p.Close()
 
-	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetID())
+	_, err = p.heartbeatStore.GetValue(ctx, publishEntity.GetName())
 	assert.Equal(t, nats.ErrConnectionClosed, err)
 }