[core] Unify catalog lock interfaces and snapshot commit scopes

[JingsongLi]

Purpose

Catalog locks currently guard atomic snapshot publication while reading the head, detecting conflicts and preparing manifests happen outside that scope. Acquire a table lock before reading the head and hold it through validation, preparation and publication on every attempt. The four single-attempt entry points (tryCommitOnce, replaceManifestListOnce, rollbackToAsLatestOnce and compactManifestOnce) acquire the scope and delegate to private implementations. Retry waits and post-commit callbacks run after release; APPEND and COMPACT publications use separate scopes.

Bind locking and publication in a SnapshotCommit abstract class. Construction requires a lock, and final commit, acquire, runWithLock, ensureValid and close methods enforce the shared contract. Backend implementations provide protected publication and resource-cleanup hooks. Direct commit calls acquire and validate the bound lock; publication inside a preparation scope reuses it and revalidates the lease. FileStoreCommitImpl owns only this publisher: remove its separate lock field and withLock setter. CatalogEnvironment.snapshotCommit creates the lock and publisher together using the branch, table UUID, exact commit user and runtime table options, and owns one loaded catalog. Remove the overload accepting an externally supplied publication lock and the independent writer-lock helper. Construction failures close acquired resources, and lock-close failures still close the publisher/catalog.

Use Lock as the only catalog locking interface. Lock.acquire returns an AutoCloseable scope, which is released on the acquiring thread. The default runWithLock distinguishes acquisition failures with a checked AcquisitionException, propagates operation failures unchanged, and retains the operation's result or original exception when releasing the scope fails. Acquisition interruption stops the operation; release failures are logged and attached to an existing failure. Lock.empty() directly invokes the callable and bypasses scope acquisition, reentrancy state and lock exception handling. FileStoreCommitImpl keeps its preparation/publication body unchanged and only retries acquisition failures in its retry loop, retaining any earlier publication retry state. Catalog.createLock and CatalogLockFactory.createLock return a table-bound Lock; the factory receives the identifier/branch, table UUID, commit user and runtime options at creation. Remove CatalogLock, its named-callback adapter and the legacy factory fallback when a catalog provides its own policy. Hive/JDBC schema operations and snapshot commits use the same catalog entry point, including custom lock.type selection. The factory and publisher SPIs intentionally change without compatibility adapters.

Move lock configuration and factory selection to the concrete catalogs. Hive and JDBC define and consume lock.enabled, lock.type, lock-check-max-sleep and lock-acquire-timeout in their own options classes, preserving the existing keys, defaults and object-store enablement policy. Remove the generic CatalogOptions lock constants and migrate consumers to HiveCatalogOptions or JdbcCatalogOptions. AbstractCatalog and FileSystemCatalog do not interpret these backend options and default to a no-op lock. Update generated configuration tables and concurrency documentation.

Longer preparation requires renewable leases: Hive heartbeats and checks the server before publication; JDBC renews and releases by an internal owner and rejects expired renewal. Zero changed-row renewal results recheck the unexpired owner rather than treating an unchanged timestamp as lost ownership. Existing JDBC lock tables receive a nullable lock_owner column on catalog initialization, requiring ALTER permission. All JDBC writers should upgrade before relying on owner-checked release after expiry. The client Hive transaction timeout must match the metastore timeout.

This extracts the generic locking prerequisite from #10489 so a subsequent REST implementation can use the same abstraction. It adds no REST endpoints, request models, table options or owner-aware REST rollback API. Snapshot and file conflict validation remain required; this does not add a storage fencing token.

Tests

• JDK 8 and JDK 11: 155 core/Hive regression cases passed with normal Maven checks enabled, including all 101 FileStoreCommitTest cases.
• Flink 1 / JDK 8 and Flink 2 / JDK 11: the migrated filesystem catalog lock-factory integration test passed on both profiles, with test sources compiled from clean module outputs.
• Lock tests cover direct no-op execution without acquisition or lease validation, exact result/exception propagation, acquisition failure preventing the operation, reentrancy and validity checks, and release failure preserving successful results or the original failure.
• Commit regressions cover acquisition retry limits on a table with expired snapshots, interruption, fresh-head reads after acquisition, unknown publication followed by acquisition failure, manifest reuse across lock failures, preparation errors, and callbacks after successful publication despite release failure.
• Hive/JDBC tests cover partial acquisition cleanup, release failures, renewable lease expiry, ownership takeover, lock-table upgrades and isolated renewal workers. Configuration and environment tests cover runtime identity/options, factory policies and publisher resource lifecycles.
• Independent comparison confirmed that the preparation/publication body is unchanged by the lock-error simplification and that the no-op path bypasses the new lock lifecycle.

[leaves12138]

Reviewed the latest head b6f8585. No blocking findings.

Checked the bound SnapshotCommit/Lock contract, all four publication paths, fresh-head reads within each attempt, separate APPEND/COMPACT scopes, post-commit callbacks after release, catalog policy/identity propagation, and resource ownership. Also checked Hive/JDBC renewal, owner-checked JDBC release, lock-table migration, and the new preservation of successful/uncertain publication state across acquisition and release failures, including interruption handling.

Independent local verification on JDK 8 with normal Maven checks enabled: 145 core cases and 14 Hive cases passed, with zero failures, errors, or skips; the reactor package build succeeded. This includes all 99 FileStoreCommitTest cases. I did not run live MySQL/PostgreSQL container integration locally.

Full CI for this latest head is still in progress; please wait for it before merging. The intentionally changed lock/publisher SPIs and JDBC ALTER/rolling-upgrade prerequisites are documented.