fix(watermark): generator should not publish wm for every message (#217)

Signed-off-by: Derek Wang <[email protected]>

mkdocs.yml

@@ -80,3 +80,4 @@ nav:
       - debugging.md
       - static-code-analysis.md
       - releasing.md
+  - Numaproj: https://www.numaproj.io

pkg/sources/generator/tickgen.go

@@ -193,28 +193,29 @@ func (mg *memgen) Read(ctx context.Context, count int64) ([]*isb.ReadMessage, er
 	msgs := make([]*isb.ReadMessage, 0, count)
 	// timeout should not be re-triggered for every run of the for loop. it is for the entire Read() call.
 	timeout := time.After(mg.readTimeout)
+loop:
 	for i := int64(0); i < count; i++ {
 		// since the Read call is blocking, and runs in an infinite loop
 		// we implement Read With Wait semantics
 		select {
 		case r := <-mg.srcchan:
 			tickgenSourceReadCount.With(map[string]string{metricspkg.LabelVertex: mg.name, metricspkg.LabelPipeline: mg.pipelineName}).Inc()
 			msgs = append(msgs, newreadmessage(r.data, r.offset))
-			// publish the last message's offset with watermark, this is an optimization to avoid too many insert calls
-			// into the offset timeline store.
-			// Please note that we are inserting the watermark before the data has been persisted into ISB by the forwarder.
-			o := msgs[len(msgs)-1].ReadOffset
-			nanos, _ := o.Sequence()
-			// remove the nanosecond precision
-			mg.sourcePublishWM.PublishWatermark(processor.Watermark(time.Unix(0, nanos)), o)
-
 		case <-timeout:
 			mg.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", mg.readTimeout))
-			return msgs, nil
+			break loop
 		}
 	}
+	if len(msgs) > 0 {
+		// publish the last message's offset with watermark, this is an optimization to avoid too many insert calls
+		// into the offset timeline store.
+		// Please note that we are inserting the watermark before the data has been persisted into ISB by the forwarder.
+		o := msgs[len(msgs)-1].ReadOffset
+		nanos, _ := o.Sequence()
+		// remove the nanosecond precision
+		mg.sourcePublishWM.PublishWatermark(processor.Watermark(time.Unix(0, nanos)), o)
+	}
 	return msgs, nil
-
 }
 
 // Ack acknowledges an array of offset.

pkg/watermark/generic/jetstream/generic.go

@@ -17,6 +17,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/watermark/fetch"
 	"github.com/numaproj/numaflow/pkg/watermark/publish"
 	"github.com/numaproj/numaflow/pkg/watermark/store/jetstream"
+	"github.com/numaproj/numaflow/pkg/watermark/store/noop"
 )
 
 // BuildWatermarkProgressors is used to populate fetchWatermark, and a map of publishWatermark with edge name as the key.
@@ -87,6 +88,9 @@ func BuildSourcePublisherStores(ctx context.Context, vertexInstance *v1alpha1.Ve
 	if !vertexInstance.Vertex.IsASource() {
 		return nil, fmt.Errorf("not a source vertex")
 	}
+	if !sharedutil.IsWatermarkEnabled() {
+		return store.BuildWatermarkStore(noop.NewKVNoOpStore(), noop.NewKVNoOpStore()), nil
+	}
 	pipelineName := vertexInstance.Vertex.Spec.PipelineName
 	sourceBufferName := vertexInstance.Vertex.GetFromBuffers()[0].Name
 	// heartbeat