[BEAM-2703] KafkaIO: watermark outside the bounds of BoundedWindow

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.1.0
• Component/s: io-java-kafka
• Assignee: Raghu Angadi
• Reporter: Chris Pettitt
• Created: 31/Jul/17 20:21
• Updated: 16/May/20 13:45
• Resolved: 26/Oct/17 19:00

Description

KafkaIO appears to use an incorrect lower bound for it's initial watermark with respect to BoundedWindow.TIMESTAMP_MIN_VALUE.

KafkaIO's initial watermark:
new Instant(Long.MIN_VALUE) -> -9223372036854775808

BoundedWindow.TIMESTAMP_MIN_VALUE:
new Instant(TimeUnit.MICROSECONDS.toMillis(Long.MIN_VALUE)) -> -9223372036854775

The difference is that the last three digits have been truncated due to the micro to millis conversion.

This difference can cause errors in runners that assert that the input watermark can never regress as KafkaIO gives a value below the lower bound when no messages have been received yet. For consistency it would probably be best for it to use BoundedWindow.TIMESTAMP_MIN_VALUE.

Issue Links

• links to: GitHub Pull Request #3758