[BEAM-3699] RecordTimestamp should be the default Watermark in KafkaIO

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Duplicate
• Fix Version/s: Not applicable
• Component/s: io-java-kafka
• Assignee: Mingmin Xu
• Reporter: Mingmin Xu
• Created: 13/Feb/18 23:55
• Updated: 24/Jul/20 20:20
• Resolved: 26/Feb/18 22:53

Description

Currently, the priority to get Watermark Instance in KafkaIO is:

getWatermarkFn().apply(curRecord)
  getTimestampFn().apply(record)
    Instant.now()

I would propose to change it as below to leverage KafkaRecord.timestamp if no WatermarkFn() or TimestampFn() is available:

getWatermarkFn().apply(curRecord)
  getTimestampFn().apply(record)
    KafkaRecord(Beam.KafkaIO).timestamp

It equals to

getWatermarkFn().apply(curRecord)
  getTimestampFn().apply(record)
    KafkaRawRecord(Kafka_client).timestamp
      Instant.now()

rangadi any comments?

Issue Links

• links to: GitHub Pull Request #4687