[HUDI-6366] Prevent flink offline table service rerun completed instant

[hbgstc123]

If flink offline table service fail after commit the compaction/clustering instant, and a restart strategy is enable, then the completed instant will be rerun.

Consequence is, if the completed instant happen to be archived, there will be a file not found error; if not archived, then there will be duplicated base files.

Change Logs

check if the compaction/clustering instant is pending in active timeline before running flink offline table service

Impact

none

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

none

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

[danny0405]

if not archived, then there will be duplicated base files.

How are these duplicates generated?

[hbgstc123]

if not archived, then there will be duplicated base files.

How are these duplicates generated?

For compaction, if the first run compact fileID1_timestamp1.log and fileID1_0-1-0_timestamp1.parquet, genrate fileID1_0-1-0_timestamp2.parquet, the job fail after compaction committed, then job restart and rerun this compaction instant, this second run will again compact fileID1_timestamp1.log and fileID1_0-1-0_timestamp1.parquet, but genrate fileID1_0-1-1_timestamp2.parquet, then fail to complete because its already completed in the first run. These 2 files fileID1_0-1-0_timestamp2.parquet and fileID1_0-1-1_timestamp2.parquet are duplicated.

Similar for clustering but worse, because the second run will generate a parquet file with a new file ID, when you read from the table again the result will be wrong.

[danny0405]

the job fail after compaction committed, then job restart and rerun this compaction instant, this second run will again compact fileID1_timestamp1.log and fileID1_0-1-0_timestamp1.parquet,

We should skip the completed instant, is that the behavior of current code then?

     // fetch the instant based on the configured execution sequence
      HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();

[hbgstc123]

the job fail after compaction committed, then job restart and rerun this compaction instant, this second run will again compact fileID1_timestamp1.log and fileID1_0-1-0_timestamp1.parquet,

We should skip the completed instant, is that the behavior of current code then?

     // fetch the instant based on the configured execution sequence
      HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();

yes, but that is the client code, task fail and restart won't rerun that code.

[danny0405]

+1, nice catch ~

[danny0405]

There is a method named containsInstant, can we use that?

[danny0405]

Maybe we can also move the code into run so there is no need to keep a class member isPending.

[yihua]

Rebased this PR to fix build error.

[hudi-bot]

CI report:

• 3c3905b UNKNOWN
• 87974a6 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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