[JTaky]

Purpose:

Address the issue: #2825

Paimon's filesystem catalog over an object store (e.g. S3) needs an external lock to coordinate concurrent commits because rename is not atomic on an object store. Paimon already ships JDBC and Hive lock implementations for this; this adds a ZooKeeper one, selected via lock.type=zookeeper, aiming for the same lease-based coordination guarantees as JDBC locking.

It's built on a Curator InterProcessMutex (ephemeral-node acquire/run/release, matching Paimon's own lock contract), with two safeguards beyond a bare mutex: a session-loss epoch bumped by a ConnectionStateListener, and a synchronous checkExists() round trip against the held znode before snapshot publication. Curator's local ownership bookkeeping alone does not reliably detect session expiry. Detected ownership loss fails the commit, with stable log markers (paimon_lock_lost, paimon_lock_acquire_timeout, paimon_lock_acquire_error, paimon_lock_release_failed) for monitoring.

Like JDBC locking, this does not fence an in-flight non-atomic storage write if ownership expires during publication. Stronger guarantees require storage-side conditional publication or fencing.

Tests:

This has been running in our environment for about a month, protecting concurrent commits from two independent jobs: writer and separate compaction.

[JingsongLi]

The ZooKeeper lock has a concrete end-to-end use case for concurrent object-store commits. Reviewed this head in an isolated worktree across locking, client lifecycle and engine dependencies. Standard JDK 8 Maven checks and all 55 existing ZooKeeper/FileSystemCatalog tests passed, including genuine session-loss coverage. Two additional production-class probes expose the issues below: equivalent physical warehouses admit concurrent critical sections, and the Curator dependency version supplied by supported Spark 3.4 fails against a real ZooKeeper server.

[JingsongLi]

[P1] Normalize the physical warehouse before deriving its lock namespace

CatalogFactory normalizes the storage warehouse with Paimon Path, but passes the original options through CatalogLockContext. This hash handles S3 scheme aliases and trailing slashes, not the same Path normalization. Thus s3://test-bucket/a//b and s3://test-bucket/a/b address the identical physical warehouse but use different znodes (dot segments have the same problem). I verified Path equality and, with the real embedded ZooKeeper server and two factory-created locks, held the first table lock while a second thread successfully entered the same db/table through the other warehouse spelling. The new lock therefore does not serialize those two writers/compaction jobs, exposing the object-store snapshot overwrite risk it is intended to prevent. Please normalize with Paimon Path as CatalogFactory does before applying the S3 alias folding and hashing; add an actual mutual-exclusion test for these physical path aliases.

[JTaky]

The warehouse path is normalized with Paimon Path before S3 alias folding and hashing. Added an embedded-ZooKeeper test that verifies equivalent paths contend for the same table lock.

[JingsongLi]

[P2] Make the compatible Curator runtime available to engine users

These new dependencies are provided, so the published engine artifacts neither include nor transitively supply the Curator version used by the new lock. Supported Spark 3.4.3 supplies curator-recipes 2.13.0 (the default Hadoop 2.8.5 stack also has 2.7.1); isOwnedByCurrentThread is package-private in those versions, whereas this implementation invokes it from outside that package. Against the same real ZooKeeper server, the production factory/runWithLock probe succeeds with the PR test classpath (Curator 5.6.0) but fails with the Curator client/framework/recipes 2.13.0 set: IllegalAccessError at runWithLock:154 and again at close:247. Core tests pin the newer libraries and do not exercise that engine runtime. Flink relocated classes also cannot satisfy these unrelocated imports. Please ship an isolated compatible dependency set, or provide explicit compatible-library installation requirements and engine-classpath tests, so enabling lock.type=zookeeper on a supported distribution actually works.

[JTaky]

The lock no longer calls isOwnedByCurrentThread(); it uses getLockPath() and checks the session epoch and lock node. A Spark 3.4 test exercises the production reentrant path with Curator 2.13.0. The docs also state which unshaded dependencies engines must provide; Flink’s relocated Curator alone is insufficient.

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS. The previous physical-warehouse namespace and Spark Curator access issues are repaired, but the filesystem use case in the PR purpose is still not connected to this implementation.

[P1] Wire ZooKeeper selection into the filesystem catalog

The new ZooKeeperCatalogLockFactory is discoverable, but FileSystemCatalog still inherits AbstractCatalog.lockFactory() returning Optional.empty() (AbstractCatalog.java:143–145). FileSystemCatalogFactory/CatalogFactory do not install another locking policy. CatalogEnvironment.snapshotCommit consequently obtains EmptyLock through either catalog reconstruction or the null-factory path. Setting metastore=filesystem, lock.enabled=true and lock.type=zookeeper therefore silently leaves actual table commits unlocked. Only the existing Hive/JDBC catalog overrides currently discover lock.type.

I reproduced this through the public CatalogFactory.createCatalog entry point, with a real embedded ZooKeeper server, three independently loaded catalogs/writers/committers and six successful one-row commits per writer. A local FileIO deliberately models non-atomic object-store snapshot publication (check existence, delayed overwrite): the configured ZooKeeper case retains only 6 of the expected 18 rows and admits three concurrent snapshot publishers, exactly like the disabled-lock control. Explicitly injecting this PR's factory into the catalog policy and preserving it through catalogLoader reconstruction, with the rest of the same table/commit/FileIO path unchanged, retains all 18 unique rows and limits publication concurrency to one. No production object store or credentials were used.

Please make the supported filesystem catalog actually resolve this lock configuration and preserve it when reconstructing its catalog/table environment, then cover concurrent real table commits from catalog-created tables. Direct factory construction in the existing lock/Spark tests bypasses this missing integration. If filesystem support is intentionally deferred, narrow the advertised behavior and reject the ineffective configuration rather than implying these commits are protected.

Verification: normal JDK 8 checks and 57 ZooKeeper/FileSystemCatalog tests passed. The real Spark 3.4.3 runtime test passed with Curator client/framework/recipes 2.13.0 and ZooKeeper 3.6.3 verified in the test classpath. Additional real-server client-lifecycle stress passed 800 concurrent create/double-close operations and 12 last-close/new-client races. Both current-head Core CI jobs fail the introduced documentation check described inline; the separate Python 3.6 REST filter failure is outside this Java diff. No source fixes were pushed.

[JingsongLi]

[P2] Keep the shared lock-acquire-timeout definition consistent

This introduces the existing JDBC key lock-acquire-timeout with the same type/default but a different description. ConfigOptionsDocsCompletenessITCase explicitly rejects distinct definitions of one key, so both current-head Core CI jobs fail with "Ambiguous option lock-acquire-timeout due to distinct descriptions (JdbcCatalogOptions vs ZooKeeperCatalogLockOptions)." This is a deterministic failure in the required integration checks, not a dependency-download problem. Please use the existing generic description (or a shared ConfigOption definition) and update the generated ZooKeeper configuration HTML consistently. The default can remain eight minutes.

[JTaky]

Addressed. lock-acquire-timeout is now a shared CatalogOptions definition reused by JDBC and ZooKeeper. The generated catalog and ZooKeeper configuration tables were regenerated, and ConfigOptionsDocsCompletenessITCase passes.

[JTaky]

Requirement fit: SUPPORTED. Implementation: FINDINGS. The previous physical-warehouse namespace and Spark Curator access issues are repaired, but the filesystem use case in the PR purpose is still not connected to this implementation.

[P1] Wire ZooKeeper selection into the filesystem catalog

The new ZooKeeperCatalogLockFactory is discoverable, but FileSystemCatalog still inherits AbstractCatalog.lockFactory() returning Optional.empty() (AbstractCatalog.java:143–145). FileSystemCatalogFactory/CatalogFactory do not install another locking policy. CatalogEnvironment.snapshotCommit consequently obtains EmptyLock through either catalog reconstruction or the null-factory path. Setting metastore=filesystem, lock.enabled=true and lock.type=zookeeper therefore silently leaves actual table commits unlocked. Only the existing Hive/JDBC catalog overrides currently discover lock.type.
described inline; the separate Python 3.6 REST filter failure is outside this Java diff. No source fixes were pushed.

Addressed. FileSystemCatalog now resolves the configured catalog lock factory using the shared lock.enabled and lock.type options. The original CatalogContext is retained by FileSystemCatalogLoader, so reconstructed catalogs use the same locking policy. I added an end-to-end test that creates a filesystem catalog and table through CatalogFactory and verifies that the real table commit blocks behind the corresponding ZooKeeper lock.

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS. The filesystem factory wiring, shared option definition and Spark Curator compatibility are repaired. I ran 101 Core/JDBC/ZooKeeper tests plus the configuration-doc completeness and actual Spark 3.4 Curator runtime tests with regular Maven checks. An actual CatalogFactory concurrent-commit probe over simulated non-atomic snapshot publication retained 18/18 rows with this lock (peak publication concurrency 1), versus 6/18 without it (peak 3). SPI loading without any Curator/ZooKeeper jars and concurrent client-reference lifecycle probes also passed.

One cancellation-path issue remains below. The current JDK 11 CI failure is an ANTLR plugin dependency-resolution failure, not a test result.

[JingsongLi]

[P2] Preserve interruption while verifying or releasing the lock

When checkExists() is waiting for reconnection, Curator can throw InterruptedException from internalBlockUntilConnectedOrTimedOut outside its usual retry-loop interrupt restoration. This catch clears the interruption into nodeGone, and the later IllegalStateException drops its cause. The snapshot publication failure is then treated as retryable, so an interrupted commit can continue and publish. With a real CatalogFactory table and embedded ZooKeeper, I disconnected the server in a normal pre-publication callback, interrupted the commit thread while verification waited, then restarted the server: HEAD returned success on attempt 2 with interrupt=false and snapshot 1 published. A control changing only interrupt restoration in this class stopped on attempt 1 with interrupt=true. Please handle interrupted verification separately, restore the thread flag and retain the cause so cancellation stops retrying; preserve interruption in the scope-release catch as well, which currently bypasses Lock.release restoration.

[JTaky]

Thanks for the point. Fixed.

Interrupted checkExists() now restores the thread’s interrupt flag and throws with the InterruptedException retained as the cause, instead of treating it as a missing lock node. Scope release now passes exceptions to Lock.release, which preserves interruption and suppresses release errors onto the operation failure. I also restored interruption during explicit close(). Added focused tests for all three paths, including a check that interrupted verification does not reach publication.

[JingsongLi]

Requirement fit: SUPPORTED. No new implementation regression found after checking the interruption fix and running 189 Core/JDBC/REST/ZooKeeper and Spark 3.4 runtime tests with regular Maven checks. The JDK 8 CI failure I inspected is a Maven Central connection reset resolving scalatest-compatible, rather than a lock test failure.

Production-boundary note: the session/node checks validate ownership before publication; they do not fence an already in-flight non-atomic object-store publication. With two actual RenamingSnapshotCommit instances and embedded ZooKeeper, I paused A inside simulated non-atomic snapshot publication after validation, expired its session, let B publish snapshot 1, and resumed A. Both returned true, but snapshot 1 contained A, overwriting B. A control checking ownership after the pause rejected A and retained B, but that only moves the window and is not a complete fix for an in-flight write.

This is the existing finite-lease/publication protocol limitation, not a ZooKeeper-specific regression or a request to add another preflight check. Please keep the safety claim bounded accordingly: loss detected before publication fails visibly; loss during a non-atomic publication still needs storage-side conditional publication/fencing for a stronger guarantee. The PR description currently says the synchronous check runs after the callable, while the current production contract validates before publication.

[JTaky]

Requirement fit: SUPPORTED. No new implementation regression found after checking the interruption fix and running 189 Core/JDBC/REST/ZooKeeper and Spark 3.4 runtime tests with regular Maven checks. The JDK 8 CI failure I inspected is a Maven Central connection reset resolving scalatest-compatible, rather than a lock test failure.

Production-boundary note: the session/node checks validate ownership before publication; they do not fence an already in-flight non-atomic object-store publication. With two actual RenamingSnapshotCommit instances and embedded ZooKeeper, I paused A inside simulated non-atomic snapshot publication after validation, expired its session, let B publish snapshot 1, and resumed A. Both returned true, but snapshot 1 contained A, overwriting B. A control checking ownership after the pause rejected A and retained B, but that only moves the window and is not a complete fix for an in-flight write.

This is the existing finite-lease/publication protocol limitation, not a ZooKeeper-specific regression or a request to add another preflight check. Please keep the safety claim bounded accordingly: loss detected before publication fails visibly; loss during a non-atomic publication still needs storage-side conditional publication/fencing for a stronger guarantee. The PR description currently says the synchronous check runs after the callable, while the current production contract validates before publication.

Thanks for the detailed review and reproduction. Agreed: the ownership check runs before publication and does not fence an in-flight non-atomic write after session expiry. We aim for the same lease-based coordination guarantees as JDBC locking, not stronger storage-side fencing.

I’ve added this limitation to the documentation. I’ll also correct the PR description’s outdated “after the callable” wording and narrow its safety claims.

Rebased onto the latest master and pushed.