[Improvement]: Refactor snapshot-expiring via ProcessFactory plugin

[baiyangtx]

Why are the changes needed?

Refactor snapshot expiring action via ProcessFactory plugin

Brief change log

• Refactor Iceberg snapshot expiration from the inline scheduler into a pluggable process model (ProcessFactory + ExecuteEngine).
• Introduce IcebergProcessFactory and SnapshotsExpiringProcess, executed by the default LocalExecutionEngine with async execution, status tracking and cancellation.
• Externalize enablement/interval knobs via plugin YAML and new config options (e.g. expire-snapshots.enabled, expire-snapshots.interval).
• Add unit tests for IcebergProcessFactory and LocalExecutionEngine, and update the “last snapshots expiring” timestamp only after the actual expiration run finishes to preserve correct interval semantics.

How was this patch tested?

• Add some test cases that check the changes thoroughly including negative and positive cases if possible

• Add screenshots for manual tests if appropriate

• Run test locally before making a pull request

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[czy006]

YAML is expire-snapshots.interval

[czy006]

last result is pool.default..thread-count / pool.snapshots-expiring..thread-count, that's not get the pool

[baiyangtx]

I have fix it and add some tests

[czy006]

org.apache.amoro.process.ExecuteEngine

[czy006]

The problem is that the new scheduling path no longer preserves the old “run, then record cleanup time” behavior for snapshot expiration.

In the old implementation, SnapshotsExpiringExecutor.java executed tableMaintainer.expireSnapshots() synchronously. Only after that finished did PeriodicTableScheduler.java (line 125) update lastCleanTime and schedule the next run. So the interval was effectively measured from the end of the previous cleanup.

In the new path, ActionCoordinatorScheduler.java (line 103) only submits/registers a process and returns immediately. After that return, PeriodicTableScheduler still updates lastCleanTime right away, even though the real cleanup work has not finished yet. The actual cleanup now happens later in SnapshotsExpiringProcess.java (line 53).

[j1wonpark]

Building on your observation — the async submission also introduces a state-loss issue in LocalExecutionEngine.getStatus().

getStatus() removes the Future from the map on terminal states (isDone/isCancelled), making it non-idempotent:

Call 1: future.isDone() == true → remove → SUCCESS
Call 2: future == null          → KILLED  (wrong!)

TableProcessExecutor polls getStatus() in a loop (line 107), so if any retry or concurrent access queries the same identifier twice after completion, it gets KILLED instead of the real result.

There's also a TOCTOU race between containsKey and get across cancelingInstances/activeInstances (lines 67-70), since the compound check-then-act isn't atomic even with ConcurrentHashMap.

[baiyangtx]

The information of a completed process is not cleaned up immediately, but waits for a TTL period, which is long enough for the process executor to obtain its final status and persist the state into the database.

[czy006]

It looks like IcebergProcessFactory receives available execute engines too early.

In AmoroServiceContainer, availableExecuteEngines(executeEngineManager.installedPlugins()) is called before executeEngineManager.initialize(), so installedPlugins() is still empty at that point. As a result, IcebergProcessFactory.localEngine is never set.

Later, when snapshot expiration is triggered, triggerExpireSnapshot() returns Optional.empty() because localEngine == null, so no SnapshotsExpiringProcess is ever created or submitted.

In other words, the new expire-snapshots path is effectively disabled due to initialization order. We probably need to initialize execute engines before injecting them into process factories, or re-inject them after engine initialization.

[zhoujinsong]

Overall this is a clean refactor that migrates snapshot expiring from the inline executor into the ProcessFactory framework. A few minor notes:

1. Duplicate interval check (IcebergProcessFactory.triggerExpireSnapshot): The time-since-last-execution guard is redundant since ActionCoordinatorScheduler already controls trigger frequency via ProcessTriggerStrategy.FIXED_RATE. Consider removing the manual check from the factory to avoid two sources of truth for scheduling state.

2. retryNumber on recovery: processMeta.getRetryNumber() may deserialize as 0 if the field is not persisted, causing already-failed processes to reset their retry count on AMS restart. Worth verifying the DB mapping.

3. Empty getSummary(): SnapshotsExpiringProcess.getSummary() returns an empty map, so nothing is recorded in the process store after execution. Even a minimal summary (e.g. elapsed time) would help with debugging.

None of these are blockers. LGTM.