Repository: numaproj/numaflow
Pull request #307 (2022-11-03 04:27:06)

feat: update watermark offset bucket implementation
fix: #301 

chromevoid: > thinking how can we do a comparison in before and after performance , any ideas?

run the same pipeline which led us to this issue?
vigith: > run the same pipeline which led us to this issue?

It won't show any performance improvement. It will show that the KV entries are less.

By performance, i meant whether we can see any perceivable improvement is throughput, speed, CPU, etc.

Patch:
diff --git a/config/advanced-install/namespaced-controller-wo-crds.yaml b/config/advanced-install/namespaced-controller-wo-crds.yaml
index a2d9a524..71a8c5ac 100644
--- a/config/advanced-install/namespaced-controller-wo-crds.yaml
+++ b/config/advanced-install/namespaced-controller-wo-crds.yaml
@@ -144,7 +144,7 @@ data:
             maxAckPending: 20000
           otBucket:
             maxValueSize: 0
-            history: 1
+            history: 5
             ttl: 3h
             maxBytes: 0
             replicas: 3
diff --git a/config/base/controller-manager/numaflow-controller-config.yaml b/config/base/controller-manager/numaflow-controller-config.yaml
index fa77996f..4caf1527 100644
--- a/config/base/controller-manager/numaflow-controller-config.yaml
+++ b/config/base/controller-manager/numaflow-controller-config.yaml
@@ -60,7 +60,7 @@ data:
             maxAckPending: 20000
           otBucket:
             maxValueSize: 0
-            history: 1
+            history: 5
             ttl: 3h
             maxBytes: 0
             replicas: 3
diff --git a/config/install.yaml b/config/install.yaml
index b1a16a9e..b76d3a89 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -9274,7 +9274,7 @@ data:
             maxAckPending: 20000
           otBucket:
             maxValueSize: 0
-            history: 1
+            history: 5
             ttl: 3h
             maxBytes: 0
             replicas: 3
diff --git a/config/namespace-install.yaml b/config/namespace-install.yaml
index 384e2e5c..df11587d 100644
--- a/config/namespace-install.yaml
+++ b/config/namespace-install.yaml
@@ -9185,7 +9185,7 @@ data:
             maxAckPending: 20000
           otBucket:
             maxValueSize: 0
-            history: 1
+            history: 5
             ttl: 3h
             maxBytes: 0
             replicas: 3
diff --git a/docs/numaflow-controller-config.yaml b/docs/numaflow-controller-config.yaml
index 9832559c..8ae46ece 100644
--- a/docs/numaflow-controller-config.yaml
+++ b/docs/numaflow-controller-config.yaml
@@ -60,7 +60,7 @@ data:
             maxAckPending: 20000
           otBucket:
             maxValueSize: 0
-            history: 1
+            history: 5
             ttl: 72h
             maxBytes: 0
             replicas: 3
diff --git a/pkg/watermark/fetch/offset_timeline.go b/pkg/watermark/fetch/offset_timeline.go
index 9bae1c5c..2b803d3c 100644
--- a/pkg/watermark/fetch/offset_timeline.go
+++ b/pkg/watermark/fetch/offset_timeline.go
@@ -44,20 +44,20 @@ type OffsetTimeline struct {
 func NewOffsetTimeline(ctx context.Context, c int) *OffsetTimeline {
 	// Initialize a new empty watermarks DLL with nil values of the size capacity.
 	// This is to avoid length check: when a new element is added, the tail element will be deleted.
-	ot := OffsetTimeline{
+	offsetTimeline := OffsetTimeline{
 		ctx:      ctx,
 		capacity: c,
 		log:      logging.FromContext(ctx),
 	}
 
 	for i := 0; i < c; i++ {
-		ot.watermarks.PushBack(OffsetWatermark{
+		offsetTimeline.watermarks.PushBack(OffsetWatermark{
 			watermark: -1,
 			offset:    -1,
 		})
 	}
 
-	return &ot
+	return &offsetTimeline
 }
 
 // OffsetWatermark stores the maximum offset for the given event time
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index dfddb8f1..3e3f0e4d 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -39,7 +39,7 @@ type ProcessorManager struct {
 	otWatcher store.WatermarkKVWatcher
 	// heartbeat just tracks the heartbeat of each processing unit. we use it to mark a processing unit's status (e.g, inactive)
 	heartbeat *ProcessorHeartbeat
-	// processors has reference to the actual processing unit (ProcessorEntity) which includes offset timeline which is
+	// processors has reference to the actual processing unit (ProcessorEntitier) which includes offset timeline which is
 	// used for tracking watermark.
 	processors map[string]*ProcessorToFetch
 	lock       sync.RWMutex
diff --git a/pkg/watermark/fetch/processor_to_fetch.go b/pkg/watermark/fetch/processor_to_fetch.go
index e909ce07..b8346839 100644
--- a/pkg/watermark/fetch/processor_to_fetch.go
+++ b/pkg/watermark/fetch/processor_to_fetch.go
@@ -18,10 +18,10 @@ package fetch
 
 import (
 	"context"
-	"encoding/binary"
 	"fmt"
 	"sync"
 
+	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"go.uber.org/zap"
 
 	"github.com/numaproj/numaflow/pkg/shared/logging"
@@ -134,21 +134,19 @@ func (p *ProcessorToFetch) startTimeLineWatcher() {
 			}
 			switch value.Operation() {
 			case store.KVPut:
-				epoch, skip, err := p.entity.ParseOTWatcherKey(value.Key())
-				if err != nil {
-					p.log.Errorw("Unable to convert value.PartitionID() to int64", zap.String("received", value.Key()), zap.Error(err))
+				if value.Key() != p.entity.BuildOTWatcherKey() {
 					continue
 				}
-				// if skip is set to true, it means the key update we received is for a different processor
-				if skip {
+				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
 				}
-				uint64Value := binary.LittleEndian.Uint64(value.Value())
 				p.offsetTimeline.Put(OffsetWatermark{
-					watermark: epoch,
-					offset:    int64(uint64Value),
+					watermark: otValue.Watermark,
+					offset:    otValue.Offset,
 				})
-				p.log.Debugw("TimelineWatcher- Updates", zap.String("bucket", p.otWatcher.GetKVName()), zap.Int64("epoch", epoch), zap.Uint64("value", uint64Value))
+				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.
diff --git a/pkg/watermark/ot/ot.go b/pkg/watermark/ot/ot.go
new file mode 100644
index 00000000..65426886
--- /dev/null
+++ b/pkg/watermark/ot/ot.go
@@ -0,0 +1,34 @@
+// Package ot represents the offset-timeline pair and its corresponding encoder and decoder.
+package ot
+
+import (
+	"bytes"
+	"encoding/binary"
+)
+
+// Value is used in the JetStream offset timeline bucket as the value for the given processor entity key.
+type Value struct {
+	Offset    int64
+	Watermark int64
+}
+
+// EncodeToBytes encodes a Value object into byte array.
+func (v Value) EncodeToBytes() ([]byte, error) {
+	buf := new(bytes.Buffer)
+	err := binary.Write(buf, binary.LittleEndian, v)
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+// DecodeToOTValue decodes the given byte array into a Value object.
+func DecodeToOTValue(b []byte) (Value, error) {
+	var v Value
+	buf := bytes.NewReader(b)
+	err := binary.Read(buf, binary.LittleEndian, &v)
+	if err != nil {
+		return Value{}, err
+	}
+	return v, nil
+}
diff --git a/pkg/watermark/processor/entity.go b/pkg/watermark/processor/entity.go
index b4421201..1c41cc15 100644
--- a/pkg/watermark/processor/entity.go
+++ b/pkg/watermark/processor/entity.go
@@ -20,13 +20,10 @@ Package processor is the smallest processor entity for which the watermark will
 package processor
 
 import (
-	"fmt"
-	"strconv"
-	"strings"
 	"time"
 )
 
-// Watermark is the monotonically increasing watermark. It is tightly coupled with ProcessorEntity as
+// Watermark is the monotonically increasing watermark. It is tightly coupled with ProcessorEntitier as
 // the processor is responsible for monotonically increasing Watermark for that processor.
 // NOTE: today we support only second progression of watermark, we need to support millisecond too.
 type Watermark time.Time
@@ -49,79 +46,34 @@ func (w Watermark) Before(t time.Time) bool {
 	return time.Time(w).Before(t)
 }
 
-type entityOptions struct {
-	keySeparator string
-}
-
-// EntityOption set options for FromVertex.
-type EntityOption func(*entityOptions)
-
 // 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(Watermark) string
-	ParseOTWatcherKey(string) (int64, bool, error)
+	BuildOTWatcherKey() string
 }
 
-// ProcessorEntity implements ProcessorEntitier.
-type ProcessorEntity struct {
+// processorEntity implements ProcessorEntitier.
+type processorEntity struct {
 	// name is the name of the entity
 	name string
-	opts *entityOptions
 }
 
-var _ ProcessorEntitier = (*ProcessorEntity)(nil)
-
-// _defaultKeySeparator is the key separate used in the offset timeline kv.
-// NOTE: we can only use `_` as the separator, JetStream will not let any other special character.
-//
-//	Perhaps we can encode the key using base64, but it will have a performance hit.
-const _defaultKeySeparator = "_"
+var _ ProcessorEntitier = (*processorEntity)(nil)
 
-// NewProcessorEntity returns a new `ProcessorEntity`.
-func NewProcessorEntity(name string, inputOpts ...EntityOption) *ProcessorEntity {
-	opts := &entityOptions{
-		keySeparator: _defaultKeySeparator,
-	}
-	for _, opt := range inputOpts {
-		opt(opts)
-	}
-	return &ProcessorEntity{
+// NewProcessorEntity returns a new `ProcessorEntitier`.
+func NewProcessorEntity(name string) ProcessorEntitier {
+	return &processorEntity{
 		name: name,
-		opts: opts,
 	}
 }
 
 // GetID returns the ID of the processor.
-func (p *ProcessorEntity) GetID() string {
+func (p *processorEntity) GetID() string {
 	return p.name
 }
 
 // BuildOTWatcherKey builds the offset-timeline key name
-func (p *ProcessorEntity) BuildOTWatcherKey(watermark Watermark) string {
-	return fmt.Sprintf("%s%s%d", p.GetID(), p.opts.keySeparator, watermark.UnixMilli())
-}
-
-// ParseOTWatcherKey parses the key of the KeyValue OT watcher and returns the epoch, a boolean to indicate
-// whether the record can be skipped and error if any.
-// NOTE: _defaultKeySeparator has constraints, please make sure we will not end up with multiple values
-func (p *ProcessorEntity) ParseOTWatcherKey(key string) (epoch int64, skip bool, err error) {
-	name, epochStr, err := p.splitKey(key)
-	if err != nil {
-		return 0, false, err
-	}
-	// skip if not this processor
-	skip = name != p.GetID()
-	epoch, err = strconv.ParseInt(epochStr, 10, 64)
-
-	return epoch, skip, err
-}
-
-func (p *ProcessorEntity) splitKey(key string) (string, string, error) {
-	split := strings.Split(key, p.opts.keySeparator)
-	if len(split) != 2 {
-		return "", "", fmt.Errorf("key=%s when split using %s, did not have 2 outputs=%v", key, p.opts.keySeparator, split)
-	}
-	return split[0], split[1], nil
+func (p *processorEntity) BuildOTWatcherKey() string {
+	return p.GetID()
 }
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index d0050855..e041108f 100644
--- a/pkg/watermark/publish/publisher.go
+++ b/pkg/watermark/publish/publisher.go
@@ -18,21 +18,18 @@ package publish
 
 import (
 	"context"
-	"encoding/binary"
-	"errors"
 	"fmt"
-	"math"
 	"time"
 
-	"github.com/nats-io/nats.go"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
+	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
 	"go.uber.org/zap"
 )
 
-// Publisher interface defines how to publish Watermark for a ProcessorEntity.
+// Publisher interface defines how to publish Watermark for a ProcessorEntitier.
 type Publisher interface {
 	// PublishWatermark publishes the watermark.
 	PublishWatermark(processor.Watermark, isb.Offset)
@@ -88,7 +85,7 @@ func NewPublish(ctx context.Context, processorEntity processor.ProcessorEntitier
 	return p
 }
 
-// initialSetup inserts the default values as the ProcessorEntity starts emitting watermarks.
+// initialSetup inserts the default values as the ProcessorEntitier starts emitting watermarks.
 func (p *publish) initialSetup() {
 	p.headWatermark = p.loadLatestFromStore()
 }
@@ -111,11 +108,9 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 		return
 	}
 
-	// build key (we may or may not share the same OT bucket)
-	var key = p.entity.BuildOTWatcherKey(wm)
+	var key = p.entity.BuildOTWatcherKey()
 
-	// build value (offset)
-	value := make([]byte, 8)
+	// 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.
@@ -123,7 +118,14 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 	} else {
 		seq, _ = offset.Sequence()
 	}
-	binary.LittleEndian.PutUint64(value, uint64(seq))
+	var otValue = ot.Value{
+		Offset:    seq,
+		Watermark: wm.UnixMilli(),
+	}
+	value, err := otValue.EncodeToBytes()
+	if err != nil {
+		p.log.Errorw("Unable to publish watermark", zap.String("HB", p.heartbeatStore.GetStoreName()), zap.String("OT", p.otStore.GetStoreName()), zap.String("key", key), zap.Error(err))
+	}
 
 	for {
 		err := p.otStore.PutKV(p.ctx, key, value)
@@ -140,26 +142,21 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 // loadLatestFromStore loads the latest watermark stored in the watermark store.
 // TODO: how to repopulate if the processing unit is down for a really long time?
 func (p *publish) loadLatestFromStore() processor.Watermark {
-	// TODO: this is too much.
-	var watermarks = p.getAllOTKeysFromBucket()
-	var latestWatermark int64 = math.MinInt64
-
-	// skip all entries that do not match the prefix if we are sharing bucket
-
-	for _, wm := range watermarks {
-		epoch, skip, err := p.entity.ParseOTWatcherKey(wm)
-		if skip {
-			continue
-		}
-		if err != nil {
-			p.log.Errorw("Invalid epoch time string", zap.Error(err))
-			continue
-		}
-		if latestWatermark < epoch {
-			latestWatermark = epoch
-		}
+	var (
+		timeWatermark = time.UnixMilli(-1)
+		key           = p.entity.BuildOTWatcherKey()
+	)
+	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))
+		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))
+		return processor.Watermark(timeWatermark)
 	}
-	var timeWatermark = time.UnixMilli(latestWatermark)
+	timeWatermark = time.UnixMilli(otValue.Watermark)
 	return processor.Watermark(timeWatermark)
 }
 
@@ -202,11 +199,3 @@ func (p *publish) StopPublisher() {
 	p.otStore.Close()
 	p.heartbeatStore.Close()
 }
-
-func (p *publish) getAllOTKeysFromBucket() []string {
-	keys, err := p.otStore.GetAllKeys(p.ctx)
-	if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
-		p.log.Fatalw("Failed to get the keys", zap.String("bucket", p.heartbeatStore.GetStoreName()), zap.Error(err))
-	}
-	return keys
-}
diff --git a/pkg/watermark/store/jetstream/kv_watch.go b/pkg/watermark/store/jetstream/kv_watch.go
index ed71b821..a0fe00ae 100644
--- a/pkg/watermark/store/jetstream/kv_watch.go
+++ b/pkg/watermark/store/jetstream/kv_watch.go
@@ -113,10 +113,10 @@ func (k kvEntry) Operation() store.KVWatchOp {
 
 // Watch watches the key-value store (aka bucket).
 func (k *jetStreamWatch) Watch(ctx context.Context) (<-chan store.WatermarkKVEntry, <-chan struct{}) {
-	kvWatcher, err := k.kv.WatchAll()
+	kvWatcher, err := k.kv.WatchAll(nats.IncludeHistory())
 	for err != nil {
 		k.log.Errorw("WatchAll failed", zap.String("watcher", k.GetKVName()), zap.Error(err))
-		kvWatcher, err = k.kv.WatchAll()
+		kvWatcher, err = k.kv.WatchAll(nats.IncludeHistory())
 		time.Sleep(100 * time.Millisecond)
 	}
 


Test patch:
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 6e69f070..07ad7910 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -19,7 +19,6 @@ package pnf
 import (
 	"context"
 	"encoding/json"
-	"math"
 	"strings"
 	"testing"
 	"time"
@@ -193,7 +192,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			expected: []bool{false, true},
 			wmExpected: map[string]int64{
 				"buffer1": 120000,
-				"buffer2": math.MinInt64,
+				"buffer2": -1,
 			},
 		},
 		{
@@ -222,8 +221,8 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			pf:       createProcessAndForward(ctx, "test-drop-all", pbqManager, toBuffers3),
 			expected: []bool{true, true},
 			wmExpected: map[string]int64{
-				"buffer1": math.MinInt64,
-				"buffer2": math.MinInt64,
+				"buffer1": -1,
+				"buffer2": -1,
 			},
 		},
 	}
diff --git a/pkg/watermark/fetch/processor_manager_inmem_test.go b/pkg/watermark/fetch/processor_manager_inmem_test.go
index 75a6f36b..b6db5793 100644
--- a/pkg/watermark/fetch/processor_manager_inmem_test.go
+++ b/pkg/watermark/fetch/processor_manager_inmem_test.go
@@ -18,13 +18,13 @@ package fetch
 
 import (
 	"context"
-	"encoding/binary"
 	"fmt"
 	"strconv"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
 	"github.com/stretchr/testify/assert"
 
@@ -32,6 +32,15 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/store"
 )
 
+func otValueToBytes(offset int64, watermark int64) ([]byte, error) {
+	otValue := ot.Value{
+		Offset:    offset,
+		Watermark: watermark,
+	}
+	otValueByte, err := otValue.EncodeToBytes()
+	return otValueByte, err
+}
+
 func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	var (
 		err          error
@@ -45,23 +54,23 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 	defer cancel()
-	hb, hbWatcherCh, err := inmem.NewKVInMemKVStore(ctx, pipelineName, hbBucketName)
+	hbStore, hbWatcherCh, err := inmem.NewKVInMemKVStore(ctx, pipelineName, hbBucketName)
 	assert.NoError(t, err)
-	defer hb.Close()
-	ot, otWatcherCh, err := inmem.NewKVInMemKVStore(ctx, pipelineName, otBucketName)
+	defer hbStore.Close()
+	otStore, otWatcherCh, err := inmem.NewKVInMemKVStore(ctx, pipelineName, otBucketName)
 	assert.NoError(t, err)
-	defer ot.Close()
+	defer otStore.Close()
 
-	b := make([]byte, 8)
-	binary.LittleEndian.PutUint64(b, uint64(testOffset))
-	// this key format is meant for non-separate OT watcher
-	err = ot.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	otValueByte, err := otValueToBytes(testOffset, epoch)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
 	assert.NoError(t, err)
 
 	epoch += 60000
-	binary.LittleEndian.PutUint64(b, uint64(testOffset+5))
-	// this key format is meant for non-separate OT watcher
-	err = ot.PutKV(ctx, fmt.Sprintf("%s%s%d", "p2", "_", epoch), b)
+
+	otValueByte, err = otValueToBytes(testOffset+5, epoch)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p2", otValueByte)
 	assert.NoError(t, err)
 
 	hbWatcher, err := inmem.NewInMemWatch(ctx, "testFetch", keyspace+"_PROCESSORS", hbWatcherCh)
@@ -77,11 +86,11 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 		defer wg.Done()
 		var err error
 		for i := 0; i < 3; i++ {
-			err = hb.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err = hbStore.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
-		err = hb.DeleteKey(ctx, "p1")
+		err = hbStore.DeleteKey(ctx, "p1")
 		assert.NoError(t, err)
 	}()
 
@@ -91,7 +100,7 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 		defer wg.Done()
 		var err error
 		for i := 0; i < 20; i++ {
-			err = hb.PutKV(ctx, "p2", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err = hbStore.PutKV(ctx, "p2", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
@@ -151,7 +160,7 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 		defer wg.Done()
 		var err error
 		for i := 0; i < 5; i++ {
-			err = hb.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err = hbStore.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
@@ -197,8 +206,9 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 	assert.Equal(t, int64(100), p1.offsetTimeline.GetHeadOffset())
 
 	// publish a new watermark 101
-	binary.LittleEndian.PutUint64(b, uint64(testOffset+1))
-	err = ot.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	otValueByte, err = otValueToBytes(testOffset+1, epoch)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
 	assert.NoError(t, err)
 
 	// "p1" becomes inactive after 5 loops
@@ -222,7 +232,7 @@ func TestFetcherWithSameOTBucket_InMem(t *testing.T) {
 		defer wg.Done()
 		var err error
 		for i := 0; i < 3; i++ {
-			err = hb.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
+			err = hbStore.PutKV(ctx, "p1", []byte(fmt.Sprintf("%d", time.Now().Unix())))
 			assert.NoError(t, err)
 			time.Sleep(time.Duration(testVertex.opts.podHeartbeatRate) * time.Second)
 		}
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index 5aa0b723..9e534d12 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -20,7 +20,6 @@ package fetch
 
 import (
 	"context"
-	"encoding/binary"
 	"fmt"
 	"strconv"
 	"sync"
@@ -75,7 +74,7 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		Bucket:       keyspace + "_OT",
 		Description:  "",
 		MaxValueSize: 0,
-		History:      0,
+		History:      2,
 		TTL:          0,
 		MaxBytes:     0,
 		Storage:      nats.MemoryStorage,
@@ -99,13 +98,28 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	defer otStore.Close()
 
 	// put values into otStore
-	b := make([]byte, 8)
-	binary.LittleEndian.PutUint64(b, uint64(testOffset))
-	err = otStore.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+
+	// this first entry should not be in the offset timeline because we set the ot bucket history to 2
+	otValueByte, err := otValueToBytes(testOffset, epoch+100)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
+	assert.NoError(t, err)
+
+	otValueByte, err = otValueToBytes(testOffset+1, epoch+200)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
+	assert.NoError(t, err)
+
+	otValueByte, err = otValueToBytes(testOffset+2, epoch+300)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
 	assert.NoError(t, err)
+
 	epoch += 60000
-	binary.LittleEndian.PutUint64(b, uint64(testOffset+5))
-	err = otStore.PutKV(ctx, fmt.Sprintf("%s%s%d", "p2", "_", epoch), b)
+
+	otValueByte, err = otValueToBytes(testOffset+5, epoch)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p2", otValueByte)
 	assert.NoError(t, err)
 
 	// create watchers for heartbeat and offset timeline
@@ -154,6 +168,18 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		}
 	}
 
+	for allProcessors["p1"].offsetTimeline.Dump() != "[1651161600300:102] -> [1651161600200:101] -> [-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 [1651161600300:102] -> [1651161600200:101] -> [-1:-1]..., got %s: %s", allProcessors["p1"].offsetTimeline.Dump(), ctx.Err())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
+
 	assert.True(t, allProcessors["p1"].IsActive())
 	assert.True(t, allProcessors["p2"].IsActive())
 
@@ -181,7 +207,7 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	assert.True(t, allProcessors["p1"].IsDeleted())
 	assert.True(t, allProcessors["p2"].IsActive())
 	// "p1" should be deleted after this GetWatermark offset=101
-	// because "p1" offsetTimeline's head offset=100, which is < inputOffset 103
+	// because "p1" offsetTimeline's head offset=102, which is < inputOffset 103
 	_ = testBuffer.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(testOffset+3, 10) }))
 	allProcessors = testBuffer.processorManager.GetAllProcessors()
 	assert.Equal(t, 1, len(allProcessors))
@@ -226,21 +252,22 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 	assert.NotNil(t, p1.offsetTimeline)
 	// wait till the offsetTimeline has been populated
 	newP1head := p1.offsetTimeline.GetHeadOffset()
-	for newP1head == -1 {
+	// 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 not be equal to -1, %s", ctx.Err())
+			t.Fatalf("expected head offset to be 102, %s", ctx.Err())
 		default:
 			time.Sleep(1 * time.Millisecond)
 			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
-	assert.Equal(t, int64(100), p1.offsetTimeline.GetHeadOffset())
 
-	// publish a new watermark 101
-	binary.LittleEndian.PutUint64(b, uint64(testOffset+1))
-	err = otStore.PutKV(ctx, fmt.Sprintf("%s%s%d", "p1", "_", epoch), b)
+	// publish a new watermark 103
+	otValueByte, err = otValueToBytes(testOffset+3, epoch)
+	assert.NoError(t, err)
+	err = otStore.PutKV(ctx, "p1", otValueByte)
 	assert.NoError(t, err)
 
 	// "p1" becomes inactive after 5 loops
@@ -281,8 +308,20 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 		}
 	}
 
-	// added 101 in the previous steps for p1, so the head should be 101 after resume
-	assert.Equal(t, int64(101), p1.offsetTimeline.GetHeadOffset())
+	// 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]" {
+		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())
+			}
+		default:
+			time.Sleep(1 * time.Millisecond)
+			allProcessors = testBuffer.processorManager.GetAllProcessors()
+		}
+	}
 
 	wg.Wait()
 	cancel()
diff --git a/pkg/watermark/ot/ot_test.go b/pkg/watermark/ot/ot_test.go
new file mode 100644
index 00000000..b2d39a55
--- /dev/null
+++ b/pkg/watermark/ot/ot_test.go
@@ -0,0 +1,155 @@
+package ot
+
+import (
+	"bytes"
+	"encoding/binary"
+	"reflect"
+	"testing"
+)
+
+func TestDecodeToOTValue(t *testing.T) {
+	type args struct {
+		b []byte
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    Value
+		wantErr bool
+	}{
+		{
+			name: "decode_success_using_ot_value",
+			args: args{
+				b: func() []byte {
+					v := Value{
+						Offset:    100,
+						Watermark: 1667495100000,
+					}
+					buf := new(bytes.Buffer)
+					_ = binary.Write(buf, binary.LittleEndian, v)
+					return buf.Bytes()
+				}(),
+			},
+			want: Value{
+				Offset:    100,
+				Watermark: 1667495100000,
+			},
+			wantErr: false,
+		},
+		{
+			name: "decode_failure_using_1_field_struct",
+			args: args{
+				b: func() []byte {
+					v := struct {
+						Test int64
+					}{
+						Test: 100,
+					}
+					buf := new(bytes.Buffer)
+					_ = binary.Write(buf, binary.LittleEndian, v)
+					return buf.Bytes()
+				}(),
+			},
+			want:    Value{},
+			wantErr: true,
+		},
+		{
+			name: "decode_success_using_2_field_struct",
+			args: args{
+				b: func() []byte {
+					v := struct {
+						Test0 int64
+						Test1 int64
+					}{
+						Test0: 100,
+						Test1: 1667495100000,
+					}
+					buf := new(bytes.Buffer)
+					_ = binary.Write(buf, binary.LittleEndian, v)
+					return buf.Bytes()
+				}(),
+			},
+			want: Value{
+				Offset:    100,
+				Watermark: 1667495100000,
+			},
+			wantErr: false,
+		},
+		{
+			name: "decode_success_using_3_field_struct",
+			args: args{
+				b: func() []byte {
+					v := struct {
+						Test0 int64
+						Test1 int64
+						Test2 int64 // should be ignored
+					}{
+						Test0: 100,
+						Test1: 1667495100000,
+						Test2: 200,
+					}
+					buf := new(bytes.Buffer)
+					_ = binary.Write(buf, binary.LittleEndian, v)
+					return buf.Bytes()
+				}(),
+			},
+			want: Value{
+				Offset:    100,
+				Watermark: 1667495100000,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := DecodeToOTValue(tt.args.b)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("DecodeToOTValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("DecodeToOTValue() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestOTValue_EncodeToBytes(t *testing.T) {
+	// bytes.Buffer Write doesn't return err, so err is always nil
+	type fields struct {
+		Offset    int64
+		Watermark int64
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		want    []byte
+		wantErr bool
+	}{
+		{
+			name: "encode_success",
+			fields: fields{
+				Offset:    100,
+				Watermark: 1667495100000,
+			},
+			want:    []byte{100, 0, 0, 0, 0, 0, 0, 0, 96, 254, 115, 62, 132, 1, 0, 0},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := Value{
+				Offset:    tt.fields.Offset,
+				Watermark: tt.fields.Watermark,
+			}
+			got, err := v.EncodeToBytes()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("EncodeToBytes() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("EncodeToBytes() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/pkg/watermark/processor/entity_test.go b/pkg/watermark/processor/entity_test.go
index 243021f2..5d1934d6 100644
--- a/pkg/watermark/processor/entity_test.go
+++ b/pkg/watermark/processor/entity_test.go
@@ -41,98 +41,3 @@ func TestExampleWatermarkUnix(t *testing.T) {
 	wm := Watermark(time.UnixMilli(1651129200000))
 	assert.Equal(t, int64(1651129200000), wm.UnixMilli())
 }
-
-func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
-	tests := []struct {
-		name      string
-		p         *ProcessorEntity
-		arg       string
-		wantEpoch int64
-		wantSkip  bool
-		wantErr   assert.ErrorAssertionFunc
-	}{
-		{
-			name:      "bad_without_split_butSkip",
-			p:         NewProcessorEntity("test1"),
-			arg:       _defaultKeySeparator + "1234", // name is missing
-			wantEpoch: 1234,
-			wantSkip:  true,
-			wantErr:   assert.NoError,
-		},
-		{
-			name:      "bad_without_split_missing_separator",
-			p:         NewProcessorEntity("test1"),
-			arg:       "1234",
-			wantEpoch: 0,
-			wantSkip:  false,
-			wantErr:   assert.Error,
-		},
-		{
-			name:      "good_without_split",
-			p:         NewProcessorEntity("test1"),
-			arg:       "test1" + _defaultKeySeparator + "1234",
-			wantEpoch: 1234,
-			wantSkip:  false,
-			wantErr:   assert.NoError,
-		},
-		{
-			name:      "good_without_split_skip",
-			p:         NewProcessorEntity("test1"),
-			arg:       "test-not-this" + _defaultKeySeparator + "1234",
-			wantEpoch: 1234,
-			wantSkip:  true,
-			wantErr:   assert.NoError,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			p := tt.p
-			gotEpoch, gotSkip, err := p.ParseOTWatcherKey(tt.arg)
-			if !tt.wantErr(t, err, fmt.Sprintf("ParseOTWatcherKey(%v)", tt.arg)) {
-				return
-			}
-			assert.Equalf(t, tt.wantEpoch, gotEpoch, "ParseOTWatcherKey(%v)", tt.arg)
-			assert.Equalf(t, tt.wantSkip, gotSkip, "ParseOTWatcherKey(%v)", tt.arg)
-		})
-	}
-}
-
-func TestProcessorEntity_splitKey(t *testing.T) {
-	tests := []struct {
-		name    string
-		arg     string
-		p       *ProcessorEntity
-		want    string
-		want1   string
-		wantErr assert.ErrorAssertionFunc
-	}{
-		{
-			name:    "bad_expected_name",
-			p:       NewProcessorEntity("test1"),
-			arg:     "1234",
-			want:    "",
-			want1:   "",
-			wantErr: assert.Error,
-		},
-		{
-			name:    "good_same_bucket",
-			p:       NewProcessorEntity("test1"),
-			arg:     "p1" + _defaultKeySeparator + "1234",
-			want:    "p1",
-			want1:   "1234",
-			wantErr: assert.NoError,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			p := tt.p
-			got, got1, err := p.splitKey(tt.arg)
-			if !tt.wantErr(t, err, fmt.Sprintf("splitKey(%v)", tt.arg)) {
-				return
-			}
-			assert.Equalf(t, tt.want, got, "splitKey(%v)", tt.arg)
-			assert.Equalf(t, tt.want1, got1, "splitKey(%v)", tt.arg)
-		})
-	}
-}
diff --git a/pkg/watermark/publish/publisher_inmem_test.go b/pkg/watermark/publish/publisher_inmem_test.go
index 4fbca615..2dcd2a19 100644
--- a/pkg/watermark/publish/publisher_inmem_test.go
+++ b/pkg/watermark/publish/publisher_inmem_test.go
@@ -58,8 +58,9 @@ func TestPublisherWithSharedOTBuckets_InMem(t *testing.T) {
 	// publish a stale watermark (offset doesn't matter)
 	p.PublishWatermark(processor.Watermark(time.UnixMilli(epoch-120000).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
-	keys := p.getAllOTKeysFromBucket()
-	assert.Equal(t, []string{"publisherTestPod1_1651161600000", "publisherTestPod1_1651161660000", "publisherTestPod1_1651161720000"}, keys)
+	keys, err := p.otStore.GetAllKeys(p.ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"publisherTestPod1"}, keys)
 
 	wm := p.loadLatestFromStore()
 	assert.Equal(t, processor.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), wm.String())
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index d84bc036..b4d70b3d 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -82,8 +82,9 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	// publish a stale watermark (offset doesn't matter)
 	p.PublishWatermark(processor.Watermark(time.UnixMilli(epoch-120000).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
-	keys := p.getAllOTKeysFromBucket()
-	assert.Equal(t, []string{"publisherTestPod1_1651161600000", "publisherTestPod1_1651161660000", "publisherTestPod1_1651161720000"}, keys)
+	keys, err := p.otStore.GetAllKeys(p.ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"publisherTestPod1"}, keys)
 
 	wm := p.loadLatestFromStore()
 	assert.Equal(t, processor.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), wm.String())