[sakshichitnis27]

Purpose

Linked issue: close #4907

EnableFlussOnLakeTableProcedureITCase.testReenableExistingTableWithRegisteredSnapshot occasionally fails with TableNotExistException: Table N not found or queued for deletion in coordinator context.

enable_fluss_on_lake_table creates the Fluss table via createTableFromLake and then immediately commits the initial lake snapshot. The table is written to ZooKeeper right away, but the coordinator only adds it to its context once the table watcher's CreateTableEvent is processed. If the commit request reaches the coordinator first, it is rejected.

Brief change log

• EnableFlussOnLakeTableProcedure now retries the initial lake snapshot commit (up to 10 attempts, backoff from 100ms to 2s) when the failure is caused by TableNotExistException. Any other failure is thrown on the first attempt.
• The retry is in a small package-private helper so it can be unit tested.

Tests

• EnableFlussOnLakeTableProcedureTest: added a test that the commit is retried until the coordinator knows the table, and one that other failures are not retried.
• To reproduce the race, I temporarily added a 2s delay before the CreateTableEvent is put on the coordinator event queue in TableChangeWatcher (not included in this PR). With the delay, EnableFlussOnLakeTableProcedureITCase#testReenableExistingTableWithRegisteredSnapshot failed with the same error as in the issue before this change, and passes with it. All 8 tests in EnableFlussOnLakeTableProcedureITCase also pass with and without the delay.
• I did not run the full Flink or lake test suites locally.

API and Format

No API or storage format change.

Documentation

No documentation change.