Repository: numaproj/numaflow (go)

Issue:
Watermark shows -62135596800 when a fresh pipeline started without any traffic
**Describe the bug**


<img width="1275" alt="Screen Shot 2022-08-08 at 5 53 25 PM" src="https://user-images.githubusercontent.com/2832028/183539455-b98a9072-953f-4a1d-b39f-c76f68ede247.png">


Pull request:
fix: update default watermark to -1. Fixes #133
Fixes: #133

This change is to fix #133 where watermark shows -62135596800 when a fresh pipeline started without any traffic. The daemon service shows -62135596800 because when the OT store is empty, the watermark fetcher returns time.Time{} as default watermark, which transforms to epoch as -62135596800. In this change I am proposing -1 as default watermark to indicate there is no valid watermark yet.

Before:
![image](https://user-images.githubusercontent.com/13165977/195914250-74eb6a83-0b76-4c70-9c49-fbe4da8651f3.png)

After:
<img width="1259" alt="Screen Shot 2022-10-14 at 2 09 16 PM" src="https://user-images.githubusercontent.com/13165977/195914344-4320f3b8-f151-4ab7-82a0-3debe24eb612.png">

Signed-off-by: Keran Yang <yangkr920208@gmail.com>

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


Patch:
diff --git a/docs/quick-start.md b/docs/quick-start.md
index 9008c3ed..8f9a9898 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -106,7 +106,7 @@ kubectl logs -f even-odd-odd-sink-0-a6p0n main
 2022/09/07 22:30:19 (odd-sink) 103
 ```
 
-View the UI for the advanced pipleline at https://localhost:8443/
+View the UI for the advanced pipeline at https://localhost:8443/
 ![Numaflow UI](assets/numaflow-ui-advanced-pipeline.png)
 
 The source code of the `even-odd` [User Defined Function](./user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd). You also can replace the [Log](./sinks/log.md) Sink with some other sinks like [Kafka](./sinks/kafka.md) to forward the data to Kafka topics.
diff --git a/docs/sinks/user-defined-sinks.md b/docs/sinks/user-defined-sinks.md
index cfdfe259..9317e0b0 100644
--- a/docs/sinks/user-defined-sinks.md
+++ b/docs/sinks/user-defined-sinks.md
@@ -1,6 +1,6 @@
 # User Defined Sinks
 
-A `Pipleline` may have multiple Sinks, those sinks could either be a pre-defined sink such as `kafka`, `log`, etc, or a `User Defined Sink`.
+A `Pipeline` may have multiple Sinks, those sinks could either be a pre-defined sink such as `kafka`, `log`, etc, or a `User Defined Sink`.
 
 A pre-defined sink vertex runs single-container pods, a user defined sink runs two-container pods.
 
diff --git a/docs/user-defined-functions.md b/docs/user-defined-functions.md
index a93a7b74..3b07c748 100644
--- a/docs/user-defined-functions.md
+++ b/docs/user-defined-functions.md
@@ -1,6 +1,6 @@
 # User Defined Functions
 
-A `Pipleline` consists of multiple vertices, `Source`, `Sink` and `UDF(User Defined Functions)`.
+A `Pipeline` consists of multiple vertices, `Source`, `Sink` and `UDF(User Defined Functions)`.
 
 UDF runs as a sidecar container in a Vertex Pod, processes the received data. The communication between the main container (platform code) and the sidecar container (user code) is through gRPC over Unix Domain Socket.
 
diff --git a/pkg/daemon/server/service/pipeline_watermark_query.go b/pkg/daemon/server/service/pipeline_watermark_query.go
index e46d999b..81312760 100644
--- a/pkg/daemon/server/service/pipeline_watermark_query.go
+++ b/pkg/daemon/server/service/pipeline_watermark_query.go
@@ -118,7 +118,9 @@ func (ps *pipelineMetadataQuery) GetVertexWatermark(ctx context.Context, request
 		log.Errorf("watermark fetcher not available for vertex %s in the fetcher map", vertexName)
 		return nil, fmt.Errorf("watermark not available for given vertex, %s", vertexName)
 	}
-	vertexWatermark := time.Time(vertexFetcher.GetHeadWatermark()).Unix()
+
+	vertexWatermark := vertexFetcher.GetHeadWatermark().Unix()
+
 	v := &daemon.VertexWatermark{
 		Pipeline:           &ps.pipeline.Name,
 		Vertex:             request.Vertex,
diff --git a/pkg/watermark/fetch/edge_fetcher.go b/pkg/watermark/fetch/edge_fetcher.go
index 425250b9..98ddb80c 100644
--- a/pkg/watermark/fetch/edge_fetcher.go
+++ b/pkg/watermark/fetch/edge_fetcher.go
@@ -33,7 +33,7 @@ func NewEdgeFetcher(ctx context.Context, edgeName string, processorManager *Proc
 	}
 }
 
-// GetHeadWatermark returns the watermark using the HeadOffset (latest offset). This
+// GetHeadWatermark returns the watermark using the HeadOffset (the latest offset among all processors). This
 // can be used in showing the watermark progression for a vertex when not consuming the messages
 // directly (eg. UX, tests,)
 func (e *edgeFetcher) GetHeadWatermark() processor.Watermark {
@@ -56,9 +56,9 @@ func (e *edgeFetcher) GetHeadWatermark() processor.Watermark {
 	}
 	e.log.Debugf("GetHeadWatermark: %s", debugString.String())
 	if epoch == math.MaxInt64 {
-		return processor.Watermark(time.Time{})
+		// Use -1 as default watermark value to indicate there is no valid watermark yet.
+		return processor.Watermark(time.Unix(-1, 0))
 	}
-
 	return processor.Watermark(time.Unix(epoch, 0))
 }
 
diff --git a/pkg/watermark/fetch/processor_manager.go b/pkg/watermark/fetch/processor_manager.go
index f0f2e75d..bfcc95de 100644
--- a/pkg/watermark/fetch/processor_manager.go
+++ b/pkg/watermark/fetch/processor_manager.go
@@ -16,8 +16,7 @@ import (
 )
 
 // ProcessorManager manages the point of view of Vn-1 from Vn vertex processors (or source processor). The code is running on Vn vertex.
-// It has the mapping of all the processors which in turn has all the information about each processor
-// timelines.
+// It has the mapping of all the processors which in turn has all the information about each processor timelines.
 type ProcessorManager struct {
 	ctx        context.Context
 	hbWatcher  store.WatermarkKVWatcher
@@ -148,7 +147,6 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 			}
 			switch value.Operation() {
 			case store.KVPut:
-				var entity = processor.NewProcessorEntity(value.Key(), processor.WithSeparateOTBuckets(v.opts.separateOTBucket))
 				// do we have such a processor
 				p := v.GetProcessor(value.Key())
 				if p == nil {
@@ -156,6 +154,7 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 					// A fromProcessor needs to be added to v.processors
 					// The fromProcessor may have been deleted
 					// TODO: make capacity configurable
+					var entity = processor.NewProcessorEntity(value.Key(), processor.WithSeparateOTBuckets(v.opts.separateOTBucket))
 					var fromProcessor = NewProcessorToFetch(v.ctx, entity, 10, v.otWatcher)
 					v.addProcessor(value.Key(), fromProcessor)
 					v.log.Infow("v.AddProcessor successfully added a new fromProcessor", zap.String("fromProcessor", value.Key()))
diff --git a/pkg/watermark/fetch/processor_to_fetch.go b/pkg/watermark/fetch/processor_to_fetch.go
index 804b2bd1..c7ebbd29 100644
--- a/pkg/watermark/fetch/processor_to_fetch.go
+++ b/pkg/watermark/fetch/processor_to_fetch.go
@@ -33,7 +33,7 @@ func (s status) String() string {
 	return "unknown"
 }
 
-// ProcessorToFetch is the smallest unit of entity (from we which we fetch data) that does inorder processing or contains inorder data.
+// 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
 	entity         processor.ProcessorEntitier
@@ -58,6 +58,7 @@ func NewProcessorToFetch(ctx context.Context, processor processor.ProcessorEntit
 		otWatcher:      watcher,
 		log:            logging.FromContext(ctx),
 	}
+
 	if watcher != nil {
 		go p.startTimeLineWatcher()
 	}
diff --git a/pkg/watermark/fetch/source_fetcher.go b/pkg/watermark/fetch/source_fetcher.go
index 223370ed..c8a496f4 100644
--- a/pkg/watermark/fetch/source_fetcher.go
+++ b/pkg/watermark/fetch/source_fetcher.go
@@ -42,7 +42,8 @@ func (e *sourceFetcher) GetHeadWatermark() processor.Watermark {
 		}
 	}
 	if epoch == math.MinInt64 {
-		return processor.Watermark(time.Time{})
+		// Use -1 as default watermark value to indicate there is no valid watermark yet.
+		return processor.Watermark(time.Unix(-1, 0))
 	}
 	return processor.Watermark(time.Unix(epoch, 0))
 }
diff --git a/pkg/watermark/generic/jetstream/generic.go b/pkg/watermark/generic/jetstream/generic.go
index 2969732a..87c4daf8 100644
--- a/pkg/watermark/generic/jetstream/generic.go
+++ b/pkg/watermark/generic/jetstream/generic.go
@@ -36,16 +36,16 @@ func BuildWatermarkProgressors(ctx context.Context, vertexInstance *v1alpha1.Ver
 	// Fetcher creation
 	pipelineName := vertexInstance.Vertex.Spec.PipelineName
 	fromBuffer := vertexInstance.Vertex.GetFromBuffers()[0]
-	hbBucket := isbsvc.JetStreamProcessorBucket(pipelineName, fromBuffer.Name)
-	hbWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, hbBucket, jsclient.NewInClusterJetStreamClient())
+	hbBucketName := isbsvc.JetStreamProcessorBucket(pipelineName, fromBuffer.Name)
+	hbWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, hbBucketName, jsclient.NewInClusterJetStreamClient())
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed at new HB KVJetStreamKVWatch, HeartbeatBucket: %s, %w", hbBucket, err)
+		return nil, nil, fmt.Errorf("failed at new HB KVJetStreamKVWatch, HeartbeatBucket: %s, %w", hbBucketName, err)
 	}
 
-	otBucket := isbsvc.JetStreamOTBucket(pipelineName, fromBuffer.Name)
-	otWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, otBucket, jsclient.NewInClusterJetStreamClient())
+	otBucketName := isbsvc.JetStreamOTBucket(pipelineName, fromBuffer.Name)
+	otWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, otBucketName, jsclient.NewInClusterJetStreamClient())
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed at new OT KVJetStreamKVWatch, OTBucket: %s, %w", otBucket, err)
+		return nil, nil, fmt.Errorf("failed at new OT KVJetStreamKVWatch, OTBucket: %s, %w", otBucketName, err)
 	}
 
 	var fetchWatermark fetch.Fetcher
@@ -57,18 +57,18 @@ func BuildWatermarkProgressors(ctx context.Context, vertexInstance *v1alpha1.Ver
 
 	// Publisher map creation, we need a publisher per edge.
 	for _, buffer := range vertexInstance.Vertex.GetToBuffers() {
-		hbPublisherBucket := isbsvc.JetStreamProcessorBucket(pipelineName, buffer.Name)
+		hbPublisherBucketName := isbsvc.JetStreamProcessorBucket(pipelineName, buffer.Name)
 		// We create a separate Heartbeat bucket for each edge though it can be reused. We can reuse because heartbeat is at
 		// vertex level. We are creating a new one for the time being because controller creates a pair of buckets per edge.
-		hbStore, err := jetstream.NewKVJetStreamKVStore(ctx, pipelineName, hbPublisherBucket, jsclient.NewInClusterJetStreamClient())
+		hbStore, err := jetstream.NewKVJetStreamKVStore(ctx, pipelineName, hbPublisherBucketName, jsclient.NewInClusterJetStreamClient())
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed at new HB Publish JetStreamKVStore, HeartbeatPublisherBucket: %s, %w", hbPublisherBucket, err)
+			return nil, nil, fmt.Errorf("failed at new HB Publish JetStreamKVStore, HeartbeatPublisherBucket: %s, %w", hbPublisherBucketName, err)
 		}
 
-		otStoreBucket := isbsvc.JetStreamOTBucket(pipelineName, buffer.Name)
-		otStore, err := jetstream.NewKVJetStreamKVStore(ctx, pipelineName, otStoreBucket, jsclient.NewInClusterJetStreamClient())
+		otStoreBucketName := isbsvc.JetStreamOTBucket(pipelineName, buffer.Name)
+		otStore, err := jetstream.NewKVJetStreamKVStore(ctx, pipelineName, otStoreBucketName, jsclient.NewInClusterJetStreamClient())
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed at new OT Publish JetStreamKVStore, OTBucket: %s, %w", otStoreBucket, err)
+			return nil, nil, fmt.Errorf("failed at new OT Publish JetStreamKVStore, OTBucket: %s, %w", otStoreBucketName, err)
 		}
 
 		var processorName = fmt.Sprintf("%s-%d", vertexInstance.Vertex.Name, vertexInstance.Replica)
@@ -94,10 +94,10 @@ func BuildSourcePublisherStores(ctx context.Context, vertexInstance *v1alpha1.Ve
 	pipelineName := vertexInstance.Vertex.Spec.PipelineName
 	sourceBufferName := vertexInstance.Vertex.GetFromBuffers()[0].Name
 	// heartbeat
-	hbBucket := isbsvc.JetStreamProcessorBucket(pipelineName, sourceBufferName)
-	hbKVStore, err := jetstream.NewKVJetStreamKVStore(ctx, pipelineName, hbBucket, jsclient.NewInClusterJetStreamClient())
+	hbBucketName := isbsvc.JetStreamProcessorBucket(pipelineName, sourceBufferName)
+	hbKVStore, err := jetstream.NewKVJetStreamKVStore(ctx, pipelineName, hbBucketName, jsclient.NewInClusterJetStreamClient())
 	if err != nil {
-		return nil, fmt.Errorf("failed at new HB KVJetStreamKVStore for source, HeartbeatBucket: %s, %w", hbBucket, err)
+		return nil, fmt.Errorf("failed at new HB KVJetStreamKVStore for source, HeartbeatBucket: %s, %w", hbBucketName, err)
 	}
 
 	// OT
diff --git a/pkg/watermark/generic/noop.go b/pkg/watermark/generic/noop.go
index 272077b2..81b68b35 100644
--- a/pkg/watermark/generic/noop.go
+++ b/pkg/watermark/generic/noop.go
@@ -9,7 +9,7 @@ import (
 )
 
 // NoOpWMProgressor is a no-op watermark progressor. As the name suggests, it does not do anything, no watermark is
-// progressed.
+// progressed. NoOpWMProgressor is used when watermarking is disabled.
 type NoOpWMProgressor struct {
 }
 
diff --git a/pkg/watermark/generic/progressor.go b/pkg/watermark/generic/progressor.go
deleted file mode 100644
index 64888032..00000000
--- a/pkg/watermark/generic/progressor.go
+++ /dev/null
@@ -1,13 +0,0 @@
-package generic
-
-import (
-	"github.com/numaproj/numaflow/pkg/watermark/fetch"
-	"github.com/numaproj/numaflow/pkg/watermark/publish"
-)
-
-// Progressor interface defines how the watermark can be progressed.
-// FIXME: delete this
-type Progressor interface {
-	fetch.Fetcher
-	publish.Publisher
-}
diff --git a/pkg/watermark/processor/entity.go b/pkg/watermark/processor/entity.go
index 7cf638ad..f8729e0a 100644
--- a/pkg/watermark/processor/entity.go
+++ b/pkg/watermark/processor/entity.go
@@ -21,6 +21,10 @@ func (w Watermark) String() string {
 	return t.Format(time.RFC3339)
 }
 
+func (w Watermark) Unix() int64 {
+	return time.Time(w).Unix()
+}
+
 func (w Watermark) After(t time.Time) bool {
 	return time.Time(w).After(t)
 }
@@ -92,9 +96,9 @@ func (p *ProcessorEntity) IsOTBucketShared() bool {
 // BuildOTWatcherKey builds the offset-timeline key name
 func (p *ProcessorEntity) BuildOTWatcherKey(watermark Watermark) string {
 	if p.opts.separateOTBucket {
-		return fmt.Sprintf("%d", time.Time(watermark).Unix())
+		return fmt.Sprintf("%d", watermark.Unix())
 	} else {
-		return fmt.Sprintf("%s%s%d", p.GetID(), p.opts.keySeparator, time.Time(watermark).Unix())
+		return fmt.Sprintf("%s%s%d", p.GetID(), p.opts.keySeparator, watermark.Unix())
 	}
 }
 
diff --git a/pkg/watermark/publish/publisher.go b/pkg/watermark/publish/publisher.go
index 8271b3df..b2ecf02a 100644
--- a/pkg/watermark/publish/publisher.go
+++ b/pkg/watermark/publish/publisher.go
@@ -82,7 +82,7 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 		wm = processor.Watermark(time.Time(wm).Add(-p.opts.delay))
 	}
 	// update p.headWatermark only if wm > p.headWatermark
-	if time.Time(wm).After(time.Time(p.headWatermark)) {
+	if wm.After(time.Time(p.headWatermark)) {
 		p.headWatermark = wm
 	} else {
 		p.log.Infow("New watermark is ignored because it's older than the current watermark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
@@ -95,7 +95,7 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 	// build value (offset)
 	value := make([]byte, 8)
 	var seq int64
-	if p.opts.isSource || p.opts.isSink { // For source and sink publisher, we dont' care about the offset, also the sequence of the offset might not be integer.
+	if p.opts.isSource || p.opts.isSink { // For source and sink publisher, we don't care about the offset, also the sequence of the offset might not be integer.
 		seq = time.Now().UnixNano()
 	} else {
 		seq, _ = offset.Sequence()
diff --git a/pkg/watermark/store/jetstream/kv_watch.go b/pkg/watermark/store/jetstream/kv_watch.go
index a6d96f13..2e2d5b81 100644
--- a/pkg/watermark/store/jetstream/kv_watch.go
+++ b/pkg/watermark/store/jetstream/kv_watch.go
@@ -24,7 +24,7 @@ type jetStreamWatch struct {
 
 var _ store.WatermarkKVWatcher = (*jetStreamWatch)(nil)
 
-// NewKVJetStreamKVWatch returns KVJetStreamWatch specific to Jetsteam which implements the WatermarkKVWatcher interface.
+// NewKVJetStreamKVWatch returns KVJetStreamWatch specific to Jetstream which implements the WatermarkKVWatcher interface.
 func NewKVJetStreamKVWatch(ctx context.Context, pipelineName string, kvBucketName string, client jsclient.JetStreamClient, opts ...JSKVWatcherOption) (store.WatermarkKVWatcher, error) {
 	var err error
 	conn, err := client.Connect(ctx)


Test patch:
diff --git a/pkg/watermark/processor/entity_test.go b/pkg/watermark/processor/entity_test.go
index b5bd0eaa..b739de3c 100644
--- a/pkg/watermark/processor/entity_test.go
+++ b/pkg/watermark/processor/entity_test.go
@@ -28,6 +28,11 @@ func ExampleWatermark_String() {
 	// 2022-04-28T07:00:00Z
 }
 
+func TestExampleWatermarkUnix(t *testing.T) {
+	wm := Watermark(time.Unix(1651129200, 0))
+	assert.Equal(t, int64(1651129200), wm.Unix())
+}
+
 func TestProcessorEntity_ParseOTWatcherKey(t *testing.T) {
 	tests := []struct {
 		name      string