fix: watermark consumer fix

pkg/daemon/server/daemon_server.go

@@ -92,7 +92,7 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 
 func (ds *daemonServer) newGRPCServer(isbSvcClient isbsvc.ISBService) (*grpc.Server, error) {
 	// "Prometheus histograms are a great way to measure latency distributions of your RPCs.
-	// However, since it is bad practice to have metrics of high cardinality the latency monitoring metrics are disabled by default.
+	// However, since it is a bad practice to have metrics of high cardinality the latency monitoring metrics are disabled by default.
 	// To enable them please call the following in your server initialization code:"
 	grpc_prometheus.EnableHandlingTimeHistogram()
 

pkg/watermark/fetch/edge_fetcher.go

@@ -35,7 +35,11 @@ func NewEdgeFetcher(ctx context.Context, edgeName string, processorManager *Proc
 
 // 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,)
+// directly (eg. UX, tests)
+// NOTE
+//   - 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 {
 	var debugString strings.Builder
 	var headOffset int64 = math.MinInt64

pkg/watermark/fetch/processor_manager.go

@@ -58,7 +58,7 @@ func NewProcessorManager(ctx context.Context, watermarkStoreWatcher store.Waterm
 	return v
 }
 
-// addProcessor adds a new processor.
+// addProcessor adds a new processor. If the given processor already exists, the value will be updated.
 func (v *ProcessorManager) addProcessor(processor string, p *ProcessorToFetch) {
 	v.lock.Lock()
 	defer v.lock.Unlock()
@@ -161,7 +161,7 @@ func (v *ProcessorManager) startHeatBeatWatcher() {
 			case store.KVPut:
 				// do we have such a processor
 				p := v.GetProcessor(value.Key())
-				if p == nil {
+				if p == nil || p.IsDeleted() {
 					// if p is nil, create a new processor
 					// A fromProcessor needs to be added to v.processors
 					// The fromProcessor may have been deleted

pkg/watermark/publish/publisher.go

@@ -83,6 +83,7 @@ func (p *publish) PublishWatermark(wm processor.Watermark, offset isb.Offset) {
 	}
 	// update p.headWatermark only if wm > p.headWatermark
 	if wm.After(time.Time(p.headWatermark)) {
+		p.log.Debugw("New watermark updated for head water mark", zap.String("head", p.headWatermark.String()), zap.String("new", wm.String()))
 		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()))

pkg/watermark/store/jetstream/kv_watch.go

@@ -24,7 +24,7 @@ type jetStreamWatch struct {
 
 var _ store.WatermarkKVWatcher = (*jetStreamWatch)(nil)
 
-// NewKVJetStreamKVWatch returns KVJetStreamWatch specific to Jetstream 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)