[HUDI-8704] Instant time to completion time state upgrade migration for Flink StreamReadMonitoringFunction

[voonhous]

Change Logs

Instant time to completion time state upgrade migration for Flink StreamReadMonitoringFunction.

When performing a version upgrade, users are required to stop their job with savepoint, then restart their job with a newer Hudi-Flink-Bundle. To ensure forward compatibility, we added a state upgrader interface for StreamReadMonitoring function to allow forward compatibility for readers.

1. Added an interface to handle state upgrades.
2. Implemented StreamReadMonitoringStateUpgrader to handle issuedInstant (requestInstant) -> issuedInstant + issuedOffset (completionTime) upgrade
3. Moved outdated legacy validation logic into StreamReadMonitoringStateUpgrader to avoid confusion.

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. If not, put "none".

• 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

[voonhous]

@danny0405 Apologies that this took so long, it is ready for review now.

[hudi-bot]

CI report:

• c8b55ad Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[danny0405]

I expected this be a simple change, check the issuedOffset against the timeline for the first time of read, if the issuedOffset is an instant time, then just switch to it's completion time or modification time.

[voonhous]

Apologies for the misunderstanding, i do not quite understand what you mean, I was under the impression that this would be a backward compatibility improvement when migrating from a Hudi 0.x table to a Hudi 1.0 table.

From what i understand:

version	attribute	available in state?
0.x	issuedInstant	true
0.x	issuedOffset	false
1.x	issuedInstant	true
1.x	issuedOffset	true

Hence, i was under the impression that we needed a way to add a issuedOffset when upgrading from 0.x, where issuedOffset will be null in the state backend during first read after upgrade.