[HUDI-5095] Flink: Stores a special watermark(flag) to identify the c…

[JerryYue-M]

Change Logs

Fixed #7098 Hold a event time instance in each write task to evaluate the progress for write data

Impact

No Impact ，User can disable it use config

Risk level (write none, low medium or high below)

low

Documentation Update

No

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[danny0405]

Can we move these logic into AbstractStreamWriteFunction.java.

[JerryYue-M]

The reason why put it here is BulkInsertWriteFunction only extends BulkInsertWriteFunction instead of AbstractStreamWriteFunction, if moving this code to AbstractStreamWriteFunction will make a specified way to deal with BulkInsertWriteFunction

[JerryYue-M]

@XuQianJin-Stars what do you think about this way to describe the progress of the streaming writer

[XuQianJin-Stars]

@XuQianJin-Stars what do you think about this way to describe the progress of the streaming writer

+1， This is also the implementation method used in our internal production.

[yuzhaojing]

Please fix checkstyle to pass ci.

[JerryYue-M]

@yuzhaojing thanks for the review， I will fix the issue you mentioned

[codope]

@JerryYue-M Gentle ping to fix the checkstyle issues. This is very close to merging.

[JerryYue-M]

@codope @danny0405 @yuzhaojing @XuQianJin-Stars
all the discussion and check style issues had been fixed and CI passed already

[JerryYue-M]

@hudi-bot run azure

[JerryYue-M]

@hudi-bot run azure

[JerryYue-M]

@hudi-bot run azure

[XuQianJin-Stars]

@hudi-bot run azure

[danny0405]

Can we fetch the event time through the payload clazz event time field we the HoodieRecordPayload#getInsertValue is invoked ? And we can finalize the max timestamp
in the coordinator when commit the metadata.

[JerryYue-M]

Can we fetch the event time through the payload clazz event time field we the HoodieRecordPayload#getInsertValue is invoked ? And we can finalize the max timestamp in the coordinator when commit the metadata.
@danny0405 thanks for your suggestion, it's import for me
the reason why i think we need implement in this way is :

1. I think this is a lightweight and common behavior, no matter what type of payload it is used. if we implement this base on payload, it may be limited by payload class between flink and spark/presto eg.
2. for later we may need to add some metric for the flink write function eg: processing latency metric , and also need to extract the timestamp from the record
3. implement base payload may change some code in common module such as modify the HoodieAppendHandler or HoodieMergeHandler

[hudi-bot]

CI report:

• f29644c Azure: FAILURE Azure: FAILURE Azure: FAILURE Azure: CANCELED Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build