Flink: Fix TableMaintenance operator uid instability that breaks savepoint restore

[Guosmilesmile]

Motivation

The default value of uidSuffix in TableMaintenance.Builder was previously:

private String uidSuffix = "TableMaintenance-" + UUID.randomUUID();

This makes savepointsunusable for job recovery. When users try to restart a maintenance job from a savepoint, they hit the following error:

Cannot map checkpoint/savepoint state for operator b1b352db8bd2aef51f7534c353789605
to the new program, because the operator is not available in the new program.
If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.

Although we can explicitly configure uidSuffix to work around this problem, it shouldn't be a required field.

Changes

Change the default uidSuffix to be deterministically derived from the table name and the task names

Limitation

The new approach cannot automatically disambiguate uids in the following scenario:
Within a single Flink job, two TableMaintenance instances are created for the same table, and both pipelines are configured with the same set of maintenance task types (e.g. both contain a RewriteDataFiles, only differing in partition predicates or parameters).

In this case both pipelines see the same tableName and the same sorted set of task names, so the fallback uidSuffix values are identical, causing pipeline-level shared operators to collide.

My understanding is that in this scenario, configuring multiple tasks inside a single TableMaintenance is the appropriate usage. Multiple tasks can be add() into one pipeline, which is more resource-efficient and provides cleaner lock semantics. If a user really needs to split the tasks across multiple TableMaintenance instances, they can still disambiguate by explicitly providing a distinct uidSuffix to each instance.

[Guosmilesmile]

Hi @pvary @mxm , please take a look if you have time, thanks ~

[mxm]

Thanks for the PR @Guosmilesmile!

[mxm]

The drawback of this approach is that, you won't be able to add new maintenance tasks or drop existing ones (using the allowNonRestoredState option), without discarding existing checkpoints/savepoints. I'm wondering whether we should use a static default uid instead, or the table name like in IcebergSink.

[Guosmilesmile]

Do you mean remove maintenanceTaskName only use table name ?

[Guosmilesmile]

Initially I considered using only the table name. But considering that there might be two independent tasks running within a single Flink job, which would cause a conflict, I added the maintenanceTaskName.

[mxm]

Do you mean remove maintenanceTaskName only use table name ?

Yes.

Initially I considered using only the table name. But considering that there might be two independent tasks running within a single Flink job, which would cause a conflict, I added the maintenanceTaskName.

The same would be true for multiple instances of IcebergSink. Maybe ok to force the user to set an explicit UID for these cases.

[Guosmilesmile]

Makes sense. I'll use the table name as the parameter for this UID.

[mxm]

Thanks @Guosmilesmile!

[pvary]

Merged to main.
Thanks @Guosmilesmile for the fix and @mxm for the review!