[HUDI-4432] Checkpoint management for muti-writer scenario

[boneanxs]

Change Logs

1. Add extraPreCommitFunc for BaseHoodieWriteClient.commitStats to allow it execute customized functions(which is checking the checkpoint info and update it in HoodieStreamingSink)
2. Add a new param hoodie.datasource.write.streaming.checkpoint.identifier to identify each writer's checkpoint info, if not set, will hold an in-memory latestBatchId to avoid the issue [HUDI-4389] Make HoodieStreamingSink idempotent #6098

Impact

Existing jobs which already write _hudi_streaming_sink_checkpoint might lost old checkpoint info, as we use a user-provided identifier to get the checkpoint info, not ${sqlContext.sparkContext.applicationId}-$queryId

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

low

Documentation Update

Describe any necessary documentation update if there is any new feature, config, or user-facing change

• The config description must be updated if new configs are added or the default value of the configs are changed
• Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
  ticket number here and follow the instruction to make
  changes to the website.

Contributor's checklist

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

[nsivabalan]

@codope : can you review this code. This is the multi-writer checkpoints for spark structured streaming

[codope]

Thanks for fixing the multi-writer scenario. As it is a breaking change (will break spark streaming pipelines using prior checkpoint method), it is best to take it in the upcoming major release 0.13.0. Approving with some minor changes.

[hudi-bot]

CI report:

• 1d09693 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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