[Bug]: Running orphan-clean process is not recovered after database-HA bucket takeover

[xxubai]

What happened?

During Apache Amoro 0.9.0-incubating RC4 validation, a running local
orphan-clean process was not recovered when database-HA bucket ownership moved
from a failed AMS node to another already-running AMS node.

After the takeover:

• the surviving AMS successfully acquired the failed node's buckets;
• the affected Iceberg table was loaded by the surviving AMS;
• the persisted process remained RUNNING;
• external_process_identifier remained empty;
• finish_time remained NULL;
• the orphan file remained present.

The same process state is recovered successfully when the owner AMS is
restarted. This means the RC4 startup recovery path works, but the online
bucket-takeover recovery path does not.

Expected behavior: after the table bucket is assigned to another AMS, the new
owner should recover persisted SUBMITTED/RUNNING processes for that table.
For this orphan-clean process, the process should finish as SUCCESS and
remove the orphan file.

This was found while validating RC4 and is considered release-blocking because
maintenance processes can remain stuck after an AMS failure in database-HA
deployments.

Affects Versions

0.9.0-incubating RC4 (v0.9.0-rc4,
a93a32b4d817e41af730b2e30bfce6e292dc8575)

What table formats are you seeing the problem on?

Iceberg

What engines are you seeing the problem on?

AMS

How to reproduce

1. Start two AMS nodes backed by the same MySQL database and shared warehouse:
   • ha.type=database
   • ha.use-master-slave-mode=true
   • 16 table-service buckets
2. Create a filesystem catalog with an Iceberg table and enable orphan clean:
   • clean-orphan-file.enabled=true
   • clean-orphan-file.min-existing-time-minutes=0
3. Identify the AMS node that owns the table's bucket.
4. Create an old, unreferenced file under the table's data directory.
5. Persist a table_process record for the table with:
   • status='RUNNING'
   • process_type='CLEAN-ORPHAN-FILES'
   • execution_engine='local'
   • an empty external_process_identifier
6. Kill the owner AMS node without a graceful shutdown.
7. Wait until the other, already-running AMS acquires the table's bucket.
8. Query table_process and check the orphan file.

Observed database state after bucket takeover:

process_id:                  1462464657670146
table_id:                    1
external_process_identifier:
status:                      RUNNING
process_type:                CLEAN-ORPHAN-FILES
execution_engine:            local
finish_time:                 NULL
summary:                     {}

Observed file state:

/warehouse/rc4_db/orphan_test/data/rc4-recovery-takeover.parquet
PRESENT

As a control case, restarting the owner AMS with the same persisted process
state recovered the process to SUCCESS and deleted the orphan file.

Relevant log output

# The surviving AMS acquired all 16 buckets after the owner was killed.
assignments_json:
["2","4","6","8","10","12","14","16","1","3","5","7","9","11","13","15"]

# The active process was not recovered after takeover.
process_id          external_process_identifier  status   finish_time  summary
1462464657670146                                 RUNNING  NULL         {}

Anything else

Root cause:

ProcessService.initialize() invokes recoverProcesses(tableRuntimes), so
active processes are recovered when an AMS starts:

https://github.com/apache/amoro/blob/v0.9.0-rc4/amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java#L123-L139

During online database-HA bucket takeover, DefaultTableService dynamically
loads the newly assigned table and emits fireTableAdded():

https://github.com/apache/amoro/blob/v0.9.0-rc4/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java#L439-L468

However, ProcessService.handleTableAdded() only forwards the table-added
event to action coordinators and does not recover persisted active processes
for the newly owned table:

https://github.com/apache/amoro/blob/v0.9.0-rc4/amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java#L527-L536

Process recovery is therefore tied to service initialization rather than the
TableRuntime ownership lifecycle.

This is different from #4286 / #4288: that issue fixed process recovery during
AMS restart, while this issue occurs when ownership moves to an already-running
AMS node.

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct