fix: getWatermark to return-1 if any processor returns -1 (#402)

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

pkg/reduce/reduce_test.go

@@ -770,9 +770,7 @@ func fetcherAndPublisher(ctx context.Context, toBuffers map[string]isb.BufferWri
 			case <-ctx.Done():
 				return
 			default:
-				for key := range publishers {
-					_ = hb.PutKV(ctx, key, []byte(fmt.Sprintf("%d", time.Now().Unix())))
-				}
+				_ = hb.PutKV(ctx, fromBuffer.GetName(), []byte(fmt.Sprintf("%d", time.Now().Unix())))
 				time.Sleep(time.Duration(1) * time.Second)
 			}
 		}

pkg/watermark/fetch/edge_fetcher.go

@@ -99,15 +99,17 @@ func (e *edgeFetcher) GetWatermark(inputOffset isb.Offset) processor.Watermark {
 	for _, p := range allProcessors {
 		debugString.WriteString(fmt.Sprintf("[Processor: %v] \n", p))
 		var t = p.offsetTimeline.GetEventTime(inputOffset)
-		if t != -1 && t < epoch {
+		if t == -1 { // watermark cannot be computed, perhaps a new processing unit was added or offset fell off the timeline
+			epoch = t
+		} else if t < epoch {
 			epoch = t
 		}
 		if p.IsDeleted() && (offset > p.offsetTimeline.GetHeadOffset()) {
 			// if the pod is not active and the current offset is ahead of all offsets in Timeline
 			e.processorManager.DeleteProcessor(p.entity.GetName())
 		}
 	}
-	// if the offset is smaller than every offset in the timeline, set the value to be -1
+	// if there are no processors
 	if epoch == math.MaxInt64 {
 		epoch = -1
 	}