fix: getWatermark to return-1 if any processor returns -1

[yhl25]

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

while fetching watermark return -1, if the offset timeline store returns -1 for any processor.

[vigith]

TestFetcherWithSameOTBucket is failing

[vigith]

@whynowy, how do we handle this in Source watermark fetcher?

[jy4096]

return -1 for watermark if we don't find an entry in the offset timeline store in any processor

The PR description doesn't match the logic here.

For example, if we have processors A B C, if processor A's watermark for inputOffset is -1, then we break the for loop and skip processors B and C.
The description above means if all A B C returns -1 then we return -1.

[jy4096]

We already have the logic here
https://github.com/numaproj/numaflow/pull/402/files#diff-6af877f3295078204591f48a2180bc855394b50997f764275b25ef2ffff678daL110-R116

numaflow/pkg/watermark/fetch/edge_fetcher.go

Lines 114 to 116 in 6863142

	if epoch == math.MaxInt64 {
	epoch = -1
	}

to return -1 if no entry found in any offset timeline.

[yhl25]

updated the description

[vigith]

But we need to set the overall watermark to -1 if watermark cannot be fetched for any processing unit

[jy4096]

updated the description

👌

do we still need this logic then?

numaflow/pkg/watermark/fetch/edge_fetcher.go

Lines 114 to 116 in 6863142

	if epoch == math.MaxInt64 {
	epoch = -1
	}

[yhl25]

we don't need it

[yhl25]

I think we need it, if there are no processors, we will return int max