Repository: numaproj/numaflow
Pull request #584 (2023-03-03 03:24:54)

chore: refactor current OT Value to Watermark Barrier.
The current ot.Value has been refactored (rearranged) to look like a WMB.


Patch:
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index 8bc58fb5..1a257d29 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -36,8 +36,8 @@ import (
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // InterStepDataForward forwards the data from previous step to the current step via inter-step buffer.
@@ -233,12 +233,12 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	}
 	concurrentUDFProcessingStart := time.Now()
 
-	var processorWM processor.Watermark
+	var processorWM wmb.Watermark
 	if isdf.opts.vertexType == dfv1.VertexTypeSource {
 		// for source vertex, the udf is the source data transformer.
 		// in this case, we assign time.UnixMilli(-1) to processorWM.
 		// source data transformer applies filtering and assigns event time to source data, which doesn't require watermarks.
-		processorWM = processor.Watermark(time.UnixMilli(-1))
+		processorWM = wmb.Watermark(time.UnixMilli(-1))
 	} else {
 		// fetch watermark if available
 		// TODO: make it async (concurrent and wait later)
diff --git a/pkg/isb/message.go b/pkg/isb/message.go
index 5f495e55..5933f03e 100644
--- a/pkg/isb/message.go
+++ b/pkg/isb/message.go
@@ -28,8 +28,11 @@ const (
 	WMB                          // Watermark Barrier
 )
 
-func (mt MessageKind) String() string {
+func (mta MessageKind) String() string {
+	var mt MessageKind
 	switch mt {
+	case 0:
+		return "Empty"
 	case mt & Data:
 		return "Data"
 	case mt & WMB:
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index 587d97be..6874fd42 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -38,8 +38,8 @@ import (
 	"github.com/numaproj/numaflow/pkg/reduce/pbq"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // ProcessAndForward reads messages from pbq, invokes udf using grpc, forwards the results to ISB, and then publishes
@@ -107,7 +107,7 @@ func (p *ProcessAndForward) Forward(ctx context.Context) error {
 		metrics.LabelVertexReplicaIndex: strconv.Itoa(int(p.vertexReplica)),
 	}).Observe(float64(time.Since(startTime).Microseconds()))
 
-	processorWM := processor.Watermark(p.PartitionID.Start)
+	processorWM := wmb.Watermark(p.PartitionID.Start)
 
 	messagesToStep := p.whereToStep()
 
@@ -247,7 +247,7 @@ func (p *ProcessAndForward) writeToBuffer(ctx context.Context, bufferID string,
 }
 
 // publishWM publishes the watermark to each edge.
-func (p *ProcessAndForward) publishWM(wm processor.Watermark, writeOffsets map[string][]isb.Offset) {
+func (p *ProcessAndForward) publishWM(wm wmb.Watermark, writeOffsets map[string][]isb.Offset) {
 	// activeWatermarkBuffers records the buffers that the publisher has published
 	// a watermark in this batch processing cycle.
 	// it's used to determine which buffers should receive an idle watermark.
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index 3b9a1ed2..893c6eef 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -44,8 +44,8 @@ import (
 	"github.com/numaproj/numaflow/pkg/reduce/pbq"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 	"github.com/numaproj/numaflow/pkg/window"
 	"github.com/numaproj/numaflow/pkg/window/keyed"
 )
@@ -153,7 +153,7 @@ func (rl *ReadLoop) Process(ctx context.Context, messages []*isb.ReadMessage) {
 	// since the watermark will be same for all the messages in the batch
 	// we can invoke remove windows only once per batch
 	var closedWindows []window.AlignedKeyedWindower
-	wm := processor.Watermark(successfullyWrittenMessages[0].Watermark)
+	wm := wmb.Watermark(successfullyWrittenMessages[0].Watermark)
 
 	closedWindows = rl.windower.RemoveWindows(time.Time(wm))
 
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index bf15c389..41c79138 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -39,6 +39,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 var log = logging.NewLogger()
@@ -241,7 +242,7 @@ func (mg *memgen) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	// use the first event time as watermark to make it conservative
 	nanos, _ := msgs[0].ReadOffset.Sequence()
 	// remove the nanosecond precision
-	mg.sourcePublishWM.PublishWatermark(processor.Watermark(time.Unix(0, nanos)), nil) // Source publisher does not care about the offset
+	mg.sourcePublishWM.PublishWatermark(wmb.Watermark(time.Unix(0, nanos)), nil) // Source publisher does not care about the offset
 }
 
 // Ack acknowledges an array of offset.
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 4298f80c..6a7986c5 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -40,6 +40,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 type httpSource struct {
@@ -247,7 +248,7 @@ func (h *httpSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	}
 	if len(msgs) > 0 && !oldest.IsZero() {
 		h.logger.Debugf("Publishing watermark %v to source", oldest)
-		h.sourcePublishWM.PublishWatermark(processor.Watermark(oldest), nil) // Source publisher does not care about the offset
+		h.sourcePublishWM.PublishWatermark(wmb.Watermark(oldest), nil) // Source publisher does not care about the offset
 	}
 }
 
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index fb746bb6..61142b17 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -39,6 +39,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 type KafkaSource struct {
@@ -151,7 +152,7 @@ func (r *KafkaSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 	}
 	for p, t := range oldestTimestamps {
 		publisher := r.loadSourceWatermarkPublisher(p)
-		publisher.PublishWatermark(processor.Watermark(t), nil) // Source publisher does not care about the offset
+		publisher.PublishWatermark(wmb.Watermark(t), nil) // Source publisher does not care about the offset
 	}
 }
 
diff --git a/pkg/sources/nats/nats.go b/pkg/sources/nats/nats.go
index b83ed694..a18d92f7 100644
--- a/pkg/sources/nats/nats.go
+++ b/pkg/sources/nats/nats.go
@@ -36,6 +36,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 type natsSource struct {
@@ -243,7 +244,7 @@ func (ns *natsSource) PublishSourceWatermarks(msgs []*isb.ReadMessage) {
 		}
 	}
 	if len(msgs) > 0 && !oldest.IsZero() {
-		ns.sourcePublishWM.PublishWatermark(processor.Watermark(oldest), nil) // Source publisher does not care about the offset
+		ns.sourcePublishWM.PublishWatermark(wmb.Watermark(oldest), nil) // Source publisher does not care about the offset
 	}
 }
 
diff --git a/pkg/watermark/fetch/edge_fetcher.go b/pkg/watermark/fetch/edge_fetcher.go
index 950e67c6..79c35d55 100644
--- a/pkg/watermark/fetch/edge_fetcher.go
+++ b/pkg/watermark/fetch/edge_fetcher.go
@@ -27,8 +27,8 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // edgeFetcher is a fetcher between two vertices.
@@ -60,7 +60,7 @@ func NewEdgeFetcher(ctx context.Context, bufferName string, storeWatcher store.W
 //   - We don't use this function in the regular pods in the vertex.
 //   - UX only uses GetHeadWatermark, so the `p.IsDeleted()` check in the GetWatermark never happens.
 //     Meaning, in the UX (daemon service) we never delete any processor.
-func (e *edgeFetcher) GetHeadWatermark() processor.Watermark {
+func (e *edgeFetcher) GetHeadWatermark() wmb.Watermark {
 	var debugString strings.Builder
 	var headOffset int64 = math.MinInt64
 	var epoch int64 = math.MaxInt64
@@ -81,17 +81,17 @@ func (e *edgeFetcher) GetHeadWatermark() processor.Watermark {
 	e.log.Debugf("GetHeadWatermark: %s", debugString.String())
 	if epoch == math.MaxInt64 {
 		// Use -1 as default watermark value to indicate there is no valid watermark yet.
-		return processor.Watermark(time.UnixMilli(-1))
+		return wmb.Watermark(time.UnixMilli(-1))
 	}
-	return processor.Watermark(time.UnixMilli(epoch))
+	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
 // GetWatermark gets the smallest timestamp for the given offset
-func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) processor.Watermark {
+func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) wmb.Watermark {
 	var offset, err = inputOffset.Sequence()
 	if err != nil {
 		e.log.Errorw("Unable to get offset from isb.Offset.Sequence()", zap.Error(err))
-		return processor.Watermark(time.Unix(-1, 0))
+		return wmb.Watermark(time.Unix(-1, 0))
 	}
 	var debugString strings.Builder
 	var epoch int64 = math.MaxInt64
@@ -115,7 +115,7 @@ func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) processor.Watermark {
 	}
 	e.log.Debugf("%s[%s] get watermark for offset %d: %+v", debugString.String(), e.bufferName, offset, epoch)
 
-	return processor.Watermark(time.UnixMilli(epoch))
+	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
 // Close function closes the watchers.
diff --git a/pkg/watermark/fetch/interface.go b/pkg/watermark/fetch/interface.go
index 8aba4f3b..706e53a4 100644
--- a/pkg/watermark/fetch/interface.go
+++ b/pkg/watermark/fetch/interface.go
@@ -20,14 +20,14 @@ import (
 	"io"
 
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // Fetcher fetches watermark data from Vn-1 vertex.
 type Fetcher interface {
 	io.Closer
 	// GetWatermark returns the inorder monotonically increasing watermark of the edge connected to Vn-1.
-	GetWatermark(offset isb.Offset) processor.Watermark
+	GetWatermark(offset isb.Offset) wmb.Watermark
 	// GetHeadWatermark returns the latest watermark based on the head offset
-	GetHeadWatermark() processor.Watermark
+	GetHeadWatermark() wmb.Watermark
 }
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index 24cfa24d..2bc52e74 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -24,13 +24,13 @@ 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"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // ProcessorManager manages the point of view of Vn-1 from Vn vertex processors (or source processor). The code is running on Vn vertex.
@@ -188,7 +188,7 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 				// value is epoch
 				intValue, convErr := strconv.Atoi(string(value.Value()))
 				if convErr != nil {
-					v.log.Errorw("Unable to convert intValue.Value() to int64", zap.Error(convErr))
+					v.log.Errorw("Unable to convert intValue.WMB() to int64", zap.Error(convErr))
 				} else {
 					// insert the last seen timestamp. we use this to figure whether this processor entity is inactive.
 					v.heartbeat.Put(value.Key(), int64(intValue))
@@ -244,19 +244,19 @@ func (v *ProcessorManager) startTimeLineWatcher() {
 					v.log.Errorw("Unable to find the processor", zap.String("processorEntity", value.Key()))
 					continue
 				}
-				otValue, err := ot.DecodeToOTValue(value.Value())
+				otValue, err := wmb.DecodeToWMB(value.Value())
 				if err != nil {
 					v.log.Errorw("Unable to decode the value", zap.String("processorEntity", p.entity.GetName()), zap.Error(err))
 					continue
 				}
 				if otValue.Idle {
 					var processors = v.GetAllProcessors()
-					for processorName, processor := range processors {
+					for processorName, _processor := range processors {
 						// skip the idle processor itself, only use other processors as reference
 						if processorName != value.Key() {
 							// in any other Vn-1 processor's offset timeline, we can replace the idle watermark
 							// with any watermark whose watermark is <= otValue.Watermark
-							referredWatermarkOffset := processor.offsetTimeline.GetReferredWatermark(otValue.Watermark)
+							referredWatermarkOffset := _processor.offsetTimeline.GetReferredWatermark(otValue.Watermark)
 							// if the referred watermark is empty, skip
 							if referredWatermarkOffset.watermark != -1 {
 								p.offsetTimeline.PutIdle(referredWatermarkOffset)
diff --git a/pkg/watermark/fetch/source_fetcher.go b/pkg/watermark/fetch/source_fetcher.go
index 92344dd1..5a640cc8 100644
--- a/pkg/watermark/fetch/source_fetcher.go
+++ b/pkg/watermark/fetch/source_fetcher.go
@@ -25,8 +25,8 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // sourceFetcher is a fetcher on source buffers.
@@ -53,7 +53,7 @@ func NewSourceFetcher(ctx context.Context, sourceBufferName string, storeWatcher
 }
 
 // GetHeadWatermark returns the latest watermark of all the processors.
-func (e *sourceFetcher) GetHeadWatermark() processor.Watermark {
+func (e *sourceFetcher) GetHeadWatermark() wmb.Watermark {
 	var epoch int64 = math.MinInt64
 	for _, p := range e.processorManager.GetAllProcessors() {
 		if !p.IsActive() {
@@ -65,14 +65,14 @@ func (e *sourceFetcher) GetHeadWatermark() processor.Watermark {
 	}
 	if epoch == math.MinInt64 {
 		// Use -1 as default watermark value to indicate there is no valid watermark yet.
-		return processor.Watermark(time.UnixMilli(-1))
+		return wmb.Watermark(time.UnixMilli(-1))
 	}
-	return processor.Watermark(time.UnixMilli(epoch))
+	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
 // GetWatermark returns the lowest of the latest watermark of all the processors,
 // it ignores the input offset.
-func (e *sourceFetcher) GetWatermark(_ isb.Offset) processor.Watermark {
+func (e *sourceFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
 	var epoch int64 = math.MaxInt64
 	for _, p := range e.processorManager.GetAllProcessors() {
 		if !p.IsActive() {
@@ -85,7 +85,7 @@ func (e *sourceFetcher) GetWatermark(_ isb.Offset) processor.Watermark {
 	if epoch == math.MaxInt64 {
 		epoch = -1
 	}
-	return processor.Watermark(time.UnixMilli(epoch))
+	return wmb.Watermark(time.UnixMilli(epoch))
 }
 
 // Close function closes the watchers.
diff --git a/pkg/watermark/generic/noop.go b/pkg/watermark/generic/noop.go
index 53bbf356..572c2b42 100644
--- a/pkg/watermark/generic/noop.go
+++ b/pkg/watermark/generic/noop.go
@@ -20,8 +20,8 @@ import (
 	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // NoOpWMProgressor is a no-op watermark progressor. As the name suggests, it does not do anything, no watermark is
@@ -38,26 +38,26 @@ func NewNoOpWMProgressor() *NoOpWMProgressor {
 }
 
 // GetWatermark returns the default watermark.
-func (n NoOpWMProgressor) GetWatermark(_ isb.Offset) processor.Watermark {
-	return processor.Watermark{}
+func (n NoOpWMProgressor) GetWatermark(_ isb.Offset) wmb.Watermark {
+	return wmb.Watermark{}
 }
 
 // PublishWatermark does a no-op watermark publish.
-func (n NoOpWMProgressor) PublishWatermark(_ processor.Watermark, _ isb.Offset) {
+func (n NoOpWMProgressor) PublishWatermark(_ wmb.Watermark, _ isb.Offset) {
 }
 
 // PublishIdleWatermark does a no-op idle watermark publish.
-func (n NoOpWMProgressor) PublishIdleWatermark(processor.Watermark) {
+func (n NoOpWMProgressor) PublishIdleWatermark(wmb.Watermark) {
 }
 
 // GetLatestWatermark returns the default watermark as the latest watermark.
-func (n NoOpWMProgressor) GetLatestWatermark() processor.Watermark {
-	return processor.Watermark{}
+func (n NoOpWMProgressor) GetLatestWatermark() wmb.Watermark {
+	return wmb.Watermark{}
 }
 
 // GetHeadWatermark returns the default head watermark.
-func (n NoOpWMProgressor) GetHeadWatermark() processor.Watermark {
-	return processor.Watermark{}
+func (n NoOpWMProgressor) GetHeadWatermark() wmb.Watermark {
+	return wmb.Watermark{}
 }
 
 // Close stops the no-op progressor.
diff --git a/pkg/watermark/processor/entity.go b/pkg/watermark/processor/entity.go
index 7dd786e9..a7952e58 100644
--- a/pkg/watermark/processor/entity.go
+++ b/pkg/watermark/processor/entity.go
@@ -19,33 +19,6 @@ Package processor is the smallest processor entity for which the watermark will
 */
 package processor
 
-import (
-	"time"
-)
-
-// 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
-
-func (w Watermark) String() string {
-	var location, _ = time.LoadLocation("UTC")
-	var t = time.Time(w).In(location)
-	return t.Format(time.RFC3339Nano)
-}
-
-func (w Watermark) UnixMilli() int64 {
-	return time.Time(w).UnixMilli()
-}
-
-func (w Watermark) After(t time.Time) bool {
-	return time.Time(w).After(t)
-}
-
-func (w Watermark) Before(t time.Time) bool {
-	return time.Time(w).Before(t)
-}
-
 // ProcessorEntitier defines what can be a processor. The Processor is the smallest unit where the watermark will
 // monotonically increase.
 type ProcessorEntitier interface {
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index ddd656eb..7ff7c9bc 100644
--- a/pkg/watermark/publish/publisher.go
+++ b/pkg/watermark/publish/publisher.go
@@ -26,20 +26,20 @@ import (
 
 	"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"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 // Publisher interface defines how to publish Watermark for a ProcessorEntitier.
 type Publisher interface {
 	io.Closer
 	// PublishWatermark publishes the watermark.
-	PublishWatermark(processor.Watermark, isb.Offset)
+	PublishWatermark(wmb.Watermark, isb.Offset)
 	// PublishIdleWatermark publishes the idle watermark.
-	PublishIdleWatermark(wm processor.Watermark)
+	PublishIdleWatermark(wm wmb.Watermark)
 	// GetLatestWatermark returns the latest published watermark.
-	GetLatestWatermark() processor.Watermark
+	GetLatestWatermark() wmb.Watermark
 }
 
 // publish publishes the watermark for a processor entity.
@@ -51,7 +51,7 @@ type publish struct {
 	// osStore uses millisecond as the time unit for the value
 	otStore       store.WatermarkKVStorer
 	log           *zap.SugaredLogger
-	headWatermark processor.Watermark
+	headWatermark wmb.Watermark
 	opts          *publishOptions
 }
 
@@ -95,7 +95,7 @@ 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) {
+func (p *publish) PublishWatermark(wm wmb.Watermark, offset isb.Offset) {
 	validWM, skipWM := p.validateWatermark(wm)
 	if skipWM {
 		return
@@ -111,7 +111,7 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 	} else {
 		seq, _ = offset.Sequence()
 	}
-	var otValue = ot.Value{
+	var otValue = wmb.WMB{
 		Offset:    seq,
 		Watermark: validWM.UnixMilli(),
 	}
@@ -135,9 +135,9 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 
 // 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) {
+func (p *publish) validateWatermark(wm wmb.Watermark) (wmb.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))
+		wm = wmb.Watermark(time.Time(wm).Add(-p.opts.delay))
 	}
 	// update p.headWatermark only if wm > p.headWatermark
 	if wm.After(time.Time(p.headWatermark)) {
@@ -145,22 +145,22 @@ func (p *publish) validateWatermark(wm processor.Watermark) (processor.Watermark
 		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
+		return wmb.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 wmb.Watermark{}, true
 	}
 	return wm, false
 }
 
 // PublishIdleWatermark publishes the idle watermark and will retry until it can succeed.
-func (p *publish) PublishIdleWatermark(wm processor.Watermark) {
+func (p *publish) PublishIdleWatermark(wm wmb.Watermark) {
 	var key = p.entity.GetName()
 	validWM, skipWM := p.validateWatermark(wm)
 	if skipWM {
 		return
 	}
-	var otValue = ot.Value{
+	var otValue = wmb.WMB{
 		Offset:    0,
 		Watermark: validWM.UnixMilli(),
 		Idle:      true,
@@ -185,7 +185,7 @@ func (p *publish) PublishIdleWatermark(wm processor.Watermark) {
 
 // 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 {
+func (p *publish) loadLatestFromStore() wmb.Watermark {
 	var (
 		timeWatermark = time.UnixMilli(-1)
 		key           = p.entity.GetName()
@@ -193,20 +193,20 @@ func (p *publish) loadLatestFromStore() processor.Watermark {
 	byteValue, err := p.otStore.GetValue(p.ctx, key)
 	// could happen during boot up
 	if err != nil {
-		p.log.Warnw("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)
+		p.log.Warnw("Unable to load latest watermark from wmb store (failed to get value from wmb store)", zap.String("OT", p.otStore.GetStoreName()), zap.String("processorEntity", p.entity.GetName()), zap.Error(err))
+		return wmb.Watermark(timeWatermark)
 	}
-	otValue, err := ot.DecodeToOTValue(byteValue)
+	otValue, err := wmb.DecodeToWMB(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.GetName()), zap.Error(err))
-		return processor.Watermark(timeWatermark)
+		p.log.Errorw("Unable to load latest watermark from wmb store (failed to decode wmb value)", zap.String("OT", p.otStore.GetStoreName()), zap.String("processorEntity", p.entity.GetName()), zap.Error(err))
+		return wmb.Watermark(timeWatermark)
 	}
 	timeWatermark = time.UnixMilli(otValue.Watermark)
-	return processor.Watermark(timeWatermark)
+	return wmb.Watermark(timeWatermark)
 }
 
 // GetLatestWatermark returns the latest watermark for that processor.
-func (p *publish) GetLatestWatermark() processor.Watermark {
+func (p *publish) GetLatestWatermark() wmb.Watermark {
 	return p.headWatermark
 }
 
diff --git a/pkg/watermark/wmb/watermark.go b/pkg/watermark/wmb/watermark.go
new file mode 100644
index 00000000..457fa63d
--- /dev/null
+++ b/pkg/watermark/wmb/watermark.go
@@ -0,0 +1,42 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wmb
+
+import "time"
+
+// 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
+
+func (w Watermark) String() string {
+	var location, _ = time.LoadLocation("UTC")
+	var t = time.Time(w).In(location)
+	return t.Format(time.RFC3339Nano)
+}
+
+func (w Watermark) UnixMilli() int64 {
+	return time.Time(w).UnixMilli()
+}
+
+func (w Watermark) After(t time.Time) bool {
+	return time.Time(w).After(t)
+}
+
+func (w Watermark) Before(t time.Time) bool {
+	return time.Time(w).Before(t)
+}
diff --git a/pkg/watermark/ot/ot.go b/pkg/watermark/wmb/wmb.go
similarity index 52%
rename from pkg/watermark/ot/ot.go
rename to pkg/watermark/wmb/wmb.go
index 468109d7..f7e7d7bd 100644
--- a/pkg/watermark/ot/ot.go
+++ b/pkg/watermark/wmb/wmb.go
@@ -14,41 +14,47 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package ot represents the offset-timeline pair and its corresponding encoder and decoder.
-package ot
+// Package wmb represents the offset-timeline pair and its corresponding encoder and decoder.
+package wmb
 
 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
+// WMB is used in the KV offset timeline bucket as the value for the given processor entity key.
+type WMB struct {
 	// Idle is set to true if the given processor entity hasn't published anything
 	// to the offset timeline bucket in a batch processing cycle.
 	// Idle is used to signal an idle watermark.
 	Idle bool
+	// Offset is the monotonically increasing index/offset of the buffer (buffer is the physical representation
+	// of the partition of the edge).
+	Offset int64
+	// Watermark is tightly coupled with the offset and will be monotonically increasing for a given ProcessorEntity
+	// as the offset increases.
+	// When it is idling (Idle==true), for a given offset, the watermark can monotonically increase without offset
+	// increasing.
+	Watermark int64
 }
 
-// EncodeToBytes encodes a Value object into byte array.
-func (v Value) EncodeToBytes() ([]byte, error) {
+// EncodeToBytes encodes a WMB object into byte array.
+func (w WMB) EncodeToBytes() ([]byte, error) {
 	buf := new(bytes.Buffer)
-	err := binary.Write(buf, binary.LittleEndian, v)
+	err := binary.Write(buf, binary.LittleEndian, w)
 	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
+// DecodeToWMB decodes the given byte array into a WMB object.
+func DecodeToWMB(b []byte) (WMB, error) {
+	var v WMB
 	buf := bytes.NewReader(b)
 	err := binary.Read(buf, binary.LittleEndian, &v)
 	if err != nil {
-		return Value{}, err
+		return WMB{}, err
 	}
 	return v, nil
 }


Test patch:
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index ec8a924f..8e12a47e 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -31,11 +31,11 @@ import (
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	udfapplier "github.com/numaproj/numaflow/pkg/udf/function"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	wmstore "github.com/numaproj/numaflow/pkg/watermark/store"
 	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
@@ -64,13 +64,13 @@ func (t testForwardFetcher) Close() error {
 
 // GetWatermark uses current time as the watermark because we want to make sure
 // the test publisher is publishing watermark
-func (t testForwardFetcher) GetWatermark(_ isb.Offset) processor.Watermark {
-	return processor.Watermark(testSourceWatermark)
+func (t testForwardFetcher) GetWatermark(_ isb.Offset) wmb.Watermark {
+	return wmb.Watermark(testSourceWatermark)
 }
 
-func (t testForwardFetcher) GetHeadWatermark() processor.Watermark {
+func (t testForwardFetcher) GetHeadWatermark() wmb.Watermark {
 	// won't be used
-	return processor.Watermark{}
+	return wmb.Watermark{}
 }
 
 type myForwardTest struct {
@@ -351,7 +351,7 @@ func TestNewInterStepDataForwardToOneStep(t *testing.T) {
 	// so len(otKeys) should always be 1
 	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
 	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ := ot.DecodeToOTValue(otValue1)
+	otDecode1, _ := wmb.DecodeToWMB(otValue1)
 	assert.False(t, otDecode1.Idle)
 
 	wg.Add(1)
@@ -368,7 +368,7 @@ func TestNewInterStepDataForwardToOneStep(t *testing.T) {
 	// so len(otKeys) should always be 1
 	otKeys2, _ := otStores["to2"].GetAllKeys(ctx)
 	otValue2, _ := otStores["to2"].GetValue(ctx, otKeys2[0])
-	otDecode2, _ := ot.DecodeToOTValue(otValue2)
+	otDecode2, _ := wmb.DecodeToWMB(otValue2)
 	assert.True(t, otDecode2.Idle)
 
 	// stop will cancel the contexts and therefore the forwarder stops without waiting
@@ -444,7 +444,7 @@ func TestNewInterStepDataForward_dropAll(t *testing.T) {
 	// so len(otKeys) should always be 1
 	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
 	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ := ot.DecodeToOTValue(otValue1)
+	otDecode1, _ := wmb.DecodeToWMB(otValue1)
 	assert.True(t, otDecode1.Idle)
 
 	wg.Add(1)
@@ -461,7 +461,7 @@ func TestNewInterStepDataForward_dropAll(t *testing.T) {
 	// so len(otKeys) should always be 1
 	otKeys2, _ := otStores["to2"].GetAllKeys(ctx)
 	otValue2, _ := otStores["to2"].GetValue(ctx, otKeys2[0])
-	otDecode2, _ := ot.DecodeToOTValue(otValue2)
+	otDecode2, _ := wmb.DecodeToWMB(otValue2)
 	assert.True(t, otDecode2.Idle)
 
 	// since this is a dropping WhereTo, the buffer can never be full
@@ -539,7 +539,7 @@ func TestNewInterStepData_forwardToAll(t *testing.T) {
 	// so len(otKeys) should always be 1
 	otKeys1, _ := otStores["to1"].GetAllKeys(ctx)
 	otValue1, _ := otStores["to1"].GetValue(ctx, otKeys1[0])
-	otDecode1, _ := ot.DecodeToOTValue(otValue1)
+	otDecode1, _ := wmb.DecodeToWMB(otValue1)
 	assert.False(t, otDecode1.Idle)
 
 	wg.Add(1)
@@ -556,7 +556,7 @@ func TestNewInterStepData_forwardToAll(t *testing.T) {
 	// so len(otKeys) should always be 1
 	otKeys2, _ := otStores["to2"].GetAllKeys(ctx)
 	otValue2, _ := otStores["to2"].GetValue(ctx, otKeys2[0])
-	otDecode2, _ := ot.DecodeToOTValue(otValue2)
+	otDecode2, _ := wmb.DecodeToWMB(otValue2)
 	assert.False(t, otDecode2.Idle)
 
 	f.Stop()
diff --git a/pkg/isb/serde_test.go b/pkg/isb/serde_test.go
index d26f26f5..3bdfb482 100644
--- a/pkg/isb/serde_test.go
+++ b/pkg/isb/serde_test.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package isb
 
 import (
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index 124b2cd9..c0838f06 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -39,6 +39,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	wmstore "github.com/numaproj/numaflow/pkg/watermark/store"
 	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
 )
 
@@ -67,36 +68,36 @@ var nonKeyedVertex = &dfv1.VertexInstance{
 }
 
 type EventTypeWMProgressor struct {
-	watermarks map[string]processor.Watermark
+	watermarks map[string]wmb.Watermark
 	m          sync.Mutex
 }
 
-func (e *EventTypeWMProgressor) PublishWatermark(watermark processor.Watermark, offset isb.Offset) {
+func (e *EventTypeWMProgressor) PublishWatermark(watermark wmb.Watermark, offset isb.Offset) {
 	e.m.Lock()
 	defer e.m.Unlock()
 	e.watermarks[offset.String()] = watermark
 }
 
-func (e *EventTypeWMProgressor) PublishIdleWatermark(processor.Watermark) {
+func (e *EventTypeWMProgressor) PublishIdleWatermark(wmb.Watermark) {
 	// TODO
 }
 
-func (e *EventTypeWMProgressor) GetLatestWatermark() processor.Watermark {
-	return processor.Watermark{}
+func (e *EventTypeWMProgressor) GetLatestWatermark() wmb.Watermark {
+	return wmb.Watermark{}
 }
 
 func (e *EventTypeWMProgressor) Close() error {
 	return nil
 }
 
-func (e *EventTypeWMProgressor) GetWatermark(offset isb.Offset) processor.Watermark {
+func (e *EventTypeWMProgressor) GetWatermark(offset isb.Offset) wmb.Watermark {
 	e.m.Lock()
 	defer e.m.Unlock()
 	return e.watermarks[offset.String()]
 }
 
-func (e *EventTypeWMProgressor) GetHeadWatermark() processor.Watermark {
-	return processor.Watermark{}
+func (e *EventTypeWMProgressor) GetHeadWatermark() wmb.Watermark {
+	return wmb.Watermark{}
 }
 
 // PayloadForTest is a dummy payload for testing.
@@ -227,7 +228,7 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	to := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
 
 	wmpublisher := &EventTypeWMProgressor{
-		watermarks: make(map[string]processor.Watermark),
+		watermarks: make(map[string]wmb.Watermark),
 	}
 
 	// keep on writing <count> messages every 1 second for the supplied key
@@ -834,7 +835,7 @@ func writeMessages(ctx context.Context, count int, key string, fromBuffer *simpl
 		// write the messages to fromBuffer, so that it will be available for consuming
 		offsets, _ := fromBuffer.Write(ctx, messages)
 		for _, offset := range offsets {
-			publish.PublishWatermark(processor.Watermark(publishTime), offset)
+			publish.PublishWatermark(wmb.Watermark(publishTime), offset)
 		}
 	}
 }
@@ -869,7 +870,7 @@ func publishMessages(ctx context.Context, startTime int, messages []int, testDur
 				if count >= batchSize {
 					offsets, _ := fromBuffer.Write(ctx, inputMsgs)
 					if len(offsets) > 0 {
-						publish.PublishWatermark(processor.Watermark(time.UnixMilli(int64(et))), offsets[len(offsets)-1])
+						publish.PublishWatermark(wmb.Watermark(time.UnixMilli(int64(et))), offsets[len(offsets)-1])
 					}
 					count = 0
 				}
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 8a9a3ed2..fd26276a 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -31,10 +31,10 @@ import (
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/store/memory"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/watermark/generic"
-	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 
 	"github.com/golang/mock/gomock"
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
@@ -196,7 +196,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 		pf         ProcessAndForward
 		otStores   map[string]wmstore.WatermarkKVStorer
 		expected   []bool
-		wmExpected map[string]ot.Value
+		wmExpected map[string]wmb.WMB
 	}{
 		{
 			name: "test-forward-one",
@@ -209,7 +209,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			pf:       pf1,
 			otStores: otStores1,
 			expected: []bool{false, true},
-			wmExpected: map[string]ot.Value{
+			wmExpected: map[string]wmb.WMB{
 				"buffer1": {
 					Offset:    0,
 					Watermark: int64(60000),
@@ -233,7 +233,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			pf:       pf2,
 			otStores: otStores2,
 			expected: []bool{false, false},
-			wmExpected: map[string]ot.Value{
+			wmExpected: map[string]wmb.WMB{
 				"buffer1": {
 					Offset:    0,
 					Watermark: int64(60000),
@@ -257,7 +257,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 			pf:       pf3,
 			otStores: otStores3,
 			expected: []bool{true, true},
-			wmExpected: map[string]ot.Value{
+			wmExpected: map[string]wmb.WMB{
 				"buffer1": {
 					Offset:    0,
 					Watermark: int64(60000),
@@ -285,7 +285,7 @@ func TestProcessAndForward_Forward(t *testing.T) {
 				otKeys, _ := value.otStores[bufferName].GetAllKeys(ctx)
 				for _, otKey := range otKeys {
 					otValue, _ := value.otStores[bufferName].GetValue(ctx, otKey)
-					ot, _ := ot.DecodeToOTValue(otValue)
+					ot, _ := wmb.DecodeToWMB(otValue)
 					assert.Equal(t, ot, value.wmExpected[bufferName])
 				}
 			}
diff --git a/pkg/watermark/fetch/edge_fetcher_test.go b/pkg/watermark/fetch/edge_fetcher_test.go
index 85153d3e..6eae0809 100644
--- a/pkg/watermark/fetch/edge_fetcher_test.go
+++ b/pkg/watermark/fetch/edge_fetcher_test.go
@@ -26,6 +26,7 @@ import (
 	"go.uber.org/zap/zaptest"
 
 	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
@@ -140,7 +141,7 @@ func TestBuffer_GetWatermark(t *testing.T) {
 				log:              zaptest.NewLogger(t).Sugar(),
 			}
 			if got := b.GetWatermark(isb.SimpleStringOffset(func() string { return strconv.FormatInt(tt.args.offset, 10) })); time.Time(got).In(location) != time.UnixMilli(tt.want).In(location) {
-				t.Errorf("GetWatermark() = %v, want %v", got, processor.Watermark(time.UnixMilli(tt.want)))
+				t.Errorf("GetWatermark() = %v, want %v", got, wmb.Watermark(time.UnixMilli(tt.want)))
 			}
 			// this will always be 14 because the timeline has been populated ahead of time
 			assert.Equal(t, time.Time(b.GetHeadWatermark()).In(location), time.UnixMilli(14).In(location))
diff --git a/pkg/watermark/fetch/processor_manager_inmem_test.go b/pkg/watermark/fetch/processor_manager_inmem_test.go
index 14cea8a0..3556593c 100644
--- a/pkg/watermark/fetch/processor_manager_inmem_test.go
+++ b/pkg/watermark/fetch/processor_manager_inmem_test.go
@@ -24,16 +24,17 @@ import (
 	"testing"
 	"time"
 
-	"github.com/numaproj/numaflow/pkg/watermark/ot"
-	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
 )
 
 func otValueToBytes(offset int64, watermark int64, idle bool) ([]byte, error) {
-	otValue := ot.Value{
+	otValue := wmb.WMB{
 		Offset:    offset,
 		Watermark: watermark,
 		Idle:      idle,
diff --git a/pkg/watermark/fetch/processor_manager_test.go b/pkg/watermark/fetch/processor_manager_test.go
index 67ed1873..b4ed3ac3 100644
--- a/pkg/watermark/fetch/processor_manager_test.go
+++ b/pkg/watermark/fetch/processor_manager_test.go
@@ -100,7 +100,7 @@ func TestFetcherWithSameOTBucket(t *testing.T) {
 
 	// put values into otStore
 
-	// this first entry should not be in the offset timeline because we set the ot bucket history to 2
+	// this first entry should not be in the offset timeline because we set the wmb bucket history to 2
 	otValueByte, err := otValueToBytes(testOffset, epoch+100, false)
 	assert.NoError(t, err)
 	err = otStore.PutKV(ctx, "p1", otValueByte)
diff --git a/pkg/watermark/processor/entity_test.go b/pkg/watermark/processor/entity_test.go
index 0443b3d1..b1e2b370 100644
--- a/pkg/watermark/processor/entity_test.go
+++ b/pkg/watermark/processor/entity_test.go
@@ -22,6 +22,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
 )
 
 func TestEntity(t *testing.T) {
@@ -31,13 +33,13 @@ func TestEntity(t *testing.T) {
 
 func ExampleWatermark_String() {
 	location, _ := time.LoadLocation("UTC")
-	wm := Watermark(time.Unix(1651129200, 0).In(location))
+	wm := wmb.Watermark(time.Unix(1651129200, 0).In(location))
 	fmt.Println(wm)
 	// output:
 	// 2022-04-28T07:00:00Z
 }
 
 func TestExampleWatermarkUnix(t *testing.T) {
-	wm := Watermark(time.UnixMilli(1651129200000))
+	wm := wmb.Watermark(time.UnixMilli(1651129200000))
 	assert.Equal(t, int64(1651129200000), wm.UnixMilli())
 }
diff --git a/pkg/watermark/publish/publisher_inmem_test.go b/pkg/watermark/publish/publisher_inmem_test.go
index 1a580d1f..a44068b8 100644
--- a/pkg/watermark/publish/publisher_inmem_test.go
+++ b/pkg/watermark/publish/publisher_inmem_test.go
@@ -23,9 +23,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/numaproj/numaflow/pkg/watermark/store/inmem"
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/watermark/processor"
 	"github.com/numaproj/numaflow/pkg/watermark/store"
@@ -51,22 +53,22 @@ func TestPublisherWithSharedOTBuckets_InMem(t *testing.T) {
 	var epoch int64 = 1651161600000
 	var location, _ = time.LoadLocation("UTC")
 	for i := 0; i < 3; i++ {
-		p.PublishWatermark(processor.Watermark(time.UnixMilli(epoch).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
+		p.PublishWatermark(wmb.Watermark(time.UnixMilli(epoch).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
 		epoch += 60000
 		time.Sleep(time.Millisecond)
 	}
 	// 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) }))
+	p.PublishWatermark(wmb.Watermark(time.UnixMilli(epoch-120000).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
 	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())
+	assert.Equal(t, wmb.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), wm.String())
 
 	head := p.GetLatestWatermark()
-	assert.Equal(t, processor.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), head.String())
+	assert.Equal(t, wmb.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), head.String())
 
 	_ = p.Close()
 
diff --git a/pkg/watermark/publish/publisher_test.go b/pkg/watermark/publish/publisher_test.go
index cf534530..e945c7ce 100644
--- a/pkg/watermark/publish/publisher_test.go
+++ b/pkg/watermark/publish/publisher_test.go
@@ -23,9 +23,10 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
-	"github.com/numaproj/numaflow/pkg/watermark/ot"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/numaproj/numaflow/pkg/watermark/wmb"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	jsclient "github.com/numaproj/numaflow/pkg/shared/clients/nats"
 	natstest "github.com/numaproj/numaflow/pkg/shared/clients/nats/test"
@@ -79,31 +80,31 @@ func TestPublisherWithSharedOTBucket(t *testing.T) {
 	var epoch int64 = 1651161600000
 	var location, _ = time.LoadLocation("UTC")
 	for i := 0; i < 3; i++ {
-		p.PublishWatermark(processor.Watermark(time.UnixMilli(epoch).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
+		p.PublishWatermark(wmb.Watermark(time.UnixMilli(epoch).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(i) }))
 		epoch += 60000
 		time.Sleep(time.Millisecond)
 	}
 	// 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) }))
+	p.PublishWatermark(wmb.Watermark(time.UnixMilli(epoch-120000).In(location)), isb.SimpleStringOffset(func() string { return strconv.Itoa(0) }))
 
 	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())
+	assert.Equal(t, wmb.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), wm.String())
 
 	head := p.GetLatestWatermark()
-	assert.Equal(t, processor.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), head.String())
+	assert.Equal(t, wmb.Watermark(time.UnixMilli(epoch-60000).In(location)).String(), head.String())
 
 	// simulate the next read batch where the watermark is now $epoch
-	p.PublishIdleWatermark(processor.Watermark(time.UnixMilli(epoch).In(location)))
+	p.PublishIdleWatermark(wmb.Watermark(time.UnixMilli(epoch).In(location)))
 	keys, err = p.otStore.GetAllKeys(p.ctx)
 	assert.NoError(t, err)
 	assert.Equal(t, []string{"publisherTestPod1"}, keys)
 	otValue, err := p.otStore.GetValue(p.ctx, keys[0])
 	assert.NoError(t, err)
-	otDecode, err := ot.DecodeToOTValue(otValue)
+	otDecode, err := wmb.DecodeToWMB(otValue)
 	assert.NoError(t, err)
 	assert.True(t, otDecode.Idle)
 
diff --git a/pkg/watermark/ot/ot_test.go b/pkg/watermark/wmb/wmb_test.go
similarity index 86%
rename from pkg/watermark/ot/ot_test.go
rename to pkg/watermark/wmb/wmb_test.go
index 9b15b901..bef3d56c 100644
--- a/pkg/watermark/ot/ot_test.go
+++ b/pkg/watermark/wmb/wmb_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package ot
+package wmb
 
 import (
 	"bytes"
@@ -30,14 +30,14 @@ func TestDecodeToOTValue(t *testing.T) {
 	tests := []struct {
 		name    string
 		args    args
-		want    Value
+		want    WMB
 		wantErr bool
 	}{
 		{
 			name: "decode_success_using_ot_value",
 			args: args{
 				b: func() []byte {
-					v := Value{
+					v := WMB{
 						Offset:    100,
 						Watermark: 1667495100000,
 						Idle:      false,
@@ -47,7 +47,7 @@ func TestDecodeToOTValue(t *testing.T) {
 					return buf.Bytes()
 				}(),
 			},
-			want: Value{
+			want: WMB{
 				Offset:    100,
 				Watermark: 1667495100000,
 				Idle:      false,
@@ -68,7 +68,7 @@ func TestDecodeToOTValue(t *testing.T) {
 					return buf.Bytes()
 				}(),
 			},
-			want:    Value{},
+			want:    WMB{},
 			wantErr: true,
 		},
 		{
@@ -87,7 +87,7 @@ func TestDecodeToOTValue(t *testing.T) {
 					return buf.Bytes()
 				}(),
 			},
-			want:    Value{},
+			want:    WMB{},
 			wantErr: true,
 		},
 		{
@@ -95,20 +95,20 @@ func TestDecodeToOTValue(t *testing.T) {
 			args: args{
 				b: func() []byte {
 					v := struct {
-						Test0 int64
+						Test0 bool
 						Test1 int64
-						Test2 bool
+						Test2 int64
 					}{
-						Test0: 0,
+						Test0: true,
 						Test1: 0,
-						Test2: true,
+						Test2: 0,
 					}
 					buf := new(bytes.Buffer)
 					_ = binary.Write(buf, binary.LittleEndian, v)
 					return buf.Bytes()
 				}(),
 			},
-			want: Value{
+			want: WMB{
 				Offset:    0,
 				Watermark: 0,
 				Idle:      true,
@@ -120,14 +120,14 @@ func TestDecodeToOTValue(t *testing.T) {
 			args: args{
 				b: func() []byte {
 					v := struct {
-						Test0 int64
+						Test0 bool
 						Test1 int64
-						Test2 bool
+						Test2 int64
 						Test3 int64 // should be ignored
 					}{
-						Test0: 100,
-						Test1: 1667495100000,
-						Test2: false,
+						Test0: false,
+						Test1: 100,
+						Test2: 1667495100000,
 						Test3: 20,
 					}
 					buf := new(bytes.Buffer)
@@ -135,7 +135,7 @@ func TestDecodeToOTValue(t *testing.T) {
 					return buf.Bytes()
 				}(),
 			},
-			want: Value{
+			want: WMB{
 				Offset:    100,
 				Watermark: 1667495100000,
 				Idle:      false,
@@ -145,13 +145,13 @@ func TestDecodeToOTValue(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := DecodeToOTValue(tt.args.b)
+			got, err := DecodeToWMB(tt.args.b)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("DecodeToOTValue() error = %v, wantErr %v", err, tt.wantErr)
+				t.Errorf("DecodeToWMB() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
 			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("DecodeToOTValue() got = %v, want %v", got, tt.want)
+				t.Errorf("DecodeToWMB() got = %v, want %v", got, tt.want)
 			}
 		})
 	}
@@ -173,17 +173,17 @@ func TestOTValue_EncodeToBytes(t *testing.T) {
 		{
 			name: "encode_success",
 			fields: fields{
+				Idle:      false,
 				Offset:    100,
 				Watermark: 1667495100000,
-				Idle:      false,
 			},
-			want:    []byte{100, 0, 0, 0, 0, 0, 0, 0, 96, 254, 115, 62, 132, 1, 0, 0, 0},
+			want:    []byte{0, 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{
+			v := WMB{
 				Offset:    tt.fields.Offset,
 				Watermark: tt.fields.Watermark,
 			}