Fix: Need to investigate about Java Instant.now() time conversion to protobuf then to numaflow

[shashank10456]

The new watermark value was set to 5336473980199843232 which is 120+ years ahead from now. Need to investigate how that value is being set.

[jy4096]

Note: Findings...

1. When we construct a toDrop message, java SDK assigns Instant.MIN to event time, which translates to epoch is exactly “5336473980199843232”. While in Go SDK, we use time.Time{}, which translates to “-62135596800000"(0001-01-01 00:00:00 +0000 UTC) (by @KeranYang and @shashank10456 )

2. We drop messages in the whereTo func. In the source vertex with a transformer, we publish the source watermark before we call whereTo that’s why we are publishing dropped message’s watermark in source vertex

if the source vertex doesn’t have a transformer…. then there’s no toDrop tag in the original source message, so it’s not affected by "publish watermark before whereTo"

[KeranYang]

The issue we're encountering arises from how negative numbers are represented in computers (two's complement representation) and how we're interpreting the results.

• Java Side: When we use Instant.MIN, it represents the smallest representable Instant value, which is really far in the past. In two's complement, this is a large negative value.

• Protobuf: The com.google.protobuf.Timestamp represents time as seconds since epoch (1970-01-01 00:00:00 UTC) and nanoseconds of that second. When we use a very large negative number for seconds, that will be transmitted correctly via protobuf.

• Go Side: When we receive the message in Go, the negative value for seconds is interpreted correctly. Comparing it with time.Now(), Go realizes it's a time far in the past.

However, when we use time.UnixMilli(), it returns the number of milliseconds since January 1, 1970 UTC. If we try to print the result directly, it's showing us a very large positive number because the two's complement representation of a large negative number looks like a large positive number if interpreted as an unsigned value.

[KeranYang]

The issue of dropping messages has been fixed by https://github.com/numaproj/numaflow/issues/1357 . I am closing this issue.