You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/12 05:54:08 UTC
(paimon) branch master updated: [core] Unify catalog lock interfaces and snapshot commit scopes (#10498)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c2d4563581 [core] Unify catalog lock interfaces and snapshot commit scopes (#10498)
c2d4563581 is described below
commit c2d4563581b9b69ea84226f6703f5240323dc292
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Oct 12 13:54:01 2026 +0800
[core] Unify catalog lock interfaces and snapshot commit scopes (#10498)
---
docs/docs/concepts/concurrency-control.md | 29 +
docs/generated/catalog_configuration.html | 24 -
docs/generated/hive_catalog_configuration.html | 24 +
docs/generated/jdbc_catalog_configuration.html | 24 +
.../org/apache/paimon/options/CatalogOptions.java | 24 -
.../java/org/apache/paimon/AbstractFileStore.java | 8 +-
.../org/apache/paimon/catalog/AbstractCatalog.java | 39 +-
.../java/org/apache/paimon/catalog/Catalog.java | 24 +
.../org/apache/paimon/catalog/CatalogLock.java | 36 -
.../apache/paimon/catalog/CatalogLockFactory.java | 23 +-
.../paimon/catalog/CatalogSnapshotCommit.java | 11 +-
.../org/apache/paimon/catalog/DelegateCatalog.java | 11 +
.../apache/paimon/catalog/FileSystemCatalog.java | 9 +-
.../paimon/catalog/RenamingSnapshotCommit.java | 23 +-
.../org/apache/paimon/catalog/SnapshotCommit.java | 55 +-
.../jdbc/AbstractDistributedLockDialect.java | 163 ++++-
.../java/org/apache/paimon/jdbc/JdbcCatalog.java | 31 +-
.../org/apache/paimon/jdbc/JdbcCatalogLock.java | 194 ++++-
.../apache/paimon/jdbc/JdbcCatalogLockFactory.java | 14 +-
.../org/apache/paimon/jdbc/JdbcCatalogOptions.java | 28 +
.../paimon/jdbc/MysqlDistributedLockDialect.java | 28 +-
.../jdbc/PostgresqlDistributedLockDialect.java | 29 +-
.../paimon/jdbc/SqlLiteDistributedLockDialect.java | 15 +-
.../paimon/operation/FileStoreCommitImpl.java | 203 +++++-
.../java/org/apache/paimon/operation/Lock.java | 135 +++-
.../apache/paimon/table/CatalogEnvironment.java | 61 +-
.../dataevolution/CompactReassignReuseTest.java | 27 +-
.../DataEvolutionRowIdReassignerTest.java | 9 +-
.../catalog/CatalogLockConfigurationTest.java | 116 +++
.../paimon/catalog/CatalogSnapshotCommitTest.java | 4 +-
.../apache/paimon/catalog/SnapshotCommitTest.java | 199 +++++
.../paimon/catalog/TestCatalogLockFactory.java | 58 ++
.../jdbc/JdbcCatalogLockConfigurationTest.java | 145 ++++
.../apache/paimon/jdbc/JdbcCatalogLockTest.java | 561 +++++++++++++++
.../org/apache/paimon/jdbc/JdbcCatalogTest.java | 8 +-
.../apache/paimon/jdbc/PostgresqlCatalogTest.java | 4 +-
.../apache/paimon/operation/FileDeletionTest.java | 3 +-
.../paimon/operation/FileStoreCommitTest.java | 801 ++++++++++++++++++++-
.../java/org/apache/paimon/operation/LockTest.java | 291 ++++++++
.../paimon/table/CatalogEnvironmentTest.java | 252 +++++++
.../services/org.apache.paimon.factories.Factory | 2 +
.../paimon/flink/FileSystemCatalogITCase.java | 24 +-
.../apache/paimon/flink/JdbcCatalogLockITCase.java | 86 +++
.../apache/paimon/flink/JdbcCatalogViewITCase.java | 3 +-
.../java/org/apache/paimon/hive/HiveCatalog.java | 27 +-
.../org/apache/paimon/hive/HiveCatalogLock.java | 178 ++++-
.../apache/paimon/hive/HiveCatalogLockFactory.java | 17 +-
.../org/apache/paimon/hive/HiveCatalogOptions.java | 27 +
.../hive/HiveCatalogLockConfigurationTest.java | 127 ++++
.../apache/paimon/hive/HiveCatalogLockTest.java | 350 +++++++++
.../apache/paimon/hive/HiveCatalogITCaseBase.java | 13 +-
.../org/apache/paimon/hive/HiveLocationTest.java | 2 +-
52 files changed, 4243 insertions(+), 356 deletions(-)
diff --git a/docs/docs/concepts/concurrency-control.md b/docs/docs/concepts/concurrency-control.md
index cdc1f85d4a..25f5fbe62f 100644
--- a/docs/docs/concepts/concurrency-control.md
+++ b/docs/docs/concepts/concurrency-control.md
@@ -72,6 +72,35 @@ The commit mechanism depends on the catalog and storage:
All writers of the same table must use a compatible commit mechanism and shared locking
configuration. See [Catalog](./catalog) when choosing the metadata backend.
+## Commit Locks
+
+Configured catalog locks protect every snapshot publication attempt, including the first attempt.
+The writer acquires the lock before reading the latest snapshot, validating changes and preparing
+manifests. It checks that the lock is still valid immediately before publishing. Each snapshot has
+its own scope: APPEND and COMPACT snapshots produced by the same commit acquire separate locks.
+Post-commit callbacks run after the scope ends. Metadata-only manifest replacement validates the
+caller's prepared layout against the current head while holding the lock.
+
+`Catalog.createLock` supplies a table-bound `Lock` with one `runWithLock` contract. `CatalogLockFactory`
+binds the table, UUID and writer identity when creating a `Lock`.
+`SnapshotCommit` owns this lock together with its publisher resources. Its `commit` method always
+uses the bound lock, including direct publication. Writers use `SnapshotCommit.runWithLock` to
+cover head reads and preparation; publication reuses that scope and revalidates the lease.
+Each catalog interprets its own lock configuration. Hive and JDBC define the catalog options
+`lock.enabled`, `lock.type`, `lock-check-max-sleep` and `lock-acquire-timeout` in their own options
+classes. For these catalogs, locking is enabled on object stores and disabled on other filesystems
+unless explicitly set. The filesystem catalog does not interpret these options. Catalogs without
+a locking policy use a no-op lock.
+Hive locks send heartbeats and validate ownership before publication. The client's
+`hive.txn.timeout` must match the metastore timeout. JDBC locks renew their
+leases and use an internal owner ID for renewal and release. Existing JDBC lock tables receive a
+nullable `lock_owner` column when the catalog initializes; this requires permission to alter that
+table. Older clients can still read the table, but all writers should upgrade before relying on
+owner-checked release after lease expiry. Each active Hive or JDBC scope owns a renewal worker
+which is stopped when the scope ends, so blocked renewals do not occupy other scopes' scheduling
+threads. Filesystem locks still depend on the storage's atomic publication behavior; they do not
+provide a storage fencing token.
+
## Files conflict
A writer validates file-level changes as well as the snapshot ID. For example, if two compactors
diff --git a/docs/generated/catalog_configuration.html b/docs/generated/catalog_configuration.html
index eea1dd9b28..b13aeef6de 100644
--- a/docs/generated/catalog_configuration.html
+++ b/docs/generated/catalog_configuration.html
@@ -152,30 +152,6 @@ under the License.
<td>String</td>
<td>Comma-separated list of file types to cache. Supported values: meta, global-index, bucket-index, data, file-index, or * for all of them.</td>
</tr>
- <tr>
- <td><h5>lock-acquire-timeout</h5></td>
- <td style="word-wrap: break-word;">8 min</td>
- <td>Duration</td>
- <td>The maximum time to wait for acquiring the lock.</td>
- </tr>
- <tr>
- <td><h5>lock-check-max-sleep</h5></td>
- <td style="word-wrap: break-word;">8 s</td>
- <td>Duration</td>
- <td>The maximum sleep time when retrying to check the lock.</td>
- </tr>
- <tr>
- <td><h5>lock.enabled</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>Boolean</td>
- <td>Enable Catalog Lock.</td>
- </tr>
- <tr>
- <td><h5>lock.type</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>The Lock Type for Catalog, such as 'hive', 'zookeeper'.</td>
- </tr>
<tr>
<td><h5>metastore</h5></td>
<td style="word-wrap: break-word;">"filesystem"</td>
diff --git a/docs/generated/hive_catalog_configuration.html b/docs/generated/hive_catalog_configuration.html
index 0ce809098a..3568fc044d 100644
--- a/docs/generated/hive_catalog_configuration.html
+++ b/docs/generated/hive_catalog_configuration.html
@@ -77,6 +77,30 @@ If you don't want to access the location by the filesystem of hive when using a
you can set this option to true.
</td>
</tr>
+ <tr>
+ <td><h5>lock-acquire-timeout</h5></td>
+ <td style="word-wrap: break-word;">8 min</td>
+ <td>Duration</td>
+ <td>The maximum time to wait for acquiring the lock.</td>
+ </tr>
+ <tr>
+ <td><h5>lock-check-max-sleep</h5></td>
+ <td style="word-wrap: break-word;">8 s</td>
+ <td>Duration</td>
+ <td>The maximum sleep time when retrying to check the lock.</td>
+ </tr>
+ <tr>
+ <td><h5>lock.enabled</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Boolean</td>
+ <td>Enable catalog locking. Defaults to enabled on object stores.</td>
+ </tr>
+ <tr>
+ <td><h5>lock.type</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The catalog lock factory identifier. Defaults to the catalog's built-in lock factory.</td>
+ </tr>
<tr>
<td><h5>metastore.client.class</h5></td>
<td style="word-wrap: break-word;">"org.apache.hadoop.hive.metastore.HiveMetaStoreClient"</td>
diff --git a/docs/generated/jdbc_catalog_configuration.html b/docs/generated/jdbc_catalog_configuration.html
index 9b36f3f372..af3f0e7a9a 100644
--- a/docs/generated/jdbc_catalog_configuration.html
+++ b/docs/generated/jdbc_catalog_configuration.html
@@ -32,11 +32,35 @@ under the License.
<td>String</td>
<td>Custom jdbc catalog store key.</td>
</tr>
+ <tr>
+ <td><h5>lock-acquire-timeout</h5></td>
+ <td style="word-wrap: break-word;">8 min</td>
+ <td>Duration</td>
+ <td>The maximum time to wait for acquiring the lock.</td>
+ </tr>
+ <tr>
+ <td><h5>lock-check-max-sleep</h5></td>
+ <td style="word-wrap: break-word;">8 s</td>
+ <td>Duration</td>
+ <td>The maximum sleep time when retrying to check the lock.</td>
+ </tr>
<tr>
<td><h5>lock-key-max-length</h5></td>
<td style="word-wrap: break-word;">255</td>
<td>Integer</td>
<td>Set the maximum length of the lock key. The 'lock-key' is composed of concatenating three fields : 'catalog-key', 'database', and 'table'.</td>
</tr>
+ <tr>
+ <td><h5>lock.enabled</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Boolean</td>
+ <td>Enable catalog locking. Defaults to enabled on object stores.</td>
+ </tr>
+ <tr>
+ <td><h5>lock.type</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The catalog lock factory identifier. Defaults to the catalog's built-in lock factory.</td>
+ </tr>
</tbody>
</table>
diff --git a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java
index ebb46d1eeb..2321d00c6c 100644
--- a/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java
@@ -53,30 +53,6 @@ public class CatalogOptions {
.defaultValue(CatalogTableType.MANAGED)
.withDescription("Type of table.");
- public static final ConfigOption<Boolean> LOCK_ENABLED =
- ConfigOptions.key("lock.enabled")
- .booleanType()
- .noDefaultValue()
- .withDescription("Enable Catalog Lock.");
-
- public static final ConfigOption<String> LOCK_TYPE =
- ConfigOptions.key("lock.type")
- .stringType()
- .noDefaultValue()
- .withDescription("The Lock Type for Catalog, such as 'hive', 'zookeeper'.");
-
- public static final ConfigOption<Duration> LOCK_CHECK_MAX_SLEEP =
- key("lock-check-max-sleep")
- .durationType()
- .defaultValue(Duration.ofSeconds(8))
- .withDescription("The maximum sleep time when retrying to check the lock.");
-
- public static final ConfigOption<Duration> LOCK_ACQUIRE_TIMEOUT =
- key("lock-acquire-timeout")
- .durationType()
- .defaultValue(Duration.ofMinutes(8))
- .withDescription("The maximum time to wait for acquiring the lock.");
-
public static final ConfigOption<Integer> CLIENT_POOL_SIZE =
key("client-pool-size")
.intType()
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index c8719aae32..fe381cbaae 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -20,7 +20,6 @@ package org.apache.paimon;
import org.apache.paimon.CoreOptions.ExternalPathStrategy;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.catalog.RenamingSnapshotCommit;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.catalog.TableRollback;
import org.apache.paimon.data.InternalRow;
@@ -43,7 +42,6 @@ import org.apache.paimon.metastore.VisibilityWaitCallback;
import org.apache.paimon.operation.ChainTablePartitionExpire;
import org.apache.paimon.operation.ChangelogDeletion;
import org.apache.paimon.operation.FileStoreCommitImpl;
-import org.apache.paimon.operation.Lock;
import org.apache.paimon.operation.ManifestsReader;
import org.apache.paimon.operation.NormalPartitionExpire;
import org.apache.paimon.operation.PartitionExpire;
@@ -290,10 +288,8 @@ abstract class AbstractFileStore<T> implements FileStore<T> {
@Override
public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
SnapshotManager snapshotManager = snapshotManager();
- SnapshotCommit snapshotCommit = catalogEnvironment.snapshotCommit(snapshotManager);
- if (snapshotCommit == null) {
- snapshotCommit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
- }
+ SnapshotCommit snapshotCommit =
+ catalogEnvironment.snapshotCommit(snapshotManager, options, commitUser);
ConflictDetection.Factory conflictDetectFactory =
scanner ->
ConflictDetection.create(
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index e0f02de21c..08f28ce9fc 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -22,7 +22,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.PagedList;
import org.apache.paimon.Snapshot;
import org.apache.paimon.TableType;
-import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
@@ -30,6 +29,7 @@ import org.apache.paimon.fs.cache.CachingFileIO;
import org.apache.paimon.fs.cache.LocalCacheManager;
import org.apache.paimon.function.Function;
import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
@@ -80,8 +80,6 @@ import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase;
import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem;
import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable;
import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
-import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
-import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
/** Common implementation of {@link Catalog}. */
public abstract class AbstractCatalog implements Catalog {
@@ -122,19 +120,28 @@ public abstract class AbstractCatalog implements Catalog {
return fileIO;
}
+ @Override
+ public Lock createLock(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ @Nullable String commitUser,
+ Options tableOptions) {
+ return lockFactory()
+ .map(
+ factory ->
+ factory.createLock(
+ lockContext().orElse(null),
+ identifier,
+ tableUuid,
+ commitUser,
+ tableOptions))
+ .map(Lock::reentrant)
+ .orElseGet(Lock::empty);
+ }
+
+ /** Lock factory selected by the catalog's own locking policy. */
public Optional<CatalogLockFactory> lockFactory() {
- if (!lockEnabled()) {
- return Optional.empty();
- }
-
- String lock = context.options().get(LOCK_TYPE);
- if (lock == null) {
- return defaultLockFactory();
- }
-
- return Optional.of(
- FactoryUtil.discoverFactory(
- AbstractCatalog.class.getClassLoader(), CatalogLockFactory.class, lock));
+ return Optional.empty();
}
public Optional<CatalogLockFactory> defaultLockFactory() {
@@ -146,7 +153,7 @@ public abstract class AbstractCatalog implements Catalog {
}
protected boolean lockEnabled() {
- return context.options().getOptional(LOCK_ENABLED).orElse(fileIO.isObjectStore());
+ return false;
}
protected boolean allowCustomTablePath() {
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 7e824845db..2f61c526ff 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -25,6 +25,8 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.consumer.ConsumerInfo;
import org.apache.paimon.function.Function;
import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
@@ -1321,6 +1323,28 @@ public interface Catalog extends AutoCloseable {
TableQueryAuthResult authTableQuery(Identifier identifier, @Nullable List<String> select)
throws TableNotExistException;
+ // ========================== Locking ==============================
+
+ /**
+ * Create a lock bound to the table incarnation and exact writer identity.
+ *
+ * <p>The caller owns the returned lock and must close it. A lease may end when its snapshot is
+ * published; each additional snapshot must use a new {@link Lock#runWithLock} scope. Catalogs
+ * without configured locking return a no-op lock.
+ *
+ * @param identifier table and branch to lock
+ * @param tableUuid table incarnation, required by catalogs that fence table recreation
+ * @param commitUser exact logical writer identity for commits; null for catalog operations
+ * @param tableOptions runtime table options controlling lock support
+ */
+ default Lock createLock(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ @Nullable String commitUser,
+ Options tableOptions) {
+ return Lock.empty();
+ }
+
// ==================== Catalog Information ==========================
/** Catalog options for re-creating this catalog. */
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLock.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLock.java
deleted file mode 100644
index 4f147c6aad..0000000000
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLock.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.catalog;
-
-import org.apache.paimon.annotation.Public;
-
-import java.io.Closeable;
-import java.util.concurrent.Callable;
-
-/**
- * An interface that allows source and sink to use global lock to some transaction-related things.
- *
- * @since 0.4.0
- */
-@Public
-public interface CatalogLock extends Closeable {
-
- /** Run with catalog lock. The caller should tell catalog the database and table name. */
- <T> T runWithLock(String database, String table, Callable<T> callable) throws Exception;
-}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockFactory.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockFactory.java
index d964ebda4e..2c0da82274 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockFactory.java
@@ -19,11 +19,30 @@
package org.apache.paimon.catalog;
import org.apache.paimon.factories.Factory;
+import org.apache.paimon.factories.FactoryUtil;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import javax.annotation.Nullable;
import java.io.Serializable;
-/** Factory to create {@link CatalogLock}. */
+/** Factory to create table-bound {@link Lock} instances. */
public interface CatalogLockFactory extends Factory, Serializable {
- CatalogLock createLock(CatalogLockContext context);
+ /**
+ * Create a lock bound to the table and its branch. The UUID and writer identity may be null for
+ * catalog operations; snapshot commits supply them along with the runtime table options.
+ */
+ Lock createLock(
+ CatalogLockContext context,
+ Identifier identifier,
+ @Nullable String tableUuid,
+ @Nullable String commitUser,
+ Options tableOptions);
+
+ static CatalogLockFactory discover(String identifier) {
+ return FactoryUtil.discoverFactory(
+ CatalogLockFactory.class.getClassLoader(), CatalogLockFactory.class, identifier);
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java
index fc50e33b24..3cec803449 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java
@@ -19,6 +19,7 @@
package org.apache.paimon.catalog;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.partition.PartitionStatistics;
import javax.annotation.Nullable;
@@ -26,20 +27,22 @@ import javax.annotation.Nullable;
import java.util.List;
/** A {@link SnapshotCommit} using {@link Catalog} to commit. */
-public class CatalogSnapshotCommit implements SnapshotCommit {
+public class CatalogSnapshotCommit extends SnapshotCommit {
private final Catalog catalog;
private final Identifier identifier;
@Nullable private final String uuid;
- public CatalogSnapshotCommit(Catalog catalog, Identifier identifier, @Nullable String uuid) {
+ public CatalogSnapshotCommit(
+ Catalog catalog, Identifier identifier, @Nullable String uuid, Lock lock) {
+ super(lock);
this.catalog = catalog;
this.identifier = identifier;
this.uuid = uuid;
}
@Override
- public boolean commit(
+ protected boolean commitImpl(
@Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
@@ -51,7 +54,7 @@ public class CatalogSnapshotCommit implements SnapshotCommit {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
catalog.close();
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index bd342e1013..41052b8baf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -22,6 +22,8 @@ import org.apache.paimon.PagedList;
import org.apache.paimon.Snapshot;
import org.apache.paimon.function.Function;
import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
@@ -51,6 +53,15 @@ public abstract class DelegateCatalog implements Catalog {
this.wrapped = wrapped;
}
+ @Override
+ public Lock createLock(
+ Identifier identifier,
+ @Nullable String tableUuid,
+ @Nullable String commitUser,
+ Options tableOptions) {
+ return wrapped.createLock(identifier, tableUuid, commitUser, tableOptions);
+ }
+
public Catalog wrapped() {
return wrapped;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
index cb68850a19..771dbdfe32 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
@@ -21,6 +21,7 @@ package org.apache.paimon.catalog;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
@@ -33,7 +34,6 @@ import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.concurrent.Callable;
import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE;
@@ -140,12 +140,7 @@ public class FileSystemCatalog extends AbstractCatalog {
}
public <T> T runWithLock(Identifier identifier, Callable<T> callable) throws Exception {
- Optional<CatalogLockFactory> lockFactory = lockFactory();
- try (Lock lock =
- lockFactory
- .map(factory -> factory.createLock(lockContext().orElse(null)))
- .map(l -> Lock.fromCatalog(l, identifier))
- .orElseGet(Lock::empty)) {
+ try (Lock lock = createLock(identifier, null, null, new Options())) {
return lock.runWithLock(callable);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
index b880d5c4a2..16d577ecc7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
@@ -37,20 +37,19 @@ import java.util.concurrent.Callable;
* <p>Note that when the file system is local or HDFS, rename is atomic. But if the file system is
* object storage, we need additional lock protection.
*/
-public class RenamingSnapshotCommit implements SnapshotCommit {
+public class RenamingSnapshotCommit extends SnapshotCommit {
private final SnapshotManager snapshotManager;
private final FileIO fileIO;
- private final Lock lock;
public RenamingSnapshotCommit(SnapshotManager snapshotManager, Lock lock) {
+ super(lock);
this.snapshotManager = snapshotManager;
this.fileIO = snapshotManager.fileIO();
- this.lock = lock;
}
@Override
- public boolean commit(
+ protected boolean commitImpl(
@Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
@@ -93,18 +92,8 @@ public class RenamingSnapshotCommit implements SnapshotCommit {
}
return committed;
};
- return lock.runWithLock(
- () ->
- // fs.rename may not returns false if target file
- // already exists, or even not atomic
- // as we're relying on external locking, we can first
- // check if file exist then rename to work around this
- // case
- !fileIO.exists(newSnapshotPath) && callable.call());
- }
-
- @Override
- public void close() throws Exception {
- this.lock.close();
+ // fs.rename may not return false if the target file already exists, or even be atomic.
+ // The bound lock lets us check existence before publishing.
+ return !fileIO.exists(newSnapshotPath) && callable.call();
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
index ff912e2139..e4f0f5e47c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
@@ -19,19 +19,68 @@
package org.apache.paimon.catalog;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.partition.PartitionStatistics;
import javax.annotation.Nullable;
import java.util.List;
+import java.util.concurrent.Callable;
-/** Interface to commit snapshot atomically. */
-public interface SnapshotCommit extends AutoCloseable {
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
- boolean commit(
+/** A snapshot publisher bound to the lock protecting its preparation and publication. */
+public abstract class SnapshotCommit implements Lock {
+
+ private final Lock lock;
+
+ protected SnapshotCommit(Lock lock) {
+ this.lock =
+ Lock.reentrant(checkNotNull(lock, "Snapshot publication requires a bound lock."));
+ }
+
+ /** Publish with the bound lock, reusing an enclosing preparation scope when present. */
+ public final boolean commit(
+ @Nullable String baseSnapshotUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ return runWithLock(() -> commitImpl(baseSnapshotUuid, snapshot, branch, statistics));
+ }
+
+ @Override
+ public final AutoCloseable acquire() throws Exception {
+ return lock.acquire();
+ }
+
+ @Override
+ public final <T> T runWithLock(Callable<T> callable) throws Exception {
+ return lock.runWithLock(callable);
+ }
+
+ @Override
+ public final void ensureValid() {
+ lock.ensureValid();
+ }
+
+ @Override
+ public final void close() throws Exception {
+ try {
+ lock.close();
+ } finally {
+ closeImpl();
+ }
+ }
+
+ /** Backend publication, invoked only after acquiring and validating the bound lock. */
+ protected abstract boolean commitImpl(
@Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
List<PartitionStatistics> statistics)
throws Exception;
+
+ /** Release backend resources after closing the lock, including when lock closure fails. */
+ protected void closeImpl() throws Exception {}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/AbstractDistributedLockDialect.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/AbstractDistributedLockDialect.java
index b8bf3e62d8..ba008cd45f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/AbstractDistributedLockDialect.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/AbstractDistributedLockDialect.java
@@ -20,9 +20,11 @@ package org.apache.paimon.jdbc;
import org.apache.paimon.options.Options;
+import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
import java.sql.SQLException;
/** Jdbc distributed lock interface. */
@@ -35,18 +37,55 @@ public abstract class AbstractDistributedLockDialect implements JdbcDistributedL
connections.run(
conn -> {
DatabaseMetaData dbMeta = conn.getMetaData();
- ResultSet tableExists =
+ try (ResultSet tableExists =
dbMeta.getTables(
- null, null, JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME, null);
- if (tableExists.next()) {
- return true;
+ null, null, JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME, null)) {
+ if (!tableExists.next()) {
+ try (PreparedStatement statement =
+ conn.prepareStatement(
+ String.format(getCreateTableSql(), lockKeyMaxLength))) {
+ statement.execute();
+ }
+ }
}
- String createDistributedLockTableSql =
- String.format(getCreateTableSql(), lockKeyMaxLength);
- return conn.prepareStatement(createDistributedLockTableSql).execute();
+ // Older catalogs created the lock table without an owner column.
+ if (!hasOwnerColumn(conn)) {
+ try (PreparedStatement statement =
+ conn.prepareStatement(
+ "ALTER TABLE "
+ + JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
+ + " ADD COLUMN lock_owner VARCHAR(36)")) {
+ statement.execute();
+ } catch (SQLException e) {
+ // Another catalog may have upgraded the table concurrently.
+ if (!hasOwnerColumn(conn)) {
+ throw e;
+ }
+ }
+ }
+ return true;
});
}
+ private boolean hasOwnerColumn(Connection connection) throws SQLException {
+ // Resolve the same table as the lock SQL. JDBC metadata patterns can match other
+ // catalogs, schemas, tables or columns, particularly because these names contain '_'.
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT * FROM "
+ + JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
+ + " WHERE 1 = 0");
+ ResultSet result = statement.executeQuery()) {
+ ResultSetMetaData columns = result.getMetaData();
+ for (int i = 1; i <= columns.getColumnCount(); i++) {
+ if ("lock_owner".equalsIgnoreCase(columns.getColumnName(i))) {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
public abstract String getCreateTableSql();
@Override
@@ -65,6 +104,116 @@ public abstract class AbstractDistributedLockDialect implements JdbcDistributedL
});
}
+ boolean acquireOwned(JdbcClientPool connections, String lockId, String owner, long leaseMillis)
+ throws SQLException, InterruptedException {
+ tryReleaseTimedOutLock(connections, lockId);
+ return connections.run(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "INSERT INTO "
+ + JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
+ + " (lock_id, expire_time_seconds, lock_owner) VALUES (?, ?, ?)")) {
+ statement.setString(1, lockId);
+ // A second of precision is lost by the timestamp column; keep a
+ // conservative
+ // millisecond deadline in the holder and round the database TTL up.
+ statement.setLong(2, Math.max(1, leaseMillis / 1000 + 2));
+ statement.setString(3, owner);
+ return statement.executeUpdate() > 0;
+ } catch (SQLException e) {
+ if (isDuplicateKey(e)) {
+ return false;
+ }
+ throw e;
+ }
+ });
+ }
+
+ private boolean isDuplicateKey(SQLException e) {
+ return "23505".equals(e.getSQLState())
+ || (e.getErrorCode() == 1062 && "23000".equals(e.getSQLState()))
+ || e.getErrorCode() == 1555
+ || e.getErrorCode() == 2067
+ || (e.getErrorCode() == 19
+ && e.getMessage() != null
+ && e.getMessage().contains("UNIQUE constraint failed"));
+ }
+
+ protected String getExpirationCondition() {
+ throw new UnsupportedOperationException("Owned leases are not supported by this dialect.");
+ }
+
+ protected String getRenewalTime() {
+ return "CURRENT_TIMESTAMP";
+ }
+
+ protected String getOwnedExpirationCondition() {
+ return getExpirationCondition();
+ }
+
+ boolean renewOwned(JdbcClientPool connections, String lockId, String owner)
+ throws SQLException, InterruptedException {
+ boolean renewed =
+ updateOwned(
+ connections,
+ "UPDATE "
+ + JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
+ + " SET acquired_at = CASE WHEN acquired_at > "
+ + getRenewalTime()
+ + " THEN acquired_at ELSE "
+ + getRenewalTime()
+ + " END WHERE lock_id = ? AND lock_owner = ? AND NOT ("
+ + getOwnedExpirationCondition()
+ + ")",
+ lockId,
+ owner);
+ if (renewed) {
+ return true;
+ }
+ // Drivers configured to report changed rows can return zero when a timestamp with
+ // second precision has not changed. Confirm that the same unexpired lease still exists.
+ return connections.run(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT 1 FROM "
+ + JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
+ + " WHERE lock_id = ? AND lock_owner = ? AND NOT ("
+ + getOwnedExpirationCondition()
+ + ")")) {
+ statement.setString(1, lockId);
+ statement.setString(2, owner);
+ try (ResultSet result = statement.executeQuery()) {
+ return result.next();
+ }
+ }
+ });
+ }
+
+ boolean releaseOwned(JdbcClientPool connections, String lockId, String owner)
+ throws SQLException, InterruptedException {
+ return updateOwned(
+ connections,
+ "DELETE FROM "
+ + JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
+ + " WHERE lock_id = ? AND lock_owner = ?",
+ lockId,
+ owner);
+ }
+
+ private boolean updateOwned(JdbcClientPool connections, String sql, String lockId, String owner)
+ throws SQLException, InterruptedException {
+ return connections.run(
+ connection -> {
+ try (PreparedStatement statement = connection.prepareStatement(sql)) {
+ statement.setString(1, lockId);
+ statement.setString(2, owner);
+ return statement.executeUpdate() > 0;
+ }
+ });
+ }
+
public abstract String getLockAcquireSql();
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
index e73db8a5c0..3dd830a7ff 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
@@ -83,8 +83,8 @@ import static org.apache.paimon.catalog.CatalogUtils.checkNotBranch;
import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemDatabase;
import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable;
import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable;
-import static org.apache.paimon.jdbc.JdbcCatalogLock.acquireTimeout;
-import static org.apache.paimon.jdbc.JdbcCatalogLock.checkMaxSleep;
+import static org.apache.paimon.jdbc.JdbcCatalogOptions.LOCK_ENABLED;
+import static org.apache.paimon.jdbc.JdbcCatalogOptions.LOCK_TYPE;
import static org.apache.paimon.jdbc.JdbcUtils.deleteProperties;
import static org.apache.paimon.jdbc.JdbcUtils.execute;
import static org.apache.paimon.jdbc.JdbcUtils.insertProperties;
@@ -758,6 +758,20 @@ public class JdbcCatalog extends AbstractCatalog {
return false;
}
+ @Override
+ public Optional<CatalogLockFactory> lockFactory() {
+ if (!lockEnabled()) {
+ return Optional.empty();
+ }
+ String type = options.get(LOCK_TYPE);
+ return type == null ? defaultLockFactory() : Optional.of(CatalogLockFactory.discover(type));
+ }
+
+ @Override
+ protected boolean lockEnabled() {
+ return options.getOptional(LOCK_ENABLED).orElse(fileIO.isObjectStore());
+ }
+
@Override
public Optional<CatalogLockFactory> defaultLockFactory() {
return Optional.of(new JdbcCatalogLockFactory());
@@ -775,17 +789,8 @@ public class JdbcCatalog extends AbstractCatalog {
// check-then-act.
java.util.concurrent.locks.Lock localLock = LOCAL_LOCKS.get(lockKey(identifier));
localLock.lock();
- try {
- if (!lockEnabled()) {
- return callable.call();
- }
- JdbcCatalogLock lock =
- new JdbcCatalogLock(
- connections,
- catalogKey,
- checkMaxSleep(options.toMap()),
- acquireTimeout(options.toMap()));
- return Lock.fromCatalog(lock, identifier).runWithLock(callable);
+ try (Lock lock = createLock(identifier, null, null, new Options())) {
+ return lock.runWithLock(callable);
} finally {
localLock.unlock();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLock.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLock.java
index 307f92f0a5..f4f497f6eb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLock.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLock.java
@@ -18,66 +18,206 @@
package org.apache.paimon.jdbc;
-import org.apache.paimon.catalog.CatalogLock;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.utils.ExecutorThreadFactory;
import org.apache.paimon.utils.TimeUtils;
+import javax.annotation.Nullable;
+
import java.io.IOException;
-import java.sql.SQLException;
import java.time.Duration;
import java.util.Map;
-import java.util.concurrent.Callable;
+import java.util.UUID;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongSupplier;
-import static org.apache.paimon.options.CatalogOptions.LOCK_ACQUIRE_TIMEOUT;
-import static org.apache.paimon.options.CatalogOptions.LOCK_CHECK_MAX_SLEEP;
+import static org.apache.paimon.jdbc.JdbcCatalogOptions.LOCK_ACQUIRE_TIMEOUT;
+import static org.apache.paimon.jdbc.JdbcCatalogOptions.LOCK_CHECK_MAX_SLEEP;
/** Jdbc catalog lock. */
-public class JdbcCatalogLock implements CatalogLock {
+public class JdbcCatalogLock implements Lock {
+ @Nullable private final ScheduledExecutorService renewer;
+ private final LongSupplier nanoTime;
+ private final ThreadLocal<Lease> currentLease = new ThreadLocal<>();
private final JdbcClientPool connections;
private final long checkMaxSleep;
private final long acquireTimeout;
private final String catalogKey;
+ private final Identifier identifier;
public JdbcCatalogLock(
JdbcClientPool connections,
String catalogKey,
+ Identifier identifier,
long checkMaxSleep,
long acquireTimeout) {
+ this(
+ connections,
+ catalogKey,
+ identifier,
+ checkMaxSleep,
+ acquireTimeout,
+ null,
+ System::nanoTime);
+ }
+
+ JdbcCatalogLock(
+ JdbcClientPool connections,
+ String catalogKey,
+ Identifier identifier,
+ long checkMaxSleep,
+ long acquireTimeout,
+ @Nullable ScheduledExecutorService renewer,
+ LongSupplier nanoTime) {
+ this.renewer = renewer;
+ this.nanoTime = nanoTime;
this.connections = connections;
this.checkMaxSleep = checkMaxSleep;
this.acquireTimeout = acquireTimeout;
this.catalogKey = catalogKey;
+ this.identifier = identifier;
}
@Override
- public <T> T runWithLock(String database, String table, Callable<T> callable) throws Exception {
- String lockUniqueName = String.format("%s.%s.%s", catalogKey, database, table);
- lock(lockUniqueName);
+ public AutoCloseable acquire() throws Exception {
+ String lockUniqueName =
+ String.format(
+ "%s.%s.%s",
+ catalogKey, identifier.getDatabaseName(), identifier.getObjectName());
+ String owner = UUID.randomUUID().toString();
+ long startedAt = nanoTime.getAsLong();
+ AbstractDistributedLockDialect dialect =
+ (AbstractDistributedLockDialect)
+ DistributedLockDialectFactory.create(connections.getProtocol());
+ long nextSleep = 50;
+ while (true) {
+ long requestedAt = nanoTime.getAsLong();
+ if (dialect.acquireOwned(connections, lockUniqueName, owner, acquireTimeout)) {
+ Lease lease = null;
+ try {
+ lease = new Lease(dialect, lockUniqueName, owner, requestedAt);
+ currentLease.set(lease);
+ lease.ensureValid();
+ Lease acquired = lease;
+ return () -> release(acquired, dialect, lockUniqueName, owner);
+ } catch (Exception | Error e) {
+ Lease partial = lease;
+ Lock.release(() -> release(partial, dialect, lockUniqueName, owner), e);
+ throw e;
+ }
+ }
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(nanoTime.getAsLong() - startedAt);
+ if (elapsed >= acquireTimeout) {
+ throw new IllegalStateException(
+ "Acquire lock failed with time: " + Duration.ofMillis(elapsed));
+ }
+ nextSleep = Math.min(nextSleep * 2, checkMaxSleep);
+ Thread.sleep(Math.min(nextSleep, acquireTimeout - elapsed));
+ }
+ }
+
+ private void release(
+ @Nullable Lease lease,
+ AbstractDistributedLockDialect dialect,
+ String lockUniqueName,
+ String owner)
+ throws Exception {
+ currentLease.remove();
try {
- return callable.call();
+ if (lease != null) {
+ lease.close();
+ }
} finally {
- JdbcUtils.release(connections, lockUniqueName);
+ // The former holder must never delete a lease acquired after its expiry.
+ dialect.releaseOwned(connections, lockUniqueName, owner);
}
}
- private void lock(String lockUniqueName) throws SQLException, InterruptedException {
- boolean lock = JdbcUtils.acquire(connections, lockUniqueName, acquireTimeout);
- long nextSleep = 50;
- long startRetry = System.currentTimeMillis();
- while (!lock) {
- nextSleep *= 2;
- if (nextSleep > checkMaxSleep) {
- nextSleep = checkMaxSleep;
+ @Override
+ public void ensureValid() {
+ Lease lease = currentLease.get();
+ if (lease == null) {
+ throw new IllegalStateException("No JDBC lock is held by this thread.");
+ }
+ lease.renew();
+ lease.ensureValid();
+ }
+
+ private static ScheduledThreadPoolExecutor createRenewer() {
+ ScheduledThreadPoolExecutor executor =
+ new ScheduledThreadPoolExecutor(1, new ExecutorThreadFactory("jdbc-lock-renew"));
+ executor.setRemoveOnCancelPolicy(true);
+ return executor;
+ }
+
+ private class Lease implements AutoCloseable {
+ private final AtomicBoolean closed = new AtomicBoolean();
+ private final ScheduledFuture<?> renewal;
+ private final ScheduledExecutorService executor;
+ private final AbstractDistributedLockDialect dialect;
+ private final String lockId;
+ private final String owner;
+ private final AtomicLong renewedAt;
+ private volatile Exception failure;
+
+ private Lease(
+ AbstractDistributedLockDialect dialect,
+ String lockId,
+ String owner,
+ long requestedAt) {
+ this.dialect = dialect;
+ this.lockId = lockId;
+ this.owner = owner;
+ this.renewedAt = new AtomicLong(requestedAt);
+ // Blocking database calls must not consume another scope's renewal worker.
+ this.executor = renewer == null ? createRenewer() : renewer;
+ long interval = Math.max(1, acquireTimeout / 3);
+ this.renewal =
+ executor.scheduleWithFixedDelay(
+ this::renew, interval, interval, TimeUnit.MILLISECONDS);
+ }
+
+ private void renew() {
+ if (closed.get() || failure != null) {
+ return;
}
- Thread.sleep(nextSleep);
- lock = JdbcUtils.acquire(connections, lockUniqueName, acquireTimeout);
- if (System.currentTimeMillis() - startRetry > acquireTimeout) {
- break;
+ try {
+ ensureValid();
+ long requestedAt = nanoTime.getAsLong();
+ if (dialect.renewOwned(connections, lockId, owner)) {
+ renewedAt.accumulateAndGet(requestedAt, Math::max);
+ } else {
+ failure = new IllegalStateException("JDBC lock ownership was lost.");
+ }
+ } catch (Exception e) {
+ failure = e;
}
}
- long retryDuration = System.currentTimeMillis() - startRetry;
- if (!lock) {
- throw new RuntimeException(
- "Acquire lock failed with time: " + Duration.ofMillis(retryDuration));
+
+ private void ensureValid() {
+ if (TimeUnit.NANOSECONDS.toMillis(nanoTime.getAsLong() - renewedAt.get())
+ >= acquireTimeout
+ && failure == null) {
+ failure = new IllegalStateException("JDBC lock expired locally.");
+ }
+ if (closed.get() || failure != null) {
+ throw new IllegalStateException("The JDBC lock is no longer usable.", failure);
+ }
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ renewal.cancel(true);
+ if (renewer == null) {
+ executor.shutdownNow();
+ }
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockFactory.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockFactory.java
index ce0a2d24ee..9c673809dd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockFactory.java
@@ -18,9 +18,13 @@
package org.apache.paimon.jdbc;
-import org.apache.paimon.catalog.CatalogLock;
import org.apache.paimon.catalog.CatalogLockContext;
import org.apache.paimon.catalog.CatalogLockFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import javax.annotation.Nullable;
import java.util.Map;
@@ -40,12 +44,18 @@ public class JdbcCatalogLockFactory implements CatalogLockFactory {
}
@Override
- public CatalogLock createLock(CatalogLockContext context) {
+ public Lock createLock(
+ CatalogLockContext context,
+ Identifier identifier,
+ @Nullable String tableUuid,
+ @Nullable String commitUser,
+ Options tableOptions) {
JdbcCatalogLockContext lockContext = (JdbcCatalogLockContext) context;
Map<String, String> optionsMap = lockContext.options().toMap();
return new JdbcCatalogLock(
lockContext.connections(),
lockContext.catalogKey(),
+ identifier,
checkMaxSleep(optionsMap),
acquireTimeout(optionsMap));
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogOptions.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogOptions.java
index 407dbbc3bf..af7b9434f3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogOptions.java
@@ -22,9 +22,37 @@ import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.ConfigOptions;
import org.apache.paimon.options.Options;
+import java.time.Duration;
+
/** Options for jdbc catalog. */
public final class JdbcCatalogOptions {
+ public static final ConfigOption<Boolean> LOCK_ENABLED =
+ ConfigOptions.key("lock.enabled")
+ .booleanType()
+ .noDefaultValue()
+ .withDescription(
+ "Enable catalog locking. Defaults to enabled on object stores.");
+
+ public static final ConfigOption<String> LOCK_TYPE =
+ ConfigOptions.key("lock.type")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The catalog lock factory identifier. Defaults to the catalog's built-in lock factory.");
+
+ public static final ConfigOption<Duration> LOCK_CHECK_MAX_SLEEP =
+ ConfigOptions.key("lock-check-max-sleep")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(8))
+ .withDescription("The maximum sleep time when retrying to check the lock.");
+
+ public static final ConfigOption<Duration> LOCK_ACQUIRE_TIMEOUT =
+ ConfigOptions.key("lock-acquire-timeout")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(8))
+ .withDescription("The maximum time to wait for acquiring the lock.");
+
public static final ConfigOption<String> CATALOG_KEY =
ConfigOptions.key("catalog-key")
.stringType()
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/MysqlDistributedLockDialect.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/MysqlDistributedLockDialect.java
index bca8aac20b..3ca70fc7d4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/MysqlDistributedLockDialect.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/MysqlDistributedLockDialect.java
@@ -32,6 +32,7 @@ public class MysqlDistributedLockDialect extends AbstractDistributedLockDialect
+ " TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,"
+ JdbcUtils.EXPIRE_TIME
+ " BIGINT DEFAULT 0 NOT NULL,"
+ + "lock_owner VARCHAR(36),"
+ "PRIMARY KEY ("
+ JdbcUtils.LOCK_ID
+ ")"
@@ -62,12 +63,31 @@ public class MysqlDistributedLockDialect extends AbstractDistributedLockDialect
public String getTryReleaseTimedOutLock() {
return "DELETE FROM "
+ JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
- + " WHERE TIMESTAMPDIFF(SECOND, "
- + JdbcUtils.ACQUIRED_AT
- + ", NOW()) >"
- + JdbcUtils.EXPIRE_TIME
+ + " WHERE "
+ + getExpirationCondition()
+ " and "
+ JdbcUtils.LOCK_ID
+ " = ?";
}
+
+ @Override
+ protected String getRenewalTime() {
+ return "SYSDATE()";
+ }
+
+ @Override
+ protected String getOwnedExpirationCondition() {
+ return "TIMESTAMPDIFF(SECOND, "
+ + JdbcUtils.ACQUIRED_AT
+ + ", SYSDATE()) > "
+ + JdbcUtils.EXPIRE_TIME;
+ }
+
+ @Override
+ protected String getExpirationCondition() {
+ return "TIMESTAMPDIFF(SECOND, "
+ + JdbcUtils.ACQUIRED_AT
+ + ", NOW()) >"
+ + JdbcUtils.EXPIRE_TIME;
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/PostgresqlDistributedLockDialect.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/PostgresqlDistributedLockDialect.java
index 93f5f350dd..501bcebeb8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/PostgresqlDistributedLockDialect.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/PostgresqlDistributedLockDialect.java
@@ -32,6 +32,7 @@ public class PostgresqlDistributedLockDialect extends AbstractDistributedLockDia
+ " TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,"
+ JdbcUtils.EXPIRE_TIME
+ " BIGINT DEFAULT 0 NOT NULL,"
+ + "lock_owner VARCHAR(36),"
+ "PRIMARY KEY ("
+ JdbcUtils.LOCK_ID
+ ")"
@@ -62,12 +63,32 @@ public class PostgresqlDistributedLockDialect extends AbstractDistributedLockDia
public String getTryReleaseTimedOutLock() {
return "DELETE FROM "
+ JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
- + " WHERE EXTRACT(EPOCH FROM AGE(NOW(), "
- + JdbcUtils.ACQUIRED_AT
- + ")) >"
- + JdbcUtils.EXPIRE_TIME
+ + " WHERE "
+ + getExpirationCondition()
+ " and "
+ JdbcUtils.LOCK_ID
+ " = ?";
}
+
+ @Override
+ protected String getRenewalTime() {
+ // CURRENT_TIMESTAMP is fixed at transaction start, including time waiting for a row lock.
+ return "clock_timestamp()";
+ }
+
+ @Override
+ protected String getOwnedExpirationCondition() {
+ return "EXTRACT(EPOCH FROM (clock_timestamp() - "
+ + JdbcUtils.ACQUIRED_AT
+ + ")) > "
+ + JdbcUtils.EXPIRE_TIME;
+ }
+
+ @Override
+ protected String getExpirationCondition() {
+ return "EXTRACT(EPOCH FROM AGE(NOW(), "
+ + JdbcUtils.ACQUIRED_AT
+ + ")) >"
+ + JdbcUtils.EXPIRE_TIME;
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/SqlLiteDistributedLockDialect.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/SqlLiteDistributedLockDialect.java
index 1ffdd61bc2..89a43186a6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/SqlLiteDistributedLockDialect.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/SqlLiteDistributedLockDialect.java
@@ -31,6 +31,7 @@ public class SqlLiteDistributedLockDialect extends AbstractDistributedLockDialec
+ " TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,"
+ JdbcUtils.EXPIRE_TIME
+ " BIGINT DEFAULT 0 NOT NULL,"
+ + "lock_owner VARCHAR(36),"
+ "PRIMARY KEY ("
+ JdbcUtils.LOCK_ID
+ ")"
@@ -61,12 +62,18 @@ public class SqlLiteDistributedLockDialect extends AbstractDistributedLockDialec
public String getTryReleaseTimedOutLock() {
return "DELETE FROM "
+ JdbcUtils.DISTRIBUTED_LOCKS_TABLE_NAME
- + " WHERE strftime('%s', 'now') - strftime('%s', "
- + JdbcUtils.ACQUIRED_AT
- + ") > "
- + JdbcUtils.EXPIRE_TIME
+ + " WHERE "
+ + getExpirationCondition()
+ " and "
+ JdbcUtils.LOCK_ID
+ " = ?";
}
+
+ @Override
+ protected String getExpirationCondition() {
+ return " strftime('%s', 'now') - strftime('%s', "
+ + JdbcUtils.ACQUIRED_AT
+ + ") > "
+ + JdbcUtils.EXPIRE_TIME;
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 18cc7a8edf..2123517e13 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -102,6 +102,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
+import java.util.concurrent.Callable;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -129,8 +130,8 @@ import static org.apache.paimon.utils.Preconditions.checkNotNull;
* user should first call {@link #filterCommitted}.
* <li>Before committing, it will first check for conflicts by checking if all files to be removed
* currently exists, and if modified files have overlapping key ranges with existing files.
- * <li>After that it use the external {@link SnapshotCommit} (if provided) or the atomic rename of
- * the file system to ensure atomicity.
+ * <li>The bound {@link SnapshotCommit} protects validation and preparation with its lock, then
+ * atomically publishes the snapshot through the catalog or file system.
* <li>If commit fails due to conflicts or exception it tries its best to clean up and aborts.
* <li>If atomic rename fails it tries again after reading the latest snapshot from step 2.
* </ol>
@@ -240,6 +241,16 @@ public class FileStoreCommitImpl implements FileStoreCommit {
this.commitCleaner = new CommitCleaner(manifestList, manifestFile, indexManifestFile);
}
+ private <T> T runWithLock(Callable<T> callable) {
+ try {
+ return snapshotCommit.runWithLock(callable);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to run commit with table lock.", e);
+ }
+ }
+
@Override
public FileStoreCommit withIOManager(IOManager ioManager) {
this.ioManager = ioManager;
@@ -894,25 +905,34 @@ public class FileStoreCommitImpl implements FileStoreCommit {
RetryCommitResult retryResult = null;
long startMillis = System.currentTimeMillis();
while (true) {
- Snapshot latestSnapshot = snapshotManager.latestSnapshot();
- CommitResult result =
- tryCommitOnce(
- retryResult,
- changesProvider,
- identifier,
- watermark,
- properties,
- commitKind,
- allowRollback,
- latestSnapshot,
- detectConflicts,
- statsFileName);
+ Exception failure;
+ try {
+ CommitResult result =
+ tryCommitOnce(
+ retryResult,
+ changesProvider,
+ identifier,
+ watermark,
+ properties,
+ commitKind,
+ allowRollback,
+ snapshotManager::latestSnapshot,
+ detectConflicts,
+ statsFileName);
+ notifyCommitSuccess(result);
- if (result.isSuccess()) {
- break;
- }
+ if (result.isSuccess()) {
+ break;
+ }
- retryResult = (RetryCommitResult) result;
+ retryResult = (RetryCommitResult) result;
+ failure = retryResult.exception;
+ } catch (Lock.AcquisitionException e) {
+ // Nothing was prepared or published. Retain any previous publication retry
+ // state, including its head, conflict scan and merged manifests.
+ failure = e;
+ LOG.warn("Failed to acquire commit lock for table {}.", tableName, e);
+ }
if (System.currentTimeMillis() - startMillis > options.commitTimeout()
|| retryCount >= options.commitMaxRetries()) {
@@ -920,7 +940,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
String.format(
"Commit failed for table %s after %s millis with %s retries, there maybe exist commit conflicts between multiple jobs.",
tableName, options.commitTimeout(), retryCount);
- throw new RuntimeException(message, retryResult.exception);
+ throw new RuntimeException(message, failure);
}
retryWaiter.retryWait(retryCount);
@@ -1027,21 +1047,58 @@ public class FileStoreCommitImpl implements FileStoreCommit {
boolean allowRollback,
@Nullable Snapshot latestSnapshot,
boolean detectConflicts,
- @Nullable String newStatsFileName) {
- return tryCommitOnce(
- retryResult,
- CommitChangesProvider.provider(deltaFiles, changelogFiles, indexFiles),
- identifier,
- watermark,
- properties,
- commitKind,
- allowRollback,
- latestSnapshot,
- detectConflicts,
- newStatsFileName);
+ @Nullable String newStatsFileName)
+ throws Lock.AcquisitionException {
+ CommitResult result =
+ tryCommitOnce(
+ retryResult,
+ CommitChangesProvider.provider(deltaFiles, changelogFiles, indexFiles),
+ identifier,
+ watermark,
+ properties,
+ commitKind,
+ allowRollback,
+ () -> latestSnapshot,
+ detectConflicts,
+ newStatsFileName);
+ notifyCommitSuccess(result);
+ return result;
}
private CommitResult tryCommitOnce(
+ @Nullable RetryCommitResult retryResult,
+ CommitChangesProvider changesProvider,
+ long identifier,
+ @Nullable Long watermark,
+ Map<String, String> properties,
+ CommitKind commitKind,
+ boolean allowRollback,
+ Supplier<Snapshot> latestSnapshotSupplier,
+ boolean detectConflicts,
+ @Nullable String newStatsFileName)
+ throws Lock.AcquisitionException {
+ try {
+ return snapshotCommit.runWithLock(
+ () ->
+ tryCommitOnceWithoutLock(
+ retryResult,
+ changesProvider,
+ identifier,
+ watermark,
+ properties,
+ commitKind,
+ allowRollback,
+ latestSnapshotSupplier.get(),
+ detectConflicts,
+ newStatsFileName));
+ } catch (Lock.AcquisitionException | RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to run commit with table lock.", e);
+ }
+ }
+
+ private CommitResult tryCommitOnceWithoutLock(
@Nullable RetryCommitResult retryResult,
CommitChangesProvider changesProvider,
long identifier,
@@ -1171,6 +1228,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
commitKind);
if (exception.isPresent()) {
if (allowRollback && rollback != null) {
+ snapshotCommit.ensureValid();
if (rollback.tryToRollback(latestSnapshot)) {
return RetryCommitResult.forRollback(exception.get());
}
@@ -1402,8 +1460,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
CommitCallback.Context context =
new CommitCallback.Context(
finalBaseFiles, finalDeltaFiles, indexFiles, newSnapshot, identifier);
- commitCallbacks.forEach(callback -> callback.call(context));
- return new SuccessCommitResult();
+ return new CompletedCommitResult(context);
}
@Nullable
@@ -1494,6 +1551,48 @@ public class FileStoreCommitImpl implements FileStoreCommit {
@Nullable String indexManifest,
@Nullable Long nextRowId,
@Nullable Map<String, String> properties) {
+ return replaceManifestListOnce(
+ latest,
+ totalRecordCount,
+ baseManifestList,
+ deltaManifestList,
+ indexManifest,
+ nextRowId,
+ properties);
+ }
+
+ private boolean replaceManifestListOnce(
+ Snapshot latest,
+ long totalRecordCount,
+ Pair<String, Long> baseManifestList,
+ Pair<String, Long> deltaManifestList,
+ @Nullable String indexManifest,
+ @Nullable Long nextRowId,
+ @Nullable Map<String, String> properties) {
+ return runWithLock(
+ () ->
+ replaceManifestListOnceWithoutLock(
+ latest,
+ totalRecordCount,
+ baseManifestList,
+ deltaManifestList,
+ indexManifest,
+ nextRowId,
+ properties));
+ }
+
+ private boolean replaceManifestListOnceWithoutLock(
+ Snapshot latest,
+ long totalRecordCount,
+ Pair<String, Long> baseManifestList,
+ Pair<String, Long> deltaManifestList,
+ @Nullable String indexManifest,
+ @Nullable Long nextRowId,
+ @Nullable Map<String, String> properties) {
+ Snapshot head = snapshotManager.latestSnapshot();
+ if (head == null || !head.equals(latest)) {
+ return false;
+ }
Snapshot newSnapshot =
new Snapshot(
latest.id() + 1,
@@ -1525,6 +1624,19 @@ public class FileStoreCommitImpl implements FileStoreCommit {
@Override
public boolean rollbackToAsLatest(Snapshot targetSnapshot) {
+ Pair<Boolean, CommitCallback.Context> result = rollbackToAsLatestOnce(targetSnapshot);
+ if (result.getRight() != null) {
+ commitCallbacks.forEach(callback -> callback.call(result.getRight()));
+ }
+ return result.getLeft();
+ }
+
+ private Pair<Boolean, CommitCallback.Context> rollbackToAsLatestOnce(Snapshot targetSnapshot) {
+ return runWithLock(() -> rollbackToAsLatestOnceWithoutLock(targetSnapshot));
+ }
+
+ private Pair<Boolean, CommitCallback.Context> rollbackToAsLatestOnceWithoutLock(
+ Snapshot targetSnapshot) {
Snapshot latest =
checkNotNull(
snapshotManager.latestSnapshot(),
@@ -1633,9 +1745,9 @@ public class FileStoreCommitImpl implements FileStoreCommit {
indexChanges,
newSnapshot,
newSnapshot.commitIdentifier());
- commitCallbacks.forEach(callback -> callback.call(context));
+ return Pair.of(true, context);
}
- return success;
+ return Pair.of(false, null);
}
/**
@@ -1703,6 +1815,10 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
private boolean compactManifestOnce() {
+ return runWithLock(this::compactManifestOnceWithoutLock);
+ }
+
+ private boolean compactManifestOnceWithoutLock() {
Snapshot latestSnapshot = snapshotManager.latestSnapshot();
if (latestSnapshot == null) {
@@ -1799,6 +1915,21 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
}
+ private void notifyCommitSuccess(CommitResult result) {
+ if (result instanceof CompletedCommitResult) {
+ CommitCallback.Context context = ((CompletedCommitResult) result).context;
+ commitCallbacks.forEach(callback -> callback.call(context));
+ }
+ }
+
+ private static class CompletedCommitResult extends SuccessCommitResult {
+ private final CommitCallback.Context context;
+
+ private CompletedCommitResult(CommitCallback.Context context) {
+ this.context = context;
+ }
+ }
+
@Override
public void close() {
IOUtils.closeAllQuietly(commitPreCallbacks);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/Lock.java b/paimon-core/src/main/java/org/apache/paimon/operation/Lock.java
index b65e3c3d0c..aa658c657c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/Lock.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/Lock.java
@@ -19,28 +19,87 @@
package org.apache.paimon.operation;
import org.apache.paimon.annotation.Public;
-import org.apache.paimon.catalog.CatalogLock;
-import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.utils.ExceptionUtils;
+
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
import java.util.concurrent.Callable;
-/**
- * An interface that allows file store to use global lock to some transaction-related things.
- *
- * @since 0.4.0
- */
+/** A table lock covering an operation and its publication. */
@Public
public interface Lock extends AutoCloseable {
- /** Run with lock. */
- <T> T runWithLock(Callable<T> callable) throws Exception;
+ /**
+ * Acquire a scope to release on the acquiring thread. Clean up partial acquisition on failure.
+ */
+ AutoCloseable acquire() throws Exception;
+
+ /**
+ * Run an operation under the lock. Acquisition failures are distinguished from operation
+ * failures; interruption stops acquisition. Releasing a scope must not replace the operation's
+ * result or exception.
+ */
+ default <T> T runWithLock(Callable<T> callable) throws Exception {
+ AutoCloseable scope;
+ try {
+ scope = acquire();
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, InterruptedException.class).isPresent()) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ throw new AcquisitionException(e);
+ }
+
+ Throwable failure = null;
+ try {
+ return callable.call();
+ } catch (Exception | Error e) {
+ failure = e;
+ throw e;
+ } finally {
+ release(scope, failure);
+ }
+ }
+
+ /** Release a scope without losing the outcome of its operation or partial acquisition. */
+ static void release(AutoCloseable scope, @Nullable Throwable failure) {
+ try {
+ scope.close();
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, InterruptedException.class).isPresent()) {
+ Thread.currentThread().interrupt();
+ }
+ if (failure != null && failure != e) {
+ failure.addSuppressed(e);
+ }
+ LoggerFactory.getLogger(Lock.class).warn("Failed to release table lock.", e);
+ }
+ }
+
+ /** Fail before publication when a held lease has expired or renewal has failed. */
+ default void ensureValid() {}
static Lock empty() {
return new EmptyLock();
}
- /** An empty lock. */
+ /** Distinguishes failure to acquire a scope from failure of its operation or publication. */
+ class AcquisitionException extends Exception {
+ private AcquisitionException(Exception cause) {
+ super("Failed to acquire table lock.", cause);
+ }
+ }
+
+ /** An operation with no external lock configured. */
class EmptyLock implements Lock {
+ @Override
+ public AutoCloseable acquire() {
+ return () -> {};
+ }
+
@Override
public <T> T runWithLock(Callable<T> callable) throws Exception {
return callable.call();
@@ -50,33 +109,59 @@ public interface Lock extends AutoCloseable {
public void close() {}
}
- static Lock fromCatalog(CatalogLock lock, Identifier tablePath) {
- if (lock == null) {
- return new EmptyLock();
- }
- return new CatalogLockImpl(lock, tablePath);
+ /** Reuse the current scope for nested publication on the same thread. */
+ static Lock reentrant(Lock lock) {
+ return lock instanceof EmptyLock || lock instanceof ReentrantLock
+ ? lock
+ : new ReentrantLock(lock);
}
- /** A {@link Lock} to wrap {@link CatalogLock}. */
- class CatalogLockImpl implements Lock {
+ /** Keeps the outer operation's scope until preparation and publication both finish. */
+ class ReentrantLock implements Lock {
+ private final Lock delegate;
+ private final ThreadLocal<Boolean> held = ThreadLocal.withInitial(() -> false);
- private final CatalogLock catalogLock;
- private final Identifier tablePath;
+ private ReentrantLock(Lock delegate) {
+ this.delegate = delegate;
+ }
- private CatalogLockImpl(CatalogLock catalogLock, Identifier tablePath) {
- this.catalogLock = catalogLock;
- this.tablePath = tablePath;
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ if (held.get()) {
+ delegate.ensureValid();
+ return () -> {};
+ }
+ AutoCloseable scope = delegate.acquire();
+ try {
+ delegate.ensureValid();
+ held.set(true);
+ return () -> {
+ held.remove();
+ scope.close();
+ };
+ } catch (Exception | Error e) {
+ release(scope, e);
+ throw e;
+ }
}
@Override
public <T> T runWithLock(Callable<T> callable) throws Exception {
- return catalogLock.runWithLock(
- tablePath.getDatabaseName(), tablePath.getObjectName(), callable);
+ if (held.get()) {
+ delegate.ensureValid();
+ return callable.call();
+ }
+ return Lock.super.runWithLock(callable);
+ }
+
+ @Override
+ public void ensureValid() {
+ delegate.ensureValid();
}
@Override
public void close() throws Exception {
- this.catalogLock.close();
+ delegate.close();
}
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
index 66979ba811..d35c50706d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
@@ -37,6 +37,7 @@ import org.apache.paimon.rest.RESTCatalogLoader;
import org.apache.paimon.rest.RESTUtil;
import org.apache.paimon.table.source.TableQueryAuth;
import org.apache.paimon.tag.SnapshotLoaderImpl;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.SnapshotLoader;
import org.apache.paimon.utils.SnapshotManager;
@@ -44,7 +45,6 @@ import org.apache.paimon.utils.SnapshotManager;
import javax.annotation.Nullable;
import java.io.Serializable;
-import java.util.Optional;
import java.util.function.LongConsumer;
import static org.apache.paimon.options.CatalogOptions.METASTORE;
@@ -131,20 +131,53 @@ public class CatalogEnvironment implements Serializable {
return SchemaModification.create(catalog, identifier);
}
- @Nullable
- public SnapshotCommit snapshotCommit(SnapshotManager snapshotManager) {
- SnapshotCommit snapshotCommit;
- if (catalogLoader != null && supportsVersionManagement) {
- snapshotCommit = new CatalogSnapshotCommit(catalogLoader.load(), identifier, uuid);
- } else {
- Lock lock =
- Optional.ofNullable(lockFactory)
- .map(factory -> factory.createLock(lockContext))
- .map(l -> Lock.fromCatalog(l, identifier))
- .orElseGet(Lock::empty);
- snapshotCommit = new RenamingSnapshotCommit(snapshotManager, lock);
+ /** Create a publisher owning the writer's lock and its catalog resources. */
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager snapshotManager, CoreOptions options, String commitUser) {
+ Identifier lockIdentifier =
+ identifier == null
+ ? null
+ : new Identifier(
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ options.branch());
+ if (catalogLoader != null) {
+ Catalog catalog = catalogLoader.load();
+ Lock lock = null;
+ try {
+ lock =
+ catalog.createLock(
+ lockIdentifier, uuid, commitUser, options.toConfiguration());
+ if (supportsVersionManagement) {
+ return new CatalogSnapshotCommit(catalog, identifier, uuid, lock);
+ }
+ return new RenamingSnapshotCommit(snapshotManager, lock) {
+ @Override
+ protected void closeImpl() throws Exception {
+ catalog.close();
+ }
+ };
+ } catch (RuntimeException e) {
+ IOUtils.closeQuietly(lock);
+ IOUtils.closeQuietly(catalog);
+ throw e;
+ }
+ }
+ Lock lock = createFactoryLock(lockIdentifier, commitUser, options.toConfiguration());
+ try {
+ return new RenamingSnapshotCommit(snapshotManager, lock);
+ } catch (RuntimeException e) {
+ IOUtils.closeQuietly(lock);
+ throw e;
}
- return snapshotCommit;
+ }
+
+ private Lock createFactoryLock(
+ Identifier lockIdentifier, @Nullable String commitUser, Options tableOptions) {
+ return lockFactory == null
+ ? Lock.empty()
+ : lockFactory.createLock(
+ lockContext, lockIdentifier, uuid, commitUser, tableOptions);
}
@Nullable
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
index 27ecba030a..62d5a1e62a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactReassignReuseTest.java
@@ -303,11 +303,12 @@ public class CompactReassignReuseTest extends TableTestBase {
CatalogEnvironment environment =
new CatalogEnvironment(null, null, null, null, null, null, false, false) {
@Override
- public SnapshotCommit snapshotCommit(SnapshotManager manager) {
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager manager, CoreOptions options, String commitUser) {
SnapshotCommit delegate = new RenamingSnapshotCommit(manager, Lock.empty());
- return new SnapshotCommit() {
+ return new SnapshotCommit(Lock.empty()) {
@Override
- public boolean commit(
+ protected boolean commitImpl(
String baseUuid,
Snapshot snapshot,
String branch,
@@ -323,7 +324,7 @@ public class CompactReassignReuseTest extends TableTestBase {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
delegate.close();
}
};
@@ -372,11 +373,12 @@ public class CompactReassignReuseTest extends TableTestBase {
CatalogEnvironment environment =
new CatalogEnvironment(null, null, null, null, null, null, false, false) {
@Override
- public SnapshotCommit snapshotCommit(SnapshotManager manager) {
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager manager, CoreOptions options, String commitUser) {
SnapshotCommit delegate = new RenamingSnapshotCommit(manager, Lock.empty());
- return new SnapshotCommit() {
+ return new SnapshotCommit(Lock.empty()) {
@Override
- public boolean commit(
+ protected boolean commitImpl(
String baseUuid,
Snapshot snapshot,
String branch,
@@ -468,7 +470,7 @@ public class CompactReassignReuseTest extends TableTestBase {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
delegate.close();
}
};
@@ -612,11 +614,12 @@ public class CompactReassignReuseTest extends TableTestBase {
CatalogEnvironment environment =
new CatalogEnvironment(null, null, null, null, null, null, false, false) {
@Override
- public SnapshotCommit snapshotCommit(SnapshotManager manager) {
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager manager, CoreOptions options, String commitUser) {
SnapshotCommit delegate = new RenamingSnapshotCommit(manager, Lock.empty());
- return new SnapshotCommit() {
+ return new SnapshotCommit(Lock.empty()) {
@Override
- public boolean commit(
+ protected boolean commitImpl(
String baseUuid,
Snapshot snapshot,
String branch,
@@ -647,7 +650,7 @@ public class CompactReassignReuseTest extends TableTestBase {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
delegate.close();
}
};
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index 5c8bda6fdf..abf14e67b1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -886,11 +886,12 @@ public class DataEvolutionRowIdReassignerTest extends TableTestBase {
CatalogEnvironment environment =
new CatalogEnvironment(null, null, null, null, null, null, false, false) {
@Override
- public SnapshotCommit snapshotCommit(SnapshotManager manager) {
+ public SnapshotCommit snapshotCommit(
+ SnapshotManager manager, CoreOptions options, String commitUser) {
SnapshotCommit delegate = new RenamingSnapshotCommit(manager, Lock.empty());
- return new SnapshotCommit() {
+ return new SnapshotCommit(Lock.empty()) {
@Override
- public boolean commit(
+ protected boolean commitImpl(
@Nullable String baseUuid,
Snapshot snapshot,
String branch,
@@ -906,7 +907,7 @@ public class DataEvolutionRowIdReassignerTest extends TableTestBase {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
delegate.close();
}
};
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogLockConfigurationTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogLockConfigurationTest.java
new file mode 100644
index 0000000000..bd42a7f1d0
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogLockConfigurationTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.jdbc.JdbcCatalogOptions;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Optional;
+import java.util.concurrent.Callable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+/** Tests that catalogs without a locking policy ignore backend-specific lock options. */
+class CatalogLockConfigurationTest {
+ @Test
+ void abstractCatalogDoesNotInterpretBackendLockOptions() throws Exception {
+ Options options = new Options();
+ options.set(JdbcCatalogOptions.LOCK_ENABLED, true);
+ options.set(JdbcCatalogOptions.LOCK_TYPE, "not-installed");
+ AbstractCatalog catalog =
+ mock(
+ AbstractCatalog.class,
+ withSettings()
+ .useConstructor(mock(FileIO.class), CatalogContext.create(options))
+ .defaultAnswer(CALLS_REAL_METHODS));
+ try (Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"), null, "writer", new Options())) {
+ assertThat(lock.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void filesystemDoesNotInterpretBackendLockOptions(boolean objectStore) throws Exception {
+ Options options = new Options();
+ options.set(JdbcCatalogOptions.LOCK_ENABLED, true);
+ options.set(JdbcCatalogOptions.LOCK_TYPE, "not-installed");
+ FileIO fileIO = mock(FileIO.class);
+ when(fileIO.isObjectStore()).thenReturn(objectStore);
+ FileSystemCatalog catalog =
+ new FileSystemCatalog(
+ fileIO, new Path("file:/warehouse"), CatalogContext.create(options));
+ Identifier identifier = Identifier.create("db", "table");
+ try (Lock lock = catalog.createLock(identifier, "table-id", "writer", new Options())) {
+ assertThat(lock.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ assertThat(catalog.runWithLock(identifier, () -> "created")).isEqualTo("created");
+ }
+
+ @Test
+ void factoryBindsRuntimeIdentityWhenCatalogCreatesLock() throws Exception {
+ Options catalogOptions = new Options();
+ catalogOptions.set("catalog-option", "value");
+ Options tableOptions = new Options();
+ tableOptions.set("runtime-option", "dynamic-value");
+ Identifier identifier = new Identifier("db", "table", "dev");
+ CatalogLockFactory factory = mock(CatalogLockFactory.class);
+ Lock backend = mock(Lock.class);
+ when(backend.runWithLock(any()))
+ .thenAnswer(invocation -> ((Callable<?>) invocation.getArgument(0)).call());
+ when(factory.createLock(
+ any(), eq(identifier), eq("table-id"), eq("writer"), eq(tableOptions)))
+ .thenAnswer(
+ invocation -> {
+ CatalogLockContext context = invocation.getArgument(0);
+ assertThat(context.options()).isEqualTo(catalogOptions);
+ return backend;
+ });
+ AbstractCatalog catalog =
+ mock(
+ AbstractCatalog.class,
+ withSettings()
+ .useConstructor(
+ mock(FileIO.class), CatalogContext.create(catalogOptions))
+ .defaultAnswer(CALLS_REAL_METHODS));
+ when(catalog.lockFactory()).thenReturn(Optional.of(factory));
+ try (Lock lock = catalog.createLock(identifier, "table-id", "writer", tableOptions)) {
+ assertThat(lock.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ verify(factory)
+ .createLock(any(), eq(identifier), eq("table-id"), eq("writer"), eq(tableOptions));
+ verify(backend).ensureValid();
+ verify(backend).close();
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java
index 1b5cfeb1cc..81f69573fd 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.catalog;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.utils.SnapshotManagerTest;
import org.junit.jupiter.api.Test;
@@ -47,7 +48,8 @@ public class CatalogSnapshotCommitTest {
Collections.emptyList()))
.thenReturn(true);
- CatalogSnapshotCommit commit = new CatalogSnapshotCommit(catalog, identifier, "table-uuid");
+ CatalogSnapshotCommit commit =
+ new CatalogSnapshotCommit(catalog, identifier, "table-uuid", Lock.empty());
assertThat(commit.commit("base-snapshot-uuid", snapshot, "branch", Collections.emptyList()))
.isTrue();
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/SnapshotCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/SnapshotCommitTest.java
new file mode 100644
index 0000000000..72c0145804
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/SnapshotCommitTest.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.partition.PartitionStatistics;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+/** Tests the inseparable lock scope and publication contract. */
+class SnapshotCommitTest {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void directAndPreparedPublicationUseOneScope(boolean prepare) throws Exception {
+ RecordingLock lock = new RecordingLock();
+ Snapshot snapshot = mock(Snapshot.class);
+ try (RecordingSnapshotCommit publisher = new RecordingSnapshotCommit(lock)) {
+ if (prepare) {
+ assertThat(
+ publisher.runWithLock(
+ () -> {
+ lock.events.add("read-head");
+ lock.events.add("prepare");
+ return publisher.commit(
+ null,
+ snapshot,
+ "main",
+ Collections.emptyList());
+ }))
+ .isTrue();
+ } else {
+ assertThat(publisher.commit(null, snapshot, "main", Collections.emptyList()))
+ .isTrue();
+ }
+ assertThat(publisher.publications).isEqualTo(1);
+ }
+ if (prepare) {
+ assertThat(lock.events)
+ .containsExactly(
+ "acquire",
+ "validate",
+ "read-head",
+ "prepare",
+ "validate",
+ "publish",
+ "release",
+ "close-lock",
+ "close-publisher");
+ } else {
+ assertThat(lock.events)
+ .containsExactly(
+ "acquire",
+ "validate",
+ "publish",
+ "release",
+ "close-lock",
+ "close-publisher");
+ }
+ }
+
+ @Test
+ void lostLeasePreventsPublicationAndNextAttemptAcquiresNewScope() throws Exception {
+ RecordingLock lock = new RecordingLock();
+ Snapshot snapshot = mock(Snapshot.class);
+ try (RecordingSnapshotCommit publisher = new RecordingSnapshotCommit(lock)) {
+ assertThatThrownBy(
+ () ->
+ publisher.runWithLock(
+ () -> {
+ lock.expired = true;
+ return publisher.commit(
+ null,
+ snapshot,
+ "main",
+ Collections.emptyList());
+ }))
+ .hasMessage("Lost lease");
+ assertThat(publisher.publications).isZero();
+ assertThat(lock.held).isFalse();
+ lock.expired = false;
+ assertThat(publisher.commit(null, snapshot, "main", Collections.emptyList())).isTrue();
+ assertThat(publisher.publications).isEqualTo(1);
+ }
+ assertThat(lock.events)
+ .containsExactly(
+ "acquire",
+ "validate",
+ "validate",
+ "release",
+ "acquire",
+ "validate",
+ "publish",
+ "release",
+ "close-lock",
+ "close-publisher");
+ }
+
+ @Test
+ void lockCloseFailureStillClosesPublisher() {
+ RecordingLock lock = new RecordingLock();
+ lock.failClose = true;
+ RecordingSnapshotCommit publisher = new RecordingSnapshotCommit(lock);
+ assertThatThrownBy(publisher::close)
+ .isInstanceOf(IOException.class)
+ .hasMessage("Lock close failed");
+ assertThat(lock.events).containsExactly("close-lock", "close-publisher");
+ }
+
+ private static class RecordingSnapshotCommit extends SnapshotCommit {
+ private final RecordingLock lock;
+ private int publications;
+
+ private RecordingSnapshotCommit(RecordingLock lock) {
+ super(lock);
+ this.lock = lock;
+ }
+
+ @Override
+ protected boolean commitImpl(
+ String baseSnapshotUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics) {
+ assertThat(lock.held).isTrue();
+ lock.events.add("publish");
+ publications++;
+ return true;
+ }
+
+ @Override
+ protected void closeImpl() {
+ lock.events.add("close-publisher");
+ }
+ }
+
+ private static class RecordingLock implements Lock {
+ private final List<String> events = new ArrayList<>();
+ private boolean held;
+ private boolean expired;
+ private boolean failClose;
+
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ assertThat(held).isFalse();
+ events.add("acquire");
+ held = true;
+ return () -> {
+ held = false;
+ events.add("release");
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ assertThat(held).isTrue();
+ events.add("validate");
+ if (expired) {
+ throw new IllegalStateException("Lost lease");
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ events.add("close-lock");
+ if (failClose) {
+ throw new IOException("Lock close failed");
+ }
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/TestCatalogLockFactory.java b/paimon-core/src/test/java/org/apache/paimon/catalog/TestCatalogLockFactory.java
new file mode 100644
index 0000000000..0467c45a64
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/TestCatalogLockFactory.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+/** Records scopes acquired through the catalog lock factory SPI. */
+public class TestCatalogLockFactory implements CatalogLockFactory {
+ public static final String IDENTIFIER = "test-catalog-lock";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Lock createLock(
+ CatalogLockContext context,
+ Identifier identifier,
+ String tableUuid,
+ String commitUser,
+ Options tableOptions) {
+ Options options = context.options();
+ return new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ options.set("test.lock.database", identifier.getDatabaseName());
+ options.set("test.lock.table", identifier.getObjectName());
+ options.set("test.lock.held", "true");
+ return () -> {
+ options.set("test.lock.held", "false");
+ };
+ }
+
+ @Override
+ public void close() {
+ options.set("test.lock.closed", "true");
+ }
+ };
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogLockConfigurationTest.java b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogLockConfigurationTest.java
new file mode 100644
index 0000000000..68a2c7496b
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogLockConfigurationTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.jdbc;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.TestCatalogLockFactory;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.nio.file.Path;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests JDBC locking through the catalog's own options and policy. */
+class JdbcCatalogLockConfigurationTest {
+ @TempDir Path directory;
+
+ @ParameterizedTest
+ @CsvSource({"false,,false", "true,,true", "false,true,true", "true,false,false"})
+ void jdbcPreservesStorageDefaultAndExplicitOverride(
+ boolean objectStore, Boolean enabled, boolean expected) throws Exception {
+ Options options = options();
+ options.set(JdbcCatalogOptions.LOCK_ACQUIRE_TIMEOUT, Duration.ofSeconds(3));
+ options.set(JdbcCatalogOptions.LOCK_CHECK_MAX_SLEEP, Duration.ofMillis(17));
+ if (enabled != null) {
+ options.set(JdbcCatalogOptions.LOCK_ENABLED, enabled);
+ }
+ try (JdbcCatalog catalog = catalog(options, objectStore);
+ Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"), null, "writer", new Options())) {
+ lock.runWithLock(
+ () -> {
+ assertThat(lockTableExists(catalog)).isEqualTo(expected);
+ if (expected) {
+ catalog.getConnections()
+ .execute(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT lock_owner, expire_time_seconds FROM paimon_distributed_locks");
+ ResultSet result =
+ statement.executeQuery()) {
+ assertThat(result.next()).isTrue();
+ assertThat(result.getString(1)).isNotEmpty();
+ assertThat(result.getLong(2)).isEqualTo(5);
+ }
+ });
+ }
+ return null;
+ });
+ assertThat(JdbcCatalogLock.checkMaxSleep(options.toMap())).isEqualTo(17);
+ }
+ }
+
+ @Test
+ void configuredFactoryOverridesJdbcDefault() throws Exception {
+ Options options = options();
+ options.set(JdbcCatalogOptions.LOCK_ENABLED, true);
+ options.set(JdbcCatalogOptions.LOCK_TYPE, TestCatalogLockFactory.IDENTIFIER);
+ try (JdbcCatalog catalog = catalog(options, false);
+ Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"), null, "writer", new Options())) {
+ assertThat(lock.runWithLock(() -> options.get("test.lock.held"))).isEqualTo("true");
+ assertThat(
+ catalog.runWithLock(
+ Identifier.create("db", "table"),
+ () -> options.get("test.lock.held")))
+ .isEqualTo("true");
+ }
+ assertThat(options.get("test.lock.held")).isEqualTo("false");
+ assertThat(options.get("test.lock.closed")).isEqualTo("true");
+ }
+
+ @Test
+ void disabledJdbcLockDoesNotResolveFactoryOrCreateLeaseTable() throws Exception {
+ Options options = options();
+ options.set(JdbcCatalogOptions.LOCK_ENABLED, false);
+ options.set(JdbcCatalogOptions.LOCK_TYPE, "not-installed");
+ try (JdbcCatalog catalog = catalog(options, true);
+ Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"), null, "writer", new Options())) {
+ assertThat(lock.runWithLock(() -> "published")).isEqualTo("published");
+ assertThat(lockTableExists(catalog)).isFalse();
+ }
+ }
+
+ private Options options() {
+ Options options = new Options();
+ options.set(CatalogOptions.URI, "jdbc:sqlite:" + directory.resolve("catalog.db"));
+ return options;
+ }
+
+ private JdbcCatalog catalog(Options options, boolean objectStore) {
+ FileIO fileIO = mock(FileIO.class);
+ when(fileIO.isObjectStore()).thenReturn(objectStore);
+ return new JdbcCatalog(
+ fileIO, "configured", CatalogContext.create(options), directory.toString());
+ }
+
+ private boolean lockTableExists(JdbcCatalog catalog) throws Exception {
+ return catalog.getConnections()
+ .run(
+ connection -> {
+ try (ResultSet tables =
+ connection
+ .getMetaData()
+ .getTables(
+ null, null, "paimon_distributed_locks", null)) {
+ return tables.next();
+ }
+ });
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogLockTest.java b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogLockTest.java
new file mode 100644
index 0000000000..70ae7d5083
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogLockTest.java
@@ -0,0 +1,561 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.jdbc;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.client.ClientPool;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests renewable JDBC ownership using a real database and a single-connection pool. */
+class JdbcCatalogLockTest {
+ @TempDir Path directory;
+ private final ScheduledExecutorService renewer = mock(ScheduledExecutorService.class);
+ private final ScheduledFuture<?> future = mock(ScheduledFuture.class);
+ private final AtomicLong clock = new AtomicLong();
+ private final AbstractDistributedLockDialect dialect = new SqlLiteDistributedLockDialect();
+ private final Set<Thread> renewalThreads = ConcurrentHashMap.newKeySet();
+
+ @Test
+ void failedLeaseInitializationReleasesOwnershipBeforeRetry() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = lock(pool);
+ RejectedExecutionException failure =
+ new RejectedExecutionException("Renewer unavailable");
+ when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any()))
+ .thenThrow(failure);
+ AtomicInteger operations = new AtomicInteger();
+ assertThatThrownBy(() -> lock.runWithLock(operations::incrementAndGet))
+ .isInstanceOf(Lock.AcquisitionException.class)
+ .hasCause(failure);
+ assertThat(operations.get()).isZero();
+ assertThat(owner(pool)).isNull();
+ doReturn(future)
+ .when(renewer)
+ .scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any());
+ assertThat(lock.runWithLock(operations::incrementAndGet)).isEqualTo(1);
+ assertThat(owner(pool)).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void releaseFailurePreservesOperationOutcome(boolean failOperation) throws Exception {
+ AtomicBoolean failRelease = new AtomicBoolean();
+ SQLException releaseFailure = new SQLException("Release unavailable");
+ RuntimeException operationFailure = new RuntimeException("Preparation failed");
+ try (JdbcClientPool pool =
+ new JdbcClientPool(
+ 1, "jdbc:sqlite:" + directory.resolve("locks.db"), Collections.emptyMap()) {
+ @Override
+ public <R> R run(ClientPool.Action<R, Connection, SQLException> action)
+ throws SQLException, InterruptedException {
+ if (failRelease.get()) {
+ throw releaseFailure;
+ }
+ return super.run(action);
+ }
+ }) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = lock(pool);
+ java.util.concurrent.Callable<String> operation =
+ () -> {
+ failRelease.set(true);
+ if (failOperation) {
+ throw operationFailure;
+ }
+ return "published";
+ };
+ if (failOperation) {
+ assertThatThrownBy(() -> lock.runWithLock(operation))
+ .isSameAs(operationFailure)
+ .hasSuppressedException(releaseFailure);
+ } else {
+ assertThat(lock.runWithLock(operation)).isEqualTo("published");
+ }
+ failRelease.set(false);
+ assertThatThrownBy(lock::ensureValid).hasMessageContaining("No JDBC lock");
+ verify(future).cancel(true);
+ assertThat(owner(pool)).isNotEmpty();
+ // A failed release retains the row until expiry; remove it for this test's next scope.
+ dialect.releaseLock(pool, "catalog.db.table");
+ }
+ }
+
+ @Test
+ void callbackCanUseThePoolAndRenewalExtendsLifetime() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = lock(pool);
+ Runnable callback =
+ lock.runWithLock(
+ () -> {
+ // The holder must not keep a pooled connection while the callback
+ // borrows one.
+ assertThat(owner(pool)).isNotEmpty();
+ clock.set(TimeUnit.MILLISECONDS.toNanos(2000));
+ Runnable renewal = renewal();
+ renewal.run();
+ clock.set(TimeUnit.MILLISECONDS.toNanos(4000));
+ lock.ensureValid();
+ return renewal;
+ });
+ assertThat(owner(pool)).isNull();
+ callback.run();
+ assertThat(owner(pool)).isNull();
+ verify(future).cancel(true);
+ }
+ }
+
+ @Test
+ void formerHolderCannotRenewOrReleaseSuccessor() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = lock(pool);
+ lock.runWithLock(
+ () -> {
+ dialect.releaseLock(pool, "catalog.db.table");
+ assertThat(
+ dialect.acquireOwned(
+ pool, "catalog.db.table", "successor", 3000))
+ .isTrue();
+ renewal().run();
+ assertThatThrownBy(lock::ensureValid)
+ .hasMessageContaining("no longer usable");
+ assertThat(owner(pool)).isEqualTo("successor");
+ return null;
+ });
+ assertThat(owner(pool)).isEqualTo("successor");
+ }
+ }
+
+ @Test
+ void localExpiryInvalidatesHolderWithoutScheduledRenewal() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = lock(pool);
+ lock.runWithLock(
+ () -> {
+ clock.set(TimeUnit.MILLISECONDS.toNanos(3000));
+ assertThatThrownBy(lock::ensureValid)
+ .hasCauseInstanceOf(IllegalStateException.class);
+ renewal().run();
+ assertThatThrownBy(lock::ensureValid)
+ .hasMessageContaining("no longer usable");
+ return null;
+ });
+ assertThat(owner(pool)).isNull();
+ }
+ }
+
+ @Test
+ void legacyLockTableIsUpgradedWithoutRemovingRows() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ createLegacyLockTable(pool);
+ assertThat(dialect.lockAcquire(pool, "old-lock", 3000)).isTrue();
+ dialect.createTable(pool, new Options());
+ dialect.createTable(pool, new Options());
+ assertThat(dialect.lockAcquire(pool, "old-lock", 3000)).isFalse();
+ assertThat(dialect.acquireOwned(pool, "new-lock", "new-owner", 3000)).isTrue();
+ assertThat(dialect.releaseOwned(pool, "new-lock", "other-owner")).isFalse();
+ assertThat(dialect.releaseOwned(pool, "new-lock", "new-owner")).isTrue();
+ }
+ }
+
+ @Test
+ void legacyLockTableUpgradeIgnoresSimilarlyNamedTable() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ createLegacyLockTable(pool);
+ pool.execute(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "CREATE TABLE paimonXdistributedXlocks (lock_owner VARCHAR(36))")) {
+ statement.execute();
+ }
+ });
+ dialect.createTable(pool, new Options());
+ assertThat(dialect.acquireOwned(pool, "lock", "owner", 3000)).isTrue();
+ }
+ }
+
+ @Test
+ void legacyLockTableUpgradeIgnoresSimilarlyNamedColumn() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ createLegacyLockTable(pool);
+ pool.execute(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "ALTER TABLE paimon_distributed_locks ADD COLUMN lockXowner VARCHAR(36)")) {
+ statement.execute();
+ }
+ });
+ dialect.createTable(pool, new Options());
+ assertThat(dialect.acquireOwned(pool, "lock", "owner", 3000)).isTrue();
+ }
+ }
+
+ @Test
+ void blockedRenewalsDoNotPreventOtherCatalogsFromRenewing() throws Exception {
+ CountDownLatch blocked = new CountDownLatch(2);
+ CountDownLatch renewed = new CountDownLatch(4);
+ CountDownLatch release = new CountDownLatch(1);
+ ExecutorService writers = Executors.newFixedThreadPool(2);
+ List<Future<?>> scopes = new ArrayList<>();
+ try (JdbcClientPool first = observedPool("first.db", blocked, new CountDownLatch(0));
+ JdbcClientPool second = observedPool("second.db", blocked, new CountDownLatch(0));
+ JdbcClientPool healthy =
+ observedPool("healthy.db", new CountDownLatch(0), renewed)) {
+ try {
+ for (JdbcClientPool pool : new JdbcClientPool[] {first, second}) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = defaultLock(pool);
+ scopes.add(writers.submit(() -> holdScope(lock, pool, release)));
+ }
+ assertThat(blocked.await(10, TimeUnit.SECONDS)).isTrue();
+ dialect.createTable(healthy, new Options());
+ JdbcCatalogLock lock = defaultLock(healthy);
+ assertThat(
+ lock.runWithLock(
+ () -> {
+ assertThat(renewed.await(10, TimeUnit.SECONDS))
+ .isTrue();
+ lock.ensureValid();
+ return "published";
+ }))
+ .isEqualTo("published");
+ assertThat(owner(healthy)).isNull();
+ } finally {
+ release.countDown();
+ for (Future<?> scope : scopes) {
+ scope.get(10, TimeUnit.SECONDS);
+ }
+ }
+ } finally {
+ release.countDown();
+ writers.shutdownNow();
+ }
+ for (Thread thread : renewalThreads) {
+ thread.join(10000);
+ assertThat(thread.isAlive()).isFalse();
+ }
+ }
+
+ @Test
+ void delayedHeartbeatCannotReviveAnExpiredDatabaseRow() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ assertThat(dialect.acquireOwned(pool, "lock", "owner", 3000)).isTrue();
+ assertThat(dialect.acquireOwned(pool, "lock", "contender", 3000)).isFalse();
+ pool.execute(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "UPDATE paimon_distributed_locks SET acquired_at = '2000-01-01 00:00:00'")) {
+ statement.executeUpdate();
+ }
+ });
+ assertThat(dialect.renewOwned(pool, "lock", "owner")).isFalse();
+ assertThat(dialect.acquireOwned(pool, "lock", "contender", 3000)).isTrue();
+ }
+ }
+
+ @Test
+ void zeroAffectedRowsDoesNotInvalidateUnchangedLease() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ JdbcCatalogLock lock = lock(zeroUpdateCountPool(pool));
+ assertThat(
+ lock.runWithLock(
+ () -> {
+ lock.ensureValid();
+ assertThat(owner(pool)).isNotEmpty();
+ return "published";
+ }))
+ .isEqualTo("published");
+ assertThat(owner(pool)).isNull();
+ }
+ }
+
+ @Test
+ void zeroAffectedRowsStillRejectsMissingExpiredOrReplacedOwner() throws Exception {
+ try (JdbcClientPool pool = pool()) {
+ dialect.createTable(pool, new Options());
+ JdbcClientPool affectedRowsPool = zeroUpdateCountPool(pool);
+ assertThat(dialect.acquireOwned(pool, "lock", "owner", 3000)).isTrue();
+ assertThat(dialect.renewOwned(affectedRowsPool, "missing", "owner")).isFalse();
+ assertThat(dialect.renewOwned(affectedRowsPool, "lock", "other-owner")).isFalse();
+ pool.execute(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "UPDATE paimon_distributed_locks SET acquired_at = '2000-01-01 00:00:00'")) {
+ statement.executeUpdate();
+ }
+ });
+ assertThat(dialect.renewOwned(affectedRowsPool, "lock", "owner")).isFalse();
+ assertThat(dialect.acquireOwned(pool, "lock", "successor", 3000)).isTrue();
+ assertThat(dialect.renewOwned(affectedRowsPool, "lock", "owner")).isFalse();
+ assertThat(dialect.renewOwned(affectedRowsPool, "lock", "successor")).isTrue();
+ }
+ }
+
+ @Test
+ void acquisitionPropagatesSqlFailuresOtherThanDuplicateKeys() throws Exception {
+ JdbcClientPool pool = mock(JdbcClientPool.class);
+ Connection connection = mock(Connection.class);
+ PreparedStatement statement = mock(PreparedStatement.class);
+ SQLException failure = new SQLException("Permission denied", "42501");
+ when(pool.run(any()))
+ .thenAnswer(
+ invocation ->
+ ((ClientPool.Action<?, Connection, SQLException>)
+ invocation.getArgument(0))
+ .run(connection));
+ when(connection.prepareStatement(dialect.getTryReleaseTimedOutLock()))
+ .thenReturn(statement);
+ when(connection.prepareStatement(
+ "INSERT INTO paimon_distributed_locks (lock_id, expire_time_seconds, lock_owner) VALUES (?, ?, ?)"))
+ .thenThrow(failure);
+ assertThatThrownBy(() -> dialect.acquireOwned(pool, "lock", "owner", 3000))
+ .isSameAs(failure);
+ }
+
+ @Test
+ void olderHeartbeatResponseCannotShortenANewerDeadline() throws Exception {
+ JdbcClientPool pool = mock(JdbcClientPool.class);
+ Connection connection = mock(Connection.class);
+ PreparedStatement statement = mock(PreparedStatement.class);
+ PreparedStatement heartbeat = mock(PreparedStatement.class);
+ when(pool.getProtocol()).thenReturn("sqlite");
+ when(pool.run(any()))
+ .thenAnswer(
+ invocation ->
+ ((ClientPool.Action<?, Connection, SQLException>)
+ invocation.getArgument(0))
+ .run(connection));
+ when(connection.prepareStatement(anyString()))
+ .thenAnswer(
+ invocation ->
+ ((String) invocation.getArgument(0)).startsWith("UPDATE")
+ ? heartbeat
+ : statement);
+ when(statement.executeUpdate()).thenReturn(1);
+ JdbcCatalogLock lock = lock(pool);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CountDownLatch oldStarted = new CountDownLatch(1);
+ CountDownLatch oldResponse = new CountDownLatch(1);
+ AtomicInteger requests = new AtomicInteger();
+ try {
+ lock.runWithLock(
+ () -> {
+ when(heartbeat.executeUpdate())
+ .thenAnswer(
+ invocation -> {
+ if (requests.incrementAndGet() == 1) {
+ oldStarted.countDown();
+ assertThat(oldResponse.await(10, TimeUnit.SECONDS))
+ .isTrue();
+ }
+ return 1;
+ });
+ clock.set(TimeUnit.MILLISECONDS.toNanos(1000));
+ Future<?> old = executor.submit(renewal());
+ assertThat(oldStarted.await(10, TimeUnit.SECONDS)).isTrue();
+ clock.set(TimeUnit.MILLISECONDS.toNanos(2000));
+ lock.ensureValid();
+ oldResponse.countDown();
+ old.get(10, TimeUnit.SECONDS);
+ clock.set(TimeUnit.MILLISECONDS.toNanos(4000));
+ lock.ensureValid();
+ return null;
+ });
+ } finally {
+ oldResponse.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ private JdbcClientPool pool() {
+ return new JdbcClientPool(
+ 1, "jdbc:sqlite:" + directory.resolve("locks.db"), Collections.emptyMap());
+ }
+
+ private void createLegacyLockTable(JdbcClientPool pool) throws Exception {
+ pool.execute(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "CREATE TABLE paimon_distributed_locks (lock_id VARCHAR(255) PRIMARY KEY, "
+ + "acquired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, "
+ + "expire_time_seconds BIGINT DEFAULT 0 NOT NULL)")) {
+ statement.execute();
+ }
+ });
+ }
+
+ private JdbcClientPool observedPool(
+ String name, CountDownLatch starting, CountDownLatch completed) {
+ return new JdbcClientPool(
+ 1, "jdbc:sqlite:" + directory.resolve(name), Collections.emptyMap()) {
+ @Override
+ public <R> R run(ClientPool.Action<R, Connection, SQLException> action)
+ throws SQLException, InterruptedException {
+ boolean renewal = Thread.currentThread().getName().startsWith("jdbc-lock-renew");
+ if (renewal) {
+ renewalThreads.add(Thread.currentThread());
+ starting.countDown();
+ }
+ R result = super.run(action);
+ if (renewal) {
+ completed.countDown();
+ }
+ return result;
+ }
+ };
+ }
+
+ private JdbcCatalogLock defaultLock(JdbcClientPool pool) {
+ return new JdbcCatalogLock(pool, "catalog", Identifier.create("db", "table"), 10, 3000);
+ }
+
+ private Void holdConnection(CountDownLatch release) throws SQLException {
+ // Occupy this catalog's only connection while renewal tries to borrow it.
+ try {
+ assertThat(release.await(30, TimeUnit.SECONDS)).isTrue();
+ return null;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SQLException(e);
+ }
+ }
+
+ private Void holdScope(JdbcCatalogLock lock, JdbcClientPool pool, CountDownLatch release)
+ throws Exception {
+ return lock.runWithLock(() -> pool.run(connection -> holdConnection(release)));
+ }
+
+ private JdbcClientPool zeroUpdateCountPool(JdbcClientPool pool) throws Exception {
+ JdbcClientPool affectedRowsPool = mock(JdbcClientPool.class);
+ when(affectedRowsPool.getProtocol()).thenReturn(pool.getProtocol());
+ when(affectedRowsPool.run(any()))
+ .thenAnswer(
+ invocation ->
+ pool.run(
+ connection ->
+ ((ClientPool.Action<?, Connection, SQLException>)
+ invocation.getArgument(0))
+ .run(
+ zeroUpdateCountConnection(
+ connection))));
+ return affectedRowsPool;
+ }
+
+ private Connection zeroUpdateCountConnection(Connection connection) throws SQLException {
+ Connection reported = mock(Connection.class, delegatesTo(connection));
+ doAnswer(
+ preparation -> {
+ String sql = preparation.getArgument(0);
+ PreparedStatement statement = connection.prepareStatement(sql);
+ if (!sql.startsWith("UPDATE ")) {
+ return statement;
+ }
+ PreparedStatement changedRows =
+ mock(PreparedStatement.class, delegatesTo(statement));
+ doAnswer(
+ update -> {
+ statement.executeUpdate();
+ return 0;
+ })
+ .when(changedRows)
+ .executeUpdate();
+ return changedRows;
+ })
+ .when(reported)
+ .prepareStatement(anyString());
+ return reported;
+ }
+
+ private JdbcCatalogLock lock(JdbcClientPool pool) {
+ when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any()))
+ .thenAnswer(invocation -> future);
+ return new JdbcCatalogLock(
+ pool, "catalog", Identifier.create("db", "table"), 10, 3000, renewer, clock::get);
+ }
+
+ private String owner(JdbcClientPool pool) throws Exception {
+ return pool.run(
+ connection -> {
+ try (PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT lock_owner FROM paimon_distributed_locks WHERE lock_id = 'catalog.db.table'");
+ ResultSet result = statement.executeQuery()) {
+ return result.next() ? result.getString(1) : null;
+ }
+ });
+ }
+
+ private Runnable renewal() {
+ ArgumentCaptor<Runnable> callback = ArgumentCaptor.forClass(Runnable.class);
+ verify(renewer).scheduleWithFixedDelay(callback.capture(), anyLong(), anyLong(), any());
+ return callback.getValue();
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
index 55808e1e42..2cb4756e9f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
@@ -92,8 +92,8 @@ public class JdbcCatalogTest extends CatalogTestBase {
properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user");
properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
properties.put(CatalogOptions.WAREHOUSE.key(), warehouse);
- properties.put(CatalogOptions.LOCK_ENABLED.key(), "true");
- properties.put(CatalogOptions.LOCK_TYPE.key(), "jdbc");
+ properties.put(JdbcCatalogOptions.LOCK_ENABLED.key(), "true");
+ properties.put(JdbcCatalogOptions.LOCK_TYPE.key(), "jdbc");
properties.putAll(props);
JdbcCatalog catalog =
new JdbcCatalog(
@@ -446,13 +446,13 @@ public class JdbcCatalogTest extends CatalogTestBase {
private JdbcCatalog initCatalogWithoutLock() {
Map<String, String> props = Maps.newHashMap();
- props.put(CatalogOptions.LOCK_ENABLED.key(), "false");
+ props.put(JdbcCatalogOptions.LOCK_ENABLED.key(), "false");
return initCatalog(props);
}
private JdbcCatalog initCatalogWithoutLock(String uri) {
Map<String, String> props = Maps.newHashMap();
- props.put(CatalogOptions.LOCK_ENABLED.key(), "false");
+ props.put(JdbcCatalogOptions.LOCK_ENABLED.key(), "false");
return initCatalog(props, uri);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/jdbc/PostgresqlCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/jdbc/PostgresqlCatalogTest.java
index d6bdc2438f..e09e19c957 100644
--- a/paimon-core/src/test/java/org/apache/paimon/jdbc/PostgresqlCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/PostgresqlCatalogTest.java
@@ -114,8 +114,8 @@ public class PostgresqlCatalogTest {
properties.put(JdbcCatalog.PROPERTY_PREFIX + "user", USER);
properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", PASSWORD);
properties.put(CatalogOptions.WAREHOUSE.key(), warehouse);
- properties.put(CatalogOptions.LOCK_ENABLED.key(), "true");
- properties.put(CatalogOptions.LOCK_TYPE.key(), "jdbc");
+ properties.put(JdbcCatalogOptions.LOCK_ENABLED.key(), "true");
+ properties.put(JdbcCatalogOptions.LOCK_TYPE.key(), "jdbc");
properties.putAll(props);
JdbcCatalog catalog =
new JdbcCatalog(
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
index 600d5d12d8..4443813d77 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileDeletionTest.java
@@ -918,7 +918,8 @@ public class FileDeletionTest {
}
// clean given bucket and generate a new snapshot
- private void cleanBucket(TestFileStore store, BinaryRow partition, int bucket) {
+ private void cleanBucket(TestFileStore store, BinaryRow partition, int bucket)
+ throws Exception {
// manually make delete ManifestEntry
List<ManifestEntry> bucketEntries =
store.newScan().withPartitionBucket(partition, bucket).plan().files();
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index d030b3aee8..f4501aacce 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -63,8 +63,10 @@ import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.table.ExpireSnapshotsImpl;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.CommitCallback;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.CommitPreCallback;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.source.IncrementalSplit;
import org.apache.paimon.table.source.Split;
@@ -73,10 +75,13 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
+import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TraceableFileIO;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -89,6 +94,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -102,6 +108,8 @@ import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -116,6 +124,15 @@ import static org.apache.paimon.utils.HintFileUtils.LATEST;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
/** Tests for {@link FileStoreCommitImpl}. */
public class FileStoreCommitTest {
@@ -1732,6 +1749,675 @@ public class FileStoreCommitTest {
assertThat(id).isEqualTo(2);
}
+ @ParameterizedTest
+ @CsvSource({"1,checked", "1,runtime", "1,lease", "3,checked", "3,runtime", "3,lease"})
+ public void testLockAcquisitionFailuresAreRetried(int failures, String failureKind)
+ throws Exception {
+ TestFileStore store = createStore(false);
+ for (int i = 0; i < 2; i++) {
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211109", 6 + i, 1L, null, "old")),
+ gen::getPartition,
+ value -> 0);
+ }
+ store.snapshotManager().deleteSnapshot(1);
+ store.snapshotManager().commitEarliestHint(2);
+ ManifestCommittable prepared = prepareLockRetryCommittable(store);
+ AtomicInteger acquisitions = new AtomicInteger();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ if (acquisitions.incrementAndGet() <= failures) {
+ if ("checked".equals(failureKind)) {
+ throw new IOException("Transient acquisition failure");
+ }
+ if ("runtime".equals(failureKind)) {
+ throw new IllegalStateException("Transient acquisition failure");
+ }
+ return () -> {};
+ }
+ // Another writer advances the head while this attempt waits for its lock.
+ store.commitData(
+ Collections.singletonList(
+ gen.nextInsert("20211110", 9, 2L, null, "b")),
+ gen::getPartition,
+ value -> 0);
+ return () -> {};
+ }
+
+ @Override
+ public void ensureValid() {
+ if ("lease".equals(failureKind) && acquisitions.get() <= failures) {
+ throw new IllegalStateException("Transient acquisition failure");
+ }
+ }
+
+ @Override
+ public void close() {}
+ };
+ SnapshotCommit publisher = new RenamingSnapshotCommit(store.snapshotManager(), lock);
+ Options options = Options.fromMap(store.options().toMap());
+ options.set(CoreOptions.COMMIT_MAX_RETRIES, 2);
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store, "writer", publisher, new CoreOptions(options), false)) {
+ if (failures == 1) {
+ assertThat(commit.commit(prepared, false)).isEqualTo(1);
+ assertThat(store.snapshotManager().latestSnapshot().id()).isEqualTo(4);
+ assertThat(store.readKvsFromSnapshot(4)).hasSize(4);
+ assertThat(acquisitions.get()).isEqualTo(2);
+ } else {
+ assertThatThrownBy(() -> commit.commit(prepared, false))
+ .satisfies(
+ anyCauseMatches(
+ "checked".equals(failureKind)
+ ? IOException.class
+ : IllegalStateException.class,
+ "Transient acquisition failure"));
+ assertThat(acquisitions.get()).isEqualTo(3);
+ assertThat(store.snapshotManager().latestSnapshot().id()).isEqualTo(2);
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testInterruptedAcquisitionDoesNotRetryOrPublish(boolean wrapped) throws Exception {
+ TestFileStore store = createStore(false);
+ ManifestCommittable prepared = prepareLockRetryCommittable(store);
+ AtomicInteger acquisitions = new AtomicInteger();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ if (acquisitions.incrementAndGet() == 1) {
+ // Interrupted blocking calls throw with the interrupt flag cleared.
+ InterruptedException interrupted =
+ new InterruptedException("Interrupted while acquiring lock");
+ if (wrapped) {
+ throw new RuntimeException(interrupted);
+ }
+ throw interrupted;
+ }
+ return () -> {};
+ }
+
+ @Override
+ public void close() {}
+ };
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store,
+ "writer",
+ new RenamingSnapshotCommit(store.snapshotManager(), lock))) {
+ assertThatThrownBy(() -> commit.commit(prepared, false))
+ .satisfies(
+ anyCauseMatches(
+ InterruptedException.class,
+ "Interrupted while acquiring lock"));
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ } finally {
+ Thread.interrupted();
+ }
+ assertThat(acquisitions.get()).isEqualTo(1);
+ assertThat(store.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @Test
+ public void testReleaseFailureKeepsSuccessfulPublicationAndCallbacks() throws Exception {
+ TestFileStore store = createStore(false);
+ ManifestCommittable prepared = prepareLockRetryCommittable(store);
+ AtomicInteger acquisitions = new AtomicInteger();
+ AtomicBoolean held = new AtomicBoolean();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ acquisitions.incrementAndGet();
+ held.set(true);
+ return () -> {
+ held.set(false);
+ throw new IOException("Release failed after publication");
+ };
+ }
+
+ @Override
+ public void close() {}
+ };
+ CommitCallback callback = mock(CommitCallback.class);
+ doAnswer(
+ invocation -> {
+ assertThat(held.get()).isFalse();
+ return null;
+ })
+ .when(callback)
+ .call(any());
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store,
+ "writer",
+ new RenamingSnapshotCommit(store.snapshotManager(), lock),
+ store.options(),
+ false,
+ Collections.singletonList(callback))) {
+ assertThat(commit.commit(prepared, false)).isEqualTo(1);
+ }
+ assertThat(acquisitions.get()).isEqualTo(1);
+ assertThat(store.snapshotManager().latestSnapshot().id()).isEqualTo(1);
+ assertThat(store.readKvsFromSnapshot(1)).hasSize(1);
+ verify(callback, times(1)).call(any());
+ }
+
+ @Test
+ public void testUnknownPublicationSurvivesLockAcquisitionFailure() throws Exception {
+ TestFileStore store = createStore(false);
+ ManifestCommittable prepared = prepareLockRetryCommittable(store);
+ AtomicInteger acquisitions = new AtomicInteger();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ if (acquisitions.incrementAndGet() == 2) {
+ throw new IOException("Transient acquisition failure");
+ }
+ return () -> {};
+ }
+
+ @Override
+ public void close() {}
+ };
+ TestingSnapshotCommit publisher =
+ mock(
+ TestingSnapshotCommit.class,
+ withSettings().useConstructor(lock).defaultAnswer(CALLS_REAL_METHODS));
+ SnapshotCommit delegate = new RenamingSnapshotCommit(store.snapshotManager(), Lock.empty());
+ when(publisher.commitImpl(any(), any(), any(), any()))
+ .thenAnswer(
+ invocation -> {
+ assertThat(
+ delegate.commit(
+ invocation.getArgument(0),
+ invocation.getArgument(1),
+ invocation.getArgument(2),
+ invocation.getArgument(3)))
+ .isTrue();
+ throw new IOException("Publication response lost");
+ });
+ try (FileStoreCommitImpl commit = newCommitWithSnapshotCommit(store, "writer", publisher)) {
+ assertThat(commit.commit(prepared, false)).isEqualTo(1);
+ }
+ assertThat(acquisitions.get()).isEqualTo(3);
+ verify(publisher, times(1)).commitImpl(any(), any(), any(), any());
+ assertThat(store.snapshotManager().latestSnapshot().id()).isEqualTo(1);
+ assertThat(store.readKvsFromSnapshot(1)).hasSize(1);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testPreCommitFailureIsNotRetriedOrMaskedByRelease(boolean fatalError)
+ throws Exception {
+ TestFileStore store = createStore(false);
+ ManifestCommittable prepared = prepareLockRetryCommittable(store);
+ AtomicInteger acquisitions = new AtomicInteger();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ acquisitions.incrementAndGet();
+ return () -> {
+ throw new IOException("Release failed after validation failure");
+ };
+ }
+
+ @Override
+ public void close() {}
+ };
+ Throwable failure =
+ fatalError
+ ? new AssertionError("Fatal validation failure")
+ : new IllegalStateException("Validation failure");
+ CommitPreCallback preCallback = mock(CommitPreCallback.class);
+ doAnswer(
+ invocation -> {
+ throw failure;
+ })
+ .when(preCallback)
+ .call(any(), any(), any(), any());
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store,
+ "writer",
+ new RenamingSnapshotCommit(store.snapshotManager(), lock),
+ store.options(),
+ false,
+ Collections.emptyList(),
+ Collections.singletonList(preCallback))) {
+ assertThatThrownBy(() -> commit.commit(prepared, false))
+ .isSameAs(failure)
+ .satisfies(
+ error ->
+ assertThat(error.getSuppressed())
+ .hasSize(1)
+ .allSatisfy(
+ suppressed ->
+ assertThat(suppressed)
+ .satisfies(
+ anyCauseMatches(
+ IOException
+ .class,
+ "Release failed after validation failure"))));
+ }
+ assertThat(acquisitions.get()).isEqualTo(1);
+ verify(preCallback, times(1)).call(any(), any(), any(), any());
+ assertThat(store.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWithoutLockPreCommitFailureFollowsOriginalPath(boolean interrupted)
+ throws Exception {
+ TestFileStore store = createStore(false);
+ ManifestCommittable prepared = prepareLockRetryCommittable(store);
+ RuntimeException failure =
+ interrupted
+ ? new RuntimeException(new InterruptedException("Action interrupted"))
+ : new IllegalStateException("Validation failed");
+ CommitPreCallback preCallback = mock(CommitPreCallback.class);
+ doAnswer(
+ invocation -> {
+ throw failure;
+ })
+ .when(preCallback)
+ .call(any(), any(), any(), any());
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store,
+ "writer",
+ new RenamingSnapshotCommit(store.snapshotManager(), Lock.empty()),
+ store.options(),
+ false,
+ Collections.emptyList(),
+ Collections.singletonList(preCallback))) {
+ assertThatThrownBy(() -> commit.commit(prepared, false))
+ .isSameAs(failure)
+ .hasNoSuppressedExceptions();
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ } finally {
+ Thread.interrupted();
+ }
+ verify(preCallback, times(1)).call(any(), any(), any(), any());
+ assertThat(store.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ private ManifestCommittable prepareLockRetryCommittable(TestFileStore store) throws Exception {
+ AtomicReference<ManifestCommittable> prepared = new AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L, null, "a")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> prepared.set(committable));
+ return checkNotNull(prepared.get());
+ }
+
+ @Test
+ public void testEveryAttemptLocksAndRefreshesHeadAfterAcquisition() throws Exception {
+ TestFileStore store = createStore(false);
+ store.commitData(
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L, null, "a")),
+ gen::getPartition,
+ value -> 0);
+ AtomicReference<ManifestCommittable> prepared = new AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 2L, null, "b")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> prepared.set(committable));
+ SnapshotCommit delegate = new RenamingSnapshotCommit(store.snapshotManager(), Lock.empty());
+ List<String> events = new ArrayList<>();
+ AtomicInteger attempts = new AtomicInteger();
+ AtomicBoolean held = new AtomicBoolean();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ assertThat(held.get()).isFalse();
+ int attempt = attempts.incrementAndGet();
+ // A writer publishes while this writer is waiting to acquire its first
+ // lock.
+ if (attempt == 1) {
+ store.commitData(
+ Collections.singletonList(
+ gen.nextInsert("20211110", 10, 3L, null, "c")),
+ gen::getPartition,
+ value -> 0);
+ }
+ held.set(true);
+ events.add("acquire");
+ return () -> {
+ held.set(false);
+ events.add("release");
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ assertThat(held.get()).isTrue();
+ }
+
+ @Override
+ public void close() {}
+ };
+ TestingSnapshotCommit publisher =
+ mock(
+ TestingSnapshotCommit.class,
+ withSettings().useConstructor(lock).defaultAnswer(CALLS_REAL_METHODS));
+ when(publisher.commitImpl(any(), any(), any(), any()))
+ .thenAnswer(
+ invocation -> {
+ assertThat(held.get()).isTrue();
+ Snapshot head = store.snapshotManager().latestSnapshot();
+ assertThat((String) invocation.getArgument(0)).isEqualTo(head.uuid());
+ assertThat(((Snapshot) invocation.getArgument(1)).id())
+ .isEqualTo(head.id() + 1);
+ events.add("publish");
+ // Force one conflict to verify that the next attempt obtains a new
+ // scope as well.
+ if (attempts.get() == 1) {
+ return false;
+ }
+ return delegate.commit(
+ invocation.getArgument(0),
+ invocation.getArgument(1),
+ invocation.getArgument(2),
+ invocation.getArgument(3));
+ });
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(store, "morax-job", publisher)) {
+ commit.commit(checkNotNull(prepared.get()), false);
+ }
+ Snapshot latest = store.snapshotManager().latestSnapshot();
+ assertThat(latest.commitUser()).isEqualTo("morax-job");
+ assertThat(store.readKvsFromSnapshot(latest.id())).hasSize(3);
+ assertThat(events)
+ .containsExactly("acquire", "publish", "release", "acquire", "publish", "release");
+ }
+
+ @Test
+ public void testAppendAndCompactAcquireSeparateScopesAndCallbacksRunAfterRelease()
+ throws Exception {
+ TestFileStore store = createStore(false);
+ KeyValue row = gen.nextInsert("20211110", 8, 1L, null, "a");
+ List<ManifestCommittable> prepared = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ store.commitDataImpl(
+ Collections.singletonList(row),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> prepared.add(committable));
+ }
+ CommitMessageImpl append = (CommitMessageImpl) prepared.get(0).fileCommittables().get(0);
+ CommitMessageImpl replacement =
+ (CommitMessageImpl) prepared.get(1).fileCommittables().get(0);
+ ManifestCommittable committable = new ManifestCommittable(23L);
+ committable.addFileCommittable(
+ new CommitMessageImpl(
+ append.partition(),
+ append.bucket(),
+ append.totalBuckets(),
+ append.newFilesIncrement(),
+ new CompactIncrement(
+ append.newFilesIncrement().newFiles(),
+ replacement.newFilesIncrement().newFiles(),
+ Collections.emptyList())));
+ List<String> events = new ArrayList<>();
+ AtomicBoolean held = new AtomicBoolean();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ assertThat(held.getAndSet(true)).isFalse();
+ events.add("acquire");
+ return () -> {
+ held.set(false);
+ events.add("release");
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ assertThat(held.get()).isTrue();
+ }
+
+ @Override
+ public void close() {}
+ };
+ TestingSnapshotCommit publisher =
+ mock(
+ TestingSnapshotCommit.class,
+ withSettings().useConstructor(lock).defaultAnswer(CALLS_REAL_METHODS));
+ SnapshotCommit delegate = new RenamingSnapshotCommit(store.snapshotManager(), Lock.empty());
+ when(publisher.commitImpl(any(), any(), any(), any()))
+ .thenAnswer(
+ invocation -> {
+ assertThat(held.get()).isTrue();
+ events.add("publish");
+ return delegate.commit(
+ invocation.getArgument(0),
+ invocation.getArgument(1),
+ invocation.getArgument(2),
+ invocation.getArgument(3));
+ });
+ CommitCallback callback = mock(CommitCallback.class);
+ doAnswer(
+ invocation -> {
+ assertThat(held.get()).isFalse();
+ events.add("callback");
+ return null;
+ })
+ .when(callback)
+ .call(any());
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store,
+ "writer",
+ publisher,
+ store.options(),
+ false,
+ Collections.singletonList(callback))) {
+ commit.commit(committable, false);
+ }
+ assertThat(store.snapshotManager().snapshot(1).commitKind())
+ .isEqualTo(Snapshot.CommitKind.APPEND);
+ assertThat(store.snapshotManager().snapshot(2).commitKind())
+ .isEqualTo(Snapshot.CommitKind.COMPACT);
+ assertThat(events)
+ .containsExactly(
+ "acquire",
+ "publish",
+ "release",
+ "callback",
+ "acquire",
+ "publish",
+ "release",
+ "callback");
+ assertThat(store.readKvsFromSnapshot(2)).hasSize(1);
+ }
+
+ @Test
+ public void testInvalidLockPreventsSnapshotPublication() throws Exception {
+ TestFileStore store = createStore(false);
+ AtomicReference<ManifestCommittable> prepared = new AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.next()),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> prepared.set(committable));
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ return () -> {};
+ }
+
+ @Override
+ public void ensureValid() {
+ throw new IllegalStateException("Lost lease");
+ }
+
+ @Override
+ public void close() {}
+ };
+ TestingSnapshotCommit publisher =
+ mock(
+ TestingSnapshotCommit.class,
+ withSettings().useConstructor(lock).defaultAnswer(CALLS_REAL_METHODS));
+ Options options = Options.fromMap(store.options().toMap());
+ options.set(CoreOptions.COMMIT_MAX_RETRIES, 0);
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store, "writer", publisher, new CoreOptions(options), false)) {
+ assertThatThrownBy(() -> commit.commit(checkNotNull(prepared.get()), false))
+ .satisfies(anyCauseMatches(IllegalStateException.class, "Lost lease"));
+ }
+ verify(publisher, never()).commitImpl(any(), any(), any(), any());
+ assertThat(store.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"append", "replace-manifest-list", "rollback", "compact-manifest"})
+ public void testLeaseLostDuringPreparationPreventsPublication(String operation)
+ throws Exception {
+ TestFileStore store = createStore(false);
+ List<KeyValue> rows =
+ Collections.singletonList(gen.nextInsert("20211110", 8, 1L, null, "a"));
+ Snapshot target = store.commitData(rows, gen::getPartition, value -> 0).get(0);
+ store.overwriteData(rows, gen::getPartition, value -> 0, Collections.emptyMap());
+ Snapshot head = store.snapshotManager().latestSnapshot();
+ AtomicReference<ManifestCommittable> prepared = new AtomicReference<>();
+ store.commitDataImpl(
+ Collections.singletonList(gen.nextInsert("20211110", 9, 2L, null, "b")),
+ gen::getPartition,
+ value -> 0,
+ false,
+ 23L,
+ null,
+ Collections.emptyList(),
+ (commit, committable) -> prepared.set(committable));
+ AtomicBoolean held = new AtomicBoolean();
+ AtomicInteger validityChecks = new AtomicInteger();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ assertThat(held.getAndSet(true)).isFalse();
+ return () -> {
+ held.set(false);
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ assertThat(held.get()).isTrue();
+ // The lease is valid on acquisition but expires during preparation.
+ if (validityChecks.incrementAndGet() > 1) {
+ throw new IllegalStateException("Lost lease");
+ }
+ }
+
+ @Override
+ public void close() {}
+ };
+ TestingSnapshotCommit publisher =
+ mock(
+ TestingSnapshotCommit.class,
+ withSettings().useConstructor(lock).defaultAnswer(CALLS_REAL_METHODS));
+ Options options = Options.fromMap(store.options().toMap());
+ options.set(CoreOptions.COMMIT_MAX_RETRIES, 0);
+ try (FileStoreCommitImpl commit =
+ newCommitWithSnapshotCommit(
+ store, "writer", publisher, new CoreOptions(options), false)) {
+ assertThatThrownBy(
+ () -> {
+ switch (operation) {
+ case "append":
+ commit.commit(checkNotNull(prepared.get()), false);
+ break;
+ case "replace-manifest-list":
+ commit.replaceManifestList(
+ head,
+ head.totalRecordCount(),
+ Pair.of(
+ head.baseManifestList(),
+ head.baseManifestListSize()),
+ Pair.of(
+ head.deltaManifestList(),
+ head.deltaManifestListSize()));
+ break;
+ case "rollback":
+ commit.rollbackToAsLatest(target);
+ break;
+ case "compact-manifest":
+ commit.compactManifest();
+ break;
+ default:
+ throw new AssertionError(operation);
+ }
+ })
+ .satisfies(anyCauseMatches(IllegalStateException.class, "Lost lease"));
+ }
+ verify(publisher, never()).commitImpl(any(), any(), any(), any());
+ assertThat(validityChecks.get()).isEqualTo(2);
+ assertThat(held.get()).isFalse();
+ assertThat(store.snapshotManager().latestSnapshot()).isEqualTo(head);
+ }
+
+ @Test
+ public void testManifestReplacementAcceptsLegacySnapshotWithoutUuid() throws Exception {
+ TestFileStore store = createStore(false);
+ store.commitData(Collections.singletonList(gen.next()), gen::getPartition, value -> 0);
+ Snapshot snapshot = store.snapshotManager().latestSnapshot();
+ ObjectNode legacyJson = JsonSerdeUtil.fromJson(snapshot.toJson(), ObjectNode.class);
+ legacyJson.remove("uuid");
+ Snapshot legacy = Snapshot.fromJson(legacyJson.toString());
+ store.fileIO()
+ .overwriteFileUtf8(
+ store.snapshotManager().snapshotPath(snapshot.id()), legacyJson.toString());
+ assertThat(store.snapshotManager().latestSnapshot().uuid()).isNull();
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ assertThat(
+ commit.replaceManifestList(
+ legacy,
+ legacy.totalRecordCount(),
+ Pair.of(
+ legacy.baseManifestList(),
+ legacy.baseManifestListSize()),
+ Pair.of(
+ legacy.deltaManifestList(),
+ legacy.deltaManifestListSize())))
+ .isTrue();
+ }
+ assertThat(store.readKvsFromSnapshot(2)).hasSize(1);
+ }
+
@Test
public void testCommitRetryAfterFalseSuccessDoesNotCleanManifest() throws Exception {
TestFileStore store = createStore(false);
@@ -1766,9 +2452,10 @@ public class FileStoreCommitTest {
assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(1);
}
- @Test
- public void testCommitRetryReusePreviousManifestMergeResultWhenBeforeStillExists()
- throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testCommitRetryReusePreviousManifestMergeResultWhenBeforeStillExists(
+ boolean lockFailures) throws Exception {
TestFileStore store =
createStore(
false,
@@ -1795,6 +2482,25 @@ public class FileStoreCommitTest {
Collections.emptyList(),
(commit, committable) -> committableRef.set(committable));
+ AtomicInteger acquisitions = new AtomicInteger();
+ Lock lock =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ int attempt = acquisitions.incrementAndGet();
+ if (lockFailures && attempt == 2) {
+ throw new IOException("Acquisition failed between publication retries");
+ }
+ return () -> {
+ if (lockFailures && attempt == 1) {
+ throw new IOException("Release failed after publication conflict");
+ }
+ };
+ }
+
+ @Override
+ public void close() {}
+ };
ConflictingSnapshotCommit snapshotCommit =
new ConflictingSnapshotCommit(
new RenamingSnapshotCommit(store.snapshotManager(), Lock.empty()),
@@ -1802,7 +2508,8 @@ public class FileStoreCommitTest {
store.manifestListFactory().create(),
store.manifestFileFactory().create(),
false,
- conflictAttempts(Collections.emptyList()));
+ conflictAttempts(Collections.emptyList()),
+ lock);
try (FileStoreCommitImpl commit =
newCommitWithSnapshotCommit(store, "retry-reuse-merge", snapshotCommit)) {
commit.commit(checkNotNull(committableRef.get()), false);
@@ -1818,6 +2525,7 @@ public class FileStoreCommitTest {
assertThat(finalBaseManifests)
.containsExactlyElementsOf(snapshotCommit.firstAttemptBaseManifests());
assertThat(store.readKvsFromSnapshot(latestSnapshot.id())).hasSize(3);
+ assertThat(acquisitions.get()).isEqualTo(lockFailures ? 3 : 2);
}
@Test
@@ -2193,6 +2901,40 @@ public class FileStoreCommitTest {
SnapshotCommit snapshotCommit,
CoreOptions options,
boolean dataEvolutionEnabled) {
+ return newCommitWithSnapshotCommit(
+ store,
+ commitUser,
+ snapshotCommit,
+ options,
+ dataEvolutionEnabled,
+ Collections.emptyList());
+ }
+
+ private FileStoreCommitImpl newCommitWithSnapshotCommit(
+ TestFileStore store,
+ String commitUser,
+ SnapshotCommit snapshotCommit,
+ CoreOptions options,
+ boolean dataEvolutionEnabled,
+ List<CommitCallback> callbacks) {
+ return newCommitWithSnapshotCommit(
+ store,
+ commitUser,
+ snapshotCommit,
+ options,
+ dataEvolutionEnabled,
+ callbacks,
+ Collections.emptyList());
+ }
+
+ private FileStoreCommitImpl newCommitWithSnapshotCommit(
+ TestFileStore store,
+ String commitUser,
+ SnapshotCommit snapshotCommit,
+ CoreOptions options,
+ boolean dataEvolutionEnabled,
+ List<CommitCallback> callbacks,
+ List<CommitPreCallback> preCallbacks) {
String tableName = store.options().path().getName();
return new FileStoreCommitImpl(
snapshotCommit,
@@ -2210,8 +2952,8 @@ public class FileStoreCommitTest {
store::newScan,
store.newStatsFileHandler(),
store.bucketMode(),
- Collections.emptyList(),
- Collections.emptyList(),
+ preCallbacks,
+ callbacks,
scanner ->
ConflictDetection.create(
tableName,
@@ -2286,7 +3028,22 @@ public class FileStoreCommitTest {
return Arrays.asList(attempts);
}
- private static class ConflictingSnapshotCommit implements SnapshotCommit {
+ /** Exposes the backend hook to Mockito while retaining the final publication scope. */
+ public abstract static class TestingSnapshotCommit extends SnapshotCommit {
+ public TestingSnapshotCommit(Lock lock) {
+ super(lock);
+ }
+
+ @Override
+ public abstract boolean commitImpl(
+ @Nullable String baseSnapshotUuid,
+ Snapshot snapshot,
+ String branch,
+ List<org.apache.paimon.partition.PartitionStatistics> statistics)
+ throws Exception;
+ }
+
+ private static class ConflictingSnapshotCommit extends SnapshotCommit {
private final SnapshotCommit delegate;
private final SnapshotManager snapshotManager;
@@ -2306,6 +3063,25 @@ public class FileStoreCommitTest {
ManifestFile manifestFile,
boolean mergeConflictManifests,
List<List<ManifestEntry>> conflictDeltaFilesByAttempt) {
+ this(
+ delegate,
+ snapshotManager,
+ manifestList,
+ manifestFile,
+ mergeConflictManifests,
+ conflictDeltaFilesByAttempt,
+ Lock.empty());
+ }
+
+ private ConflictingSnapshotCommit(
+ SnapshotCommit delegate,
+ SnapshotManager snapshotManager,
+ ManifestList manifestList,
+ ManifestFile manifestFile,
+ boolean mergeConflictManifests,
+ List<List<ManifestEntry>> conflictDeltaFilesByAttempt,
+ Lock lock) {
+ super(lock);
this.delegate = delegate;
this.snapshotManager = snapshotManager;
this.manifestList = manifestList;
@@ -2317,7 +3093,7 @@ public class FileStoreCommitTest {
}
@Override
- public boolean commit(
+ protected boolean commitImpl(
@Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
@@ -2449,22 +3225,23 @@ public class FileStoreCommitTest {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
delegate.close();
}
}
- private static class FalseSuccessSnapshotCommit implements SnapshotCommit {
+ private static class FalseSuccessSnapshotCommit extends SnapshotCommit {
private final SnapshotCommit delegate;
private boolean firstCommit = true;
private FalseSuccessSnapshotCommit(SnapshotCommit delegate) {
+ super(Lock.empty());
this.delegate = delegate;
}
@Override
- public boolean commit(
+ protected boolean commitImpl(
@Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
@@ -2480,7 +3257,7 @@ public class FileStoreCommitTest {
}
@Override
- public void close() throws Exception {
+ protected void closeImpl() throws Exception {
delegate.close();
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/LockTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/LockTest.java
new file mode 100644
index 0000000000..563a7dfc9a
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/LockTest.java
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests validity checks and reentrancy of operation locks. */
+class LockTest {
+ @Test
+ void emptyLockBypassesLockHandlingAndPropagatesTheOriginalFailure() throws Exception {
+ Lock empty = Lock.empty();
+ assertThat(Lock.reentrant(empty)).isSameAs(empty);
+ Object result = new Object();
+ assertThat(empty.runWithLock(() -> result)).isSameAs(result);
+ InterruptedException failure = new InterruptedException("Action interrupted");
+ assertThatThrownBy(
+ () ->
+ empty.runWithLock(
+ () -> {
+ throw failure;
+ }))
+ .isSameAs(failure)
+ .hasNoSuppressedExceptions();
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ }
+
+ @Test
+ void emptyLockDoesNotAcquireOrValidateAScope() throws Exception {
+ Lock empty =
+ new Lock.EmptyLock() {
+ @Override
+ public AutoCloseable acquire() {
+ throw new AssertionError("No-lock path must not acquire a scope");
+ }
+
+ @Override
+ public void ensureValid() {
+ throw new AssertionError("No-lock path must not validate a lease");
+ }
+ };
+ Object result = new Object();
+ assertThat(Lock.reentrant(empty).runWithLock(() -> result)).isSameAs(result);
+ }
+
+ @Test
+ void releaseFailureKeepsSuccessfulResult() throws Exception {
+ ScopeLock backend = new ScopeLock();
+ backend.releaseFailure = new IOException("Release failed");
+ Lock lock = Lock.reentrant(backend);
+ Object result = new Object();
+ assertThat(lock.runWithLock(() -> result)).isSameAs(result);
+ assertThat(backend.releases).isEqualTo(1);
+ assertThat(lock.<Object>runWithLock(() -> null)).isNull();
+ assertThat(backend.releases).isEqualTo(2);
+ }
+
+ @Test
+ void releaseFailureDoesNotMaskTheActionFailure() {
+ ScopeLock backend = new ScopeLock();
+ backend.releaseFailure = new IOException("Release failed");
+ RuntimeException failure = new RuntimeException("Action failed");
+ assertThatThrownBy(
+ () ->
+ Lock.reentrant(backend)
+ .runWithLock(
+ () -> {
+ throw failure;
+ }))
+ .isSameAs(failure)
+ .hasSuppressedException(backend.releaseFailure);
+ assertThat(backend.releases).isEqualTo(1);
+ }
+
+ @Test
+ void acquisitionFailureNeverRunsTheOperation() {
+ ScopeLock lock = new ScopeLock();
+ IOException failure = new IOException("Acquisition failed");
+ lock.acquisitionFailure = failure;
+ AtomicInteger operations = new AtomicInteger();
+ assertThatThrownBy(() -> lock.runWithLock(operations::incrementAndGet))
+ .isInstanceOf(Lock.AcquisitionException.class)
+ .hasCause(failure);
+ assertThat(operations.get()).isZero();
+ assertThat(lock.releases).isZero();
+ }
+
+ @Test
+ void checkedOperationFailureIsPropagatedWithoutLockClassification() {
+ ScopeLock lock = new ScopeLock();
+ IOException failure = new IOException("Operation failed");
+ assertThatThrownBy(
+ () ->
+ lock.runWithLock(
+ () -> {
+ throw failure;
+ }))
+ .isSameAs(failure)
+ .hasNoSuppressedExceptions();
+ assertThat(lock.releases).isEqualTo(1);
+ }
+
+ private static class ScopeLock implements Lock {
+ private Exception acquisitionFailure;
+ private Exception releaseFailure;
+ private int releases;
+
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ if (acquisitionFailure != null) {
+ throw acquisitionFailure;
+ }
+ return () -> {
+ releases++;
+ if (releaseFailure != null) {
+ throw releaseFailure;
+ }
+ };
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ @Test
+ void nestedPublicationUsesOneScopeAndChecksValidity() throws Exception {
+ List<String> events = new ArrayList<>();
+ Lock backend =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ events.add("acquire");
+ return () -> {
+ events.add("release");
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ events.add("validate");
+ }
+
+ @Override
+ public void close() {
+ events.add("close");
+ }
+ };
+ try (Lock lock = Lock.reentrant(backend)) {
+ lock.runWithLock(
+ () -> {
+ events.add("head");
+ return lock.runWithLock(
+ () -> {
+ events.add("publish");
+ return true;
+ });
+ });
+ lock.runWithLock(
+ () -> {
+ events.add("next");
+ return true;
+ });
+ }
+ assertThat(events)
+ .isEqualTo(
+ Arrays.asList(
+ "acquire",
+ "validate",
+ "head",
+ "validate",
+ "publish",
+ "release",
+ "acquire",
+ "validate",
+ "next",
+ "release",
+ "close"));
+ }
+
+ @Test
+ void expiredLeasePreventsNestedPublicationAndScopeIsReleased() throws Exception {
+ List<String> events = new ArrayList<>();
+ AtomicBoolean expired = new AtomicBoolean();
+ Lock backend =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ events.add("acquire");
+ return () -> {
+ events.add("release");
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ if (expired.get()) {
+ throw new IllegalStateException("Expired");
+ }
+ }
+
+ @Override
+ public void close() {}
+ };
+ Lock lock = Lock.reentrant(backend);
+ assertThatThrownBy(
+ () ->
+ lock.runWithLock(
+ () -> {
+ expired.set(true);
+ return lock.runWithLock(
+ () -> {
+ events.add("publish");
+ return null;
+ });
+ }))
+ .hasMessage("Expired");
+ assertThat(events).containsExactly("acquire", "release");
+ expired.set(false);
+ lock.runWithLock(() -> null);
+ assertThat(events).containsExactly("acquire", "release", "acquire", "release");
+ }
+
+ @Test
+ void failedInitialValidationReleasesScopeBeforeRetry() throws Exception {
+ List<String> events = new ArrayList<>();
+ AtomicBoolean expired = new AtomicBoolean(true);
+ Lock backend =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ events.add("acquire");
+ return () -> {
+ events.add("release");
+ };
+ }
+
+ @Override
+ public void ensureValid() {
+ if (expired.get()) {
+ throw new IllegalStateException("Expired");
+ }
+ }
+
+ @Override
+ public void close() {}
+ };
+ Lock lock = Lock.reentrant(backend);
+ assertThatThrownBy(
+ () ->
+ lock.runWithLock(
+ () -> {
+ events.add("publish");
+ return null;
+ }))
+ .isInstanceOf(Lock.AcquisitionException.class)
+ .hasRootCauseMessage("Expired");
+ assertThat(events).containsExactly("acquire", "release");
+ expired.set(false);
+ lock.runWithLock(
+ () -> {
+ events.add("publish");
+ return null;
+ });
+ assertThat(events).containsExactly("acquire", "release", "acquire", "publish", "release");
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
index 011975188c..c8c5f4bb63 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/CatalogEnvironmentTest.java
@@ -19,10 +19,16 @@
package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogLockContext;
+import org.apache.paimon.catalog.CatalogLockFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.options.Options;
import org.apache.paimon.rest.RESTApi;
import org.apache.paimon.rest.RESTCatalogFactory;
@@ -32,15 +38,25 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.IOException;
import java.util.Collections;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.paimon.options.CatalogOptions.METASTORE;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
/** Tests for {@link CatalogEnvironment}. */
@@ -138,6 +154,242 @@ class CatalogEnvironmentTest {
verify(environment).dependencyReadContext();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testWriterLockUsesCatalogAndRuntimeOptions(boolean supportsVersionManagement)
+ throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Lock delegate = mock(Lock.class);
+ when(delegate.runWithLock(any()))
+ .thenAnswer(invocation -> ((Callable<?>) invocation.getArgument(0)).call());
+ Options options = new Options();
+ options.set(CoreOptions.BRANCH, "dev");
+ Identifier branch = new Identifier("db", "table", "dev");
+ when(catalog.createLock(branch, "table-id", "writer", options)).thenReturn(delegate);
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ "table-id",
+ () -> catalog,
+ null,
+ null,
+ null,
+ supportsVersionManagement,
+ false);
+ try (SnapshotCommit publisher =
+ environment.snapshotCommit(
+ mock(SnapshotManager.class), new CoreOptions(options), "writer")) {
+ assertThat(publisher.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ verify(catalog).createLock(branch, "table-id", "writer", options);
+ verify(delegate).ensureValid();
+ verify(delegate).close();
+ verify(catalog).close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testLockCloseFailureStillClosesCatalog(boolean supportsVersionManagement)
+ throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Lock lock = mock(Lock.class);
+ when(catalog.createLock(any(), any(), any(), any())).thenReturn(lock);
+ doThrow(new IOException("Lock close failed")).when(lock).close();
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ "table-id",
+ () -> catalog,
+ null,
+ null,
+ null,
+ supportsVersionManagement,
+ false);
+ SnapshotCommit publisher =
+ environment.snapshotCommit(
+ mock(SnapshotManager.class), new CoreOptions(new Options()), "writer");
+ assertThatThrownBy(publisher::close)
+ .isInstanceOf(IOException.class)
+ .hasMessage("Lock close failed");
+ verify(lock).close();
+ verify(catalog).close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testPublisherCreationFailureClosesResources(boolean useCatalogLoader) throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Lock lock = mock(Lock.class);
+ when(catalog.createLock(any(), any(), any(), any())).thenReturn(lock);
+ CatalogLockFactory factory = mock(CatalogLockFactory.class);
+ when(factory.createLock(any(), any(), any(), any(), any())).thenReturn(lock);
+ SnapshotManager manager = mock(SnapshotManager.class);
+ when(manager.fileIO()).thenThrow(new IllegalStateException("IO unavailable"));
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ "table-id",
+ useCatalogLoader ? () -> catalog : null,
+ factory,
+ CatalogLockContext.fromOptions(new Options()),
+ null,
+ false,
+ false);
+ assertThatThrownBy(
+ () ->
+ environment.snapshotCommit(
+ manager, new CoreOptions(new Options()), "writer"))
+ .hasMessage("IO unavailable");
+ verify(lock).close();
+ if (useCatalogLoader) {
+ verify(catalog).close();
+ } else {
+ verifyNoInteractions(catalog);
+ }
+ }
+
+ @Test
+ void testFactoryBindsRuntimeIdentityWithoutCatalogLoader() throws Exception {
+ CatalogLockFactory factory = mock(CatalogLockFactory.class);
+ CatalogLockContext context = CatalogLockContext.fromOptions(new Options());
+ Lock backend =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ return () -> {};
+ }
+
+ @Override
+ public void close() {}
+ };
+ Options options = new Options();
+ options.set(CoreOptions.BRANCH, "dev");
+ Identifier branch = new Identifier("db", "table", "dev");
+ when(factory.createLock(context, branch, "table-id", "writer", options))
+ .thenReturn(backend);
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ "table-id",
+ null,
+ factory,
+ context,
+ null,
+ false,
+ false);
+ try (SnapshotCommit publisher =
+ environment.snapshotCommit(
+ mock(SnapshotManager.class), new CoreOptions(options), "writer")) {
+ assertThat(publisher.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ verify(factory).createLock(context, branch, "table-id", "writer", options);
+ }
+
+ @Test
+ void testCatalogPolicyIsAuthoritativeOverSerializedFactory() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ CatalogLockFactory factory = mock(CatalogLockFactory.class);
+ when(catalog.createLock(any(), any(), any(), any())).thenReturn(Lock.empty());
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ "table-id",
+ () -> catalog,
+ factory,
+ CatalogLockContext.fromOptions(new Options()),
+ null,
+ false,
+ false);
+ try (SnapshotCommit publisher =
+ environment.snapshotCommit(
+ mock(SnapshotManager.class), new CoreOptions(new Options()), "writer")) {
+ assertThat(publisher.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ verifyNoInteractions(factory);
+ verify(catalog).close();
+ }
+
+ @Test
+ void testFailedLockCreationClosesCatalog() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.createLock(any(), any(), any(), any()))
+ .thenThrow(new IllegalArgumentException("Unsupported"));
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ "table-id",
+ () -> catalog,
+ null,
+ null,
+ null,
+ true,
+ false);
+ assertThatThrownBy(
+ () ->
+ environment.snapshotCommit(
+ mock(SnapshotManager.class),
+ new CoreOptions(new Options()),
+ "writer"))
+ .hasMessage("Unsupported");
+ verify(catalog).close();
+ }
+
+ @Test
+ void testDirectSnapshotCommitUsesBoundPublicationLock() throws Exception {
+ CatalogLockFactory factory = mock(CatalogLockFactory.class);
+ AtomicBoolean held = new AtomicBoolean();
+ Lock backend =
+ new Lock() {
+ @Override
+ public AutoCloseable acquire() throws Exception {
+ held.set(true);
+ return () -> {
+ held.set(false);
+ };
+ }
+
+ @Override
+ public void close() {}
+ };
+ CatalogLockContext context = CatalogLockContext.fromOptions(new Options());
+ when(factory.createLock(
+ context, Identifier.create("db", "table"), null, "writer", new Options()))
+ .thenReturn(backend);
+ SnapshotManager manager = mock(SnapshotManager.class);
+ FileIO fileIO = mock(FileIO.class);
+ Path snapshotPath = new Path("file:/table/snapshot/snapshot-1");
+ when(manager.fileIO()).thenReturn(fileIO);
+ when(manager.branch()).thenReturn("main");
+ when(manager.snapshotPath(1L)).thenReturn(snapshotPath);
+ when(fileIO.tryToWriteAtomic(snapshotPath, "{}")).thenReturn(true);
+ Snapshot snapshot = mock(Snapshot.class);
+ when(snapshot.id()).thenReturn(1L);
+ when(snapshot.toJson())
+ .thenAnswer(
+ invocation -> {
+ assertThat(held.get()).isTrue();
+ return "{}";
+ });
+ CatalogEnvironment environment =
+ new CatalogEnvironment(
+ Identifier.create("db", "table"),
+ null,
+ null,
+ factory,
+ context,
+ null,
+ false,
+ false);
+ try (SnapshotCommit publisher =
+ environment.snapshotCommit(manager, new CoreOptions(new Options()), "writer")) {
+ assertThat(publisher.commit(null, snapshot, "main", Collections.emptyList())).isTrue();
+ }
+ assertThat(held.get()).isFalse();
+ verify(factory)
+ .createLock(
+ context, Identifier.create("db", "table"), null, "writer", new Options());
+ }
+
private static CatalogEnvironment environment(
Identifier identifier, CatalogContext catalogContext) {
return new CatalogEnvironment(
diff --git a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
index c48309213e..f22e050a9d 100644
--- a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -17,3 +17,5 @@ org.apache.paimon.mergetree.compact.aggregate.TestCustomAggFactory
org.apache.paimon.mergetree.compact.aggregate.TestMapOnlyAggFactory
org.apache.paimon.rest.auth.CustomTestDLFTokenLoaderFactory
org.apache.paimon.iceberg.RecordingIcebergMetadataCommitter$Factory
+
+org.apache.paimon.catalog.TestCatalogLockFactory
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
index 915c93680a..bb8acd72bb 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
@@ -19,12 +19,13 @@
package org.apache.paimon.flink;
import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.CatalogLock;
import org.apache.paimon.catalog.CatalogLockContext;
import org.apache.paimon.catalog.CatalogLockFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.util.AbstractTestBase;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
import org.apache.paimon.utils.BlockingIterator;
import org.apache.flink.table.api.TableEnvironment;
@@ -38,7 +39,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
@@ -148,7 +148,7 @@ public class FileSystemCatalogITCase extends AbstractTestBase {
}
@Test
- void testCatalogWithLockForSchema() throws Exception {
+ void testCatalogIgnoresBackendLockOptionsForSchema() throws Exception {
LOCK_COUNT.set(0);
tEnv.executeSql(
String.format(
@@ -165,7 +165,9 @@ public class FileSystemCatalogITCase extends AbstractTestBase {
tEnv.executeSql("CREATE TABLE table2 (a STRING, b STRING, c STRING)").await();
tEnv.executeSql("CREATE TABLE table3 (a STRING, b STRING, c STRING)").await();
tEnv.executeSql("DROP TABLE table3").await();
- assertThat(LOCK_COUNT.get()).isEqualTo(3);
+ assertThat(LOCK_COUNT.get()).isZero();
+ assertThat(collect("SHOW TABLES"))
+ .containsExactlyInAnyOrder(Row.of("table1"), Row.of("table2"));
}
private void innerTestWriteRead() throws Exception {
@@ -197,13 +199,17 @@ public class FileSystemCatalogITCase extends AbstractTestBase {
}
@Override
- public CatalogLock createLock(CatalogLockContext context) {
- return new CatalogLock() {
+ public Lock createLock(
+ CatalogLockContext context,
+ Identifier identifier,
+ String tableUuid,
+ String commitUser,
+ Options tableOptions) {
+ return new Lock() {
@Override
- public <T> T runWithLock(String database, String table, Callable<T> callable)
- throws Exception {
+ public AutoCloseable acquire() throws Exception {
LOCK_COUNT.incrementAndGet();
- return callable.call();
+ return () -> {};
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogLockITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogLockITCase.java
new file mode 100644
index 0000000000..84d371d68a
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogLockITCase.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.util.AbstractTestBase;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.table.api.TableEnvironment;
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests that SQL schema changes participate in the JDBC catalog's table locks. */
+public class JdbcCatalogLockITCase extends AbstractTestBase {
+
+ @Test
+ void testSchemaChangesAcquireCatalogLock() throws Exception {
+ TableEnvironment tEnv = tableEnvironmentBuilder().batchMode().build();
+ tEnv.executeSql(
+ String.format(
+ "CREATE CATALOG jdbc WITH ("
+ + "'type'='paimon', "
+ + "'metastore'='jdbc', "
+ + "'warehouse'='%s', "
+ + "'uri'='jdbc:sqlite:file:%s?mode=memory&cache=shared', "
+ + "'lock.enabled'='true', "
+ + "'lock-acquire-timeout'='1 s', "
+ + "'lock-check-max-sleep'='10 ms')",
+ getTempDirPath(), UUID.randomUUID()))
+ .await();
+ tEnv.useCatalog("jdbc");
+
+ Identifier identifier = Identifier.create("default", "table1");
+ String createTable = "CREATE TABLE table1 (a STRING, b STRING, c STRING)";
+ String alterTable = "ALTER TABLE table1 ADD (d STRING)";
+ try (Catalog catalog = ((FlinkCatalog) tEnv.getCatalog("jdbc").get()).catalog()) {
+ assertSchemaChangeBlocked(tEnv, catalog, identifier, createTable);
+ assertThat(catalog.listTables("default")).doesNotContain("table1");
+ tEnv.executeSql(createTable).await();
+
+ assertSchemaChangeBlocked(tEnv, catalog, identifier, alterTable);
+ assertThat(catalog.getTable(identifier).rowType().getFieldNames())
+ .containsExactly("a", "b", "c");
+ tEnv.executeSql(alterTable).await();
+ assertThat(catalog.getTable(identifier).rowType().getFieldNames())
+ .containsExactly("a", "b", "c", "d");
+ }
+ }
+
+ private void assertSchemaChangeBlocked(
+ TableEnvironment tEnv, Catalog catalog, Identifier identifier, String statement)
+ throws Exception {
+ try (Lock lock = catalog.createLock(identifier, null, "other-writer", new Options())) {
+ lock.runWithLock(
+ () -> {
+ assertThatThrownBy(() -> tEnv.executeSql(statement).await())
+ .rootCause()
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Acquire lock failed");
+ return null;
+ });
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogViewITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogViewITCase.java
index a3b1af7aeb..81d5868de1 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogViewITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/JdbcCatalogViewITCase.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink;
import org.apache.paimon.jdbc.JdbcCatalog;
+import org.apache.paimon.jdbc.JdbcCatalogOptions;
import org.apache.paimon.options.CatalogOptions;
import org.apache.flink.types.Row;
@@ -70,7 +71,7 @@ public class JdbcCatalogViewITCase extends CatalogITCaseBase {
options.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user");
options.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
// Disable lock for simpler testing
- options.put(CatalogOptions.LOCK_ENABLED.key(), "false");
+ options.put(JdbcCatalogOptions.LOCK_ENABLED.key(), "false");
return options;
}
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index c6bfee715e..e39235e9e8 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -123,6 +123,8 @@ import static org.apache.paimon.hive.HiveCatalogOptions.HIVE_CONF_DIR;
import static org.apache.paimon.hive.HiveCatalogOptions.HIVE_SKIP_UPDATE_STATS;
import static org.apache.paimon.hive.HiveCatalogOptions.IDENTIFIER;
import static org.apache.paimon.hive.HiveCatalogOptions.LOCATION_IN_PROPERTIES;
+import static org.apache.paimon.hive.HiveCatalogOptions.LOCK_ENABLED;
+import static org.apache.paimon.hive.HiveCatalogOptions.LOCK_TYPE;
import static org.apache.paimon.hive.HiveTableUtils.tryToFormatSchema;
import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE;
import static org.apache.paimon.options.CatalogOptions.FORMAT_TABLE_ENABLED;
@@ -213,6 +215,20 @@ public class HiveCatalog extends AbstractCatalog {
return !options.get(FORMAT_TABLE_ENABLED);
}
+ @Override
+ public Optional<CatalogLockFactory> lockFactory() {
+ if (!lockEnabled()) {
+ return Optional.empty();
+ }
+ String type = options.get(LOCK_TYPE);
+ return type == null ? defaultLockFactory() : Optional.of(CatalogLockFactory.discover(type));
+ }
+
+ @Override
+ protected boolean lockEnabled() {
+ return options.getOptional(LOCK_ENABLED).orElse(fileIO.isObjectStore());
+ }
+
@Override
public Optional<CatalogLockFactory> defaultLockFactory() {
return Optional.of(new HiveCatalogLockFactory());
@@ -1938,16 +1954,9 @@ public class HiveCatalog extends AbstractCatalog {
}
public <T> T runWithLock(Identifier identifier, Callable<T> callable) throws Exception {
- if (!lockEnabled()) {
- return callable.call();
+ try (Lock lock = createLock(identifier, null, null, new Options())) {
+ return lock.runWithLock(callable);
}
-
- HiveCatalogLock lock =
- new HiveCatalogLock(
- clients(),
- HiveCatalogLock.checkMaxSleep(hiveConf),
- HiveCatalogLock.acquireTimeout(hiveConf));
- return Lock.fromCatalog(lock, identifier).runWithLock(callable);
}
public static HiveConf createHiveConf(
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLock.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLock.java
index ea28d264c6..7daa5470e5 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLock.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLock.java
@@ -18,10 +18,12 @@
package org.apache.paimon.hive;
-import org.apache.paimon.catalog.CatalogLock;
+import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.client.ClientPool;
import org.apache.paimon.hive.pool.CachedClientPool;
+import org.apache.paimon.operation.Lock;
import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.ExecutorThreadFactory;
import org.apache.paimon.utils.TimeUtils;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -36,44 +38,191 @@ import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collections;
-import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongSupplier;
-import static org.apache.paimon.options.CatalogOptions.LOCK_ACQUIRE_TIMEOUT;
-import static org.apache.paimon.options.CatalogOptions.LOCK_CHECK_MAX_SLEEP;
+import static org.apache.paimon.hive.HiveCatalogOptions.LOCK_ACQUIRE_TIMEOUT;
+import static org.apache.paimon.hive.HiveCatalogOptions.LOCK_CHECK_MAX_SLEEP;
-/** Hive {@link CatalogLock}. */
-public class HiveCatalogLock implements CatalogLock {
+/** Hive catalog lock. */
+public class HiveCatalogLock implements Lock {
private static final Logger LOG = LoggerFactory.getLogger(HiveCatalogLock.class);
static final String LOCK_IDENTIFIER = "hive";
private final ClientPool<IMetaStoreClient, TException> clients;
+ private final Identifier identifier;
private final long checkMaxSleep;
private final long acquireTimeout;
+ private final long leaseMillis;
+ @Nullable private final ScheduledExecutorService renewer;
+ private final LongSupplier nanoTime;
+ private final ThreadLocal<Lease> currentLease = new ThreadLocal<>();
public HiveCatalogLock(
ClientPool<IMetaStoreClient, TException> clients,
+ Identifier identifier,
long checkMaxSleep,
long acquireTimeout) {
+ this(clients, identifier, checkMaxSleep, acquireTimeout, leaseTimeout(new HiveConf()));
+ }
+
+ public HiveCatalogLock(
+ ClientPool<IMetaStoreClient, TException> clients,
+ Identifier identifier,
+ long checkMaxSleep,
+ long acquireTimeout,
+ long leaseMillis) {
+ this(
+ clients,
+ identifier,
+ checkMaxSleep,
+ acquireTimeout,
+ leaseMillis,
+ null,
+ System::nanoTime);
+ }
+
+ HiveCatalogLock(
+ ClientPool<IMetaStoreClient, TException> clients,
+ Identifier identifier,
+ long checkMaxSleep,
+ long acquireTimeout,
+ long leaseMillis,
+ @Nullable ScheduledExecutorService renewer,
+ LongSupplier nanoTime) {
+ this.leaseMillis = leaseMillis;
+ this.renewer = renewer;
+ this.nanoTime = nanoTime;
this.clients = clients;
+ this.identifier = identifier;
this.checkMaxSleep = checkMaxSleep;
this.acquireTimeout = acquireTimeout;
}
@Override
- public <T> T runWithLock(String database, String table, Callable<T> callable) throws Exception {
- long lockId = lock(database, table);
+ public AutoCloseable acquire() throws Exception {
+ long lockId = lock(identifier.getDatabaseName(), identifier.getObjectName());
+ Lease lease = null;
try {
- return callable.call();
+ long requestedAt = nanoTime.getAsLong();
+ clients.execute(client -> client.heartbeat(0L, lockId));
+ lease = new Lease(lockId, requestedAt);
+ currentLease.set(lease);
+ lease.ensureValid();
+ Lease acquired = lease;
+ return () -> release(acquired, lockId);
+ } catch (Exception | Error e) {
+ Lease partial = lease;
+ Lock.release(() -> release(partial, lockId), e);
+ throw e;
+ }
+ }
+
+ private void release(@Nullable Lease lease, long lockId) throws Exception {
+ currentLease.remove();
+ try {
+ if (lease != null) {
+ lease.close();
+ }
} finally {
unlock(lockId);
}
}
+ @Override
+ public void ensureValid() {
+ Lease lease = currentLease.get();
+ if (lease == null) {
+ throw new IllegalStateException("No Hive lock is held by this thread.");
+ }
+ lease.heartbeat();
+ lease.ensureValid();
+ }
+
+ private static ScheduledThreadPoolExecutor createRenewer() {
+ ScheduledThreadPoolExecutor executor =
+ new ScheduledThreadPoolExecutor(1, new ExecutorThreadFactory("hive-lock-renew"));
+ executor.setRemoveOnCancelPolicy(true);
+ return executor;
+ }
+
+ private class Lease implements AutoCloseable {
+ private final ScheduledFuture<?> renewal;
+ private final ScheduledExecutorService executor;
+ private final long lockId;
+ private final AtomicBoolean closed = new AtomicBoolean();
+ private final AtomicLong renewedAt;
+ private volatile Exception failure;
+
+ private Lease(long lockId, long requestedAt) {
+ this.lockId = lockId;
+ renewedAt = new AtomicLong(requestedAt);
+ // A slow metastore heartbeat must not consume another scope's renewal worker.
+ executor = renewer == null ? createRenewer() : renewer;
+ long interval = Math.max(1, leaseMillis / 3);
+ renewal =
+ executor.scheduleWithFixedDelay(
+ () -> {
+ if (closed.get()) {
+ return;
+ }
+ heartbeat();
+ },
+ interval,
+ interval,
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void heartbeat() {
+ if (closed.get() || failure != null) {
+ return;
+ }
+ try {
+ ensureValid();
+ long requestedAt = nanoTime.getAsLong();
+ clients.execute(client -> client.heartbeat(0L, lockId));
+ renewedAt.accumulateAndGet(requestedAt, Math::max);
+ } catch (Exception e) {
+ failure = e;
+ }
+ }
+
+ private void ensureValid() {
+ if (TimeUnit.NANOSECONDS.toMillis(nanoTime.getAsLong() - renewedAt.get()) >= leaseMillis
+ && failure == null) {
+ failure = new IllegalStateException("Hive lock expired locally.");
+ }
+ if (closed.get() || failure != null) {
+ throw new IllegalStateException("The Hive lock is no longer usable.", failure);
+ }
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ renewal.cancel(true);
+ if (renewer == null) {
+ executor.shutdownNow();
+ }
+ }
+ }
+
+ public static long leaseTimeout(HiveConf conf) {
+ return HiveConf.getTimeVar(conf, HiveConf.ConfVars.HIVE_TXN_TIMEOUT, TimeUnit.MILLISECONDS);
+ }
+
private long lock(String database, String table)
throws UnknownHostException, TException, InterruptedException {
final LockComponent lockComponent =
@@ -107,7 +256,7 @@ public class HiveCatalogLock implements CatalogLock {
} finally {
if (lockResponse.getState() != LockState.ACQUIRED) {
// unlock if not acquired
- unlock(lockId);
+ Lock.release(() -> unlock(lockId), null);
}
}
@@ -125,14 +274,9 @@ public class HiveCatalogLock implements CatalogLock {
throw new RuntimeException("Acquire lock failed " + msg);
}
- private void unlock(long lockId) {
- if (lockId <= 0) {
- return;
- }
- try {
+ private void unlock(long lockId) throws TException, InterruptedException {
+ if (lockId > 0) {
clients.execute(client -> client.unlock(lockId));
- } catch (Exception e) {
- LOG.warn("Unlock failed for lockId={}", lockId, e);
}
}
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockFactory.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockFactory.java
index 764308a31b..2f5e20c05e 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockFactory.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockFactory.java
@@ -18,12 +18,16 @@
package org.apache.paimon.hive;
-import org.apache.paimon.catalog.CatalogLock;
import org.apache.paimon.catalog.CatalogLockContext;
import org.apache.paimon.catalog.CatalogLockFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
import org.apache.hadoop.hive.conf.HiveConf;
+import javax.annotation.Nullable;
+
import static org.apache.paimon.hive.HiveCatalogLock.LOCK_IDENTIFIER;
import static org.apache.paimon.hive.HiveCatalogLock.acquireTimeout;
import static org.apache.paimon.hive.HiveCatalogLock.checkMaxSleep;
@@ -36,14 +40,21 @@ public class HiveCatalogLockFactory implements CatalogLockFactory {
private static final long serialVersionUID = 1L;
@Override
- public CatalogLock createLock(CatalogLockContext context) {
+ public Lock createLock(
+ CatalogLockContext context,
+ Identifier identifier,
+ @Nullable String tableUuid,
+ @Nullable String commitUser,
+ Options tableOptions) {
checkArgument(context instanceof HiveCatalogLockContext);
HiveCatalogLockContext hiveLockContext = (HiveCatalogLockContext) context;
HiveConf conf = hiveLockContext.hiveConf().conf();
return new HiveCatalogLock(
createClients(conf, hiveLockContext.options(), hiveLockContext.clientClassName()),
+ identifier,
checkMaxSleep(conf),
- acquireTimeout(conf));
+ acquireTimeout(conf),
+ HiveCatalogLock.leaseTimeout(conf));
}
@Override
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogOptions.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogOptions.java
index 9fc16b5946..ba7c434057 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogOptions.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogOptions.java
@@ -22,6 +22,7 @@ import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.ConfigOptions;
import org.apache.paimon.options.description.Description;
+import java.time.Duration;
import java.util.concurrent.TimeUnit;
import static org.apache.paimon.options.description.TextElement.text;
@@ -29,6 +30,32 @@ import static org.apache.paimon.options.description.TextElement.text;
/** Options for hive catalog. */
public final class HiveCatalogOptions {
+ public static final ConfigOption<Boolean> LOCK_ENABLED =
+ ConfigOptions.key("lock.enabled")
+ .booleanType()
+ .noDefaultValue()
+ .withDescription(
+ "Enable catalog locking. Defaults to enabled on object stores.");
+
+ public static final ConfigOption<String> LOCK_TYPE =
+ ConfigOptions.key("lock.type")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The catalog lock factory identifier. Defaults to the catalog's built-in lock factory.");
+
+ public static final ConfigOption<Duration> LOCK_CHECK_MAX_SLEEP =
+ ConfigOptions.key("lock-check-max-sleep")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(8))
+ .withDescription("The maximum sleep time when retrying to check the lock.");
+
+ public static final ConfigOption<Duration> LOCK_ACQUIRE_TIMEOUT =
+ ConfigOptions.key("lock-acquire-timeout")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(8))
+ .withDescription("The maximum time to wait for acquiring the lock.");
+
public static final String IDENTIFIER = "hive";
public static final ConfigOption<String> HIVE_CONF_DIR =
diff --git a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogLockConfigurationTest.java b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogLockConfigurationTest.java
new file mode 100644
index 0000000000..13f7afbe2a
--- /dev/null
+++ b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogLockConfigurationTest.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.hive;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogLockFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.TestCatalogLockFactory;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests Hive policy and factory context without connecting to a metastore. */
+class HiveCatalogLockConfigurationTest {
+ @ParameterizedTest
+ @CsvSource({"false,,false", "true,,true", "false,true,true", "true,false,false"})
+ void hivePreservesStorageDefaultAndExplicitOverride(
+ boolean objectStore, Boolean enabled, boolean expected) throws Exception {
+ Options options = new Options();
+ if (enabled != null) {
+ options.set(HiveCatalogOptions.LOCK_ENABLED, enabled);
+ }
+ try (HiveCatalog catalog = catalog(options, objectStore);
+ Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"),
+ "table-id",
+ "writer",
+ new Options())) {
+ assertThat(lock.runWithLock(() -> options.getString("test.lock.held", "false")))
+ .isEqualTo(Boolean.toString(expected));
+ }
+ assertThat(options.getString("test.lock.held", "false")).isEqualTo("false");
+ assertThat(options.getString("test.lock.closed", "false"))
+ .isEqualTo(Boolean.toString(expected));
+ }
+
+ @Test
+ void configuredFactoryOverridesHiveDefault() throws Exception {
+ Options options = new Options();
+ options.set(HiveCatalogOptions.LOCK_ENABLED, true);
+ options.set(HiveCatalogOptions.LOCK_TYPE, TestCatalogLockFactory.IDENTIFIER);
+ try (HiveCatalog catalog = catalog(options, false);
+ Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"), null, "writer", new Options())) {
+ assertThat(lock.runWithLock(() -> options.get("test.lock.held"))).isEqualTo("true");
+ assertThat(
+ catalog.runWithLock(
+ Identifier.create("db", "table"),
+ () -> options.get("test.lock.held")))
+ .isEqualTo("true");
+ }
+ assertThat(options.get("test.lock.held")).isEqualTo("false");
+ }
+
+ @Test
+ void disabledHiveLockDoesNotResolveFactory() throws Exception {
+ Options options = new Options();
+ options.set(HiveCatalogOptions.LOCK_ENABLED, false);
+ options.set(HiveCatalogOptions.LOCK_TYPE, "not-installed");
+ try (HiveCatalog catalog = catalog(options, true);
+ Lock lock =
+ catalog.createLock(
+ Identifier.create("db", "table"), null, "writer", new Options())) {
+ assertThat(lock.runWithLock(() -> "published")).isEqualTo("published");
+ }
+ }
+
+ @Test
+ void durationOptionsConfigureHiveLocks() {
+ HiveConf conf = new HiveConf();
+ conf.set(HiveCatalogOptions.LOCK_CHECK_MAX_SLEEP.key(), "17 ms");
+ conf.set(HiveCatalogOptions.LOCK_ACQUIRE_TIMEOUT.key(), "3 s");
+ assertThat(HiveCatalogLock.checkMaxSleep(conf)).isEqualTo(17);
+ assertThat(HiveCatalogLock.acquireTimeout(conf)).isEqualTo(3000);
+ HiveConf defaults = new HiveConf();
+ assertThat(HiveCatalogLock.checkMaxSleep(defaults)).isEqualTo(8000);
+ assertThat(HiveCatalogLock.acquireTimeout(defaults)).isEqualTo(480000);
+ }
+
+ private HiveCatalog catalog(Options options, boolean objectStore) {
+ FileIO fileIO = mock(FileIO.class);
+ when(fileIO.isObjectStore()).thenReturn(objectStore);
+ return new HiveCatalog(
+ fileIO,
+ new HiveConf(),
+ "unused",
+ CatalogContext.create(options),
+ "file:/warehouse") {
+ @Override
+ public Optional<CatalogLockFactory> defaultLockFactory() {
+ if (options.contains(HiveCatalogOptions.LOCK_TYPE)) {
+ throw new AssertionError("Explicit factory must override the default.");
+ }
+ return Optional.of(new TestCatalogLockFactory());
+ }
+ };
+ }
+}
diff --git a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogLockTest.java b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogLockTest.java
new file mode 100644
index 0000000000..5e5a631322
--- /dev/null
+++ b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogLockTest.java
@@ -0,0 +1,350 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.hive;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.client.ClientPool;
+import org.apache.paimon.operation.Lock;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.LockResponse;
+import org.apache.hadoop.hive.metastore.api.LockState;
+import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests heartbeat lifetime and expiry of Hive catalog lock scopes. */
+class HiveCatalogLockTest {
+ private final IMetaStoreClient client = mock(IMetaStoreClient.class);
+ private final ScheduledExecutorService renewer = mock(ScheduledExecutorService.class);
+ private final ScheduledFuture<?> future = mock(ScheduledFuture.class);
+ private final AtomicLong clock = new AtomicLong();
+ private final ClientPool<IMetaStoreClient, TException> pool =
+ new ClientPool<IMetaStoreClient, TException>() {
+ @Override
+ public <R> R run(Action<R, IMetaStoreClient, TException> action) throws TException {
+ return action.run(client);
+ }
+
+ @Override
+ public void execute(ExecuteAction<IMetaStoreClient, TException> action)
+ throws TException {
+ action.run(client);
+ }
+ };
+
+ @Test
+ void failedInitialHeartbeatReleasesAcquiredLockAndPreservesCause() throws Exception {
+ HiveCatalogLock lock = lock();
+ TException failure = new TException("Initial heartbeat failed");
+ TException releaseFailure = new TException("Release failed");
+ doThrow(failure).when(client).heartbeat(0L, 7L);
+ doThrow(releaseFailure).when(client).unlock(7L);
+ AtomicInteger operations = new AtomicInteger();
+ assertThatThrownBy(() -> lock.runWithLock(operations::incrementAndGet))
+ .isInstanceOf(Lock.AcquisitionException.class)
+ .hasCause(failure);
+ assertThat(failure.getSuppressed()).containsExactly(releaseFailure);
+ assertThat(operations.get()).isZero();
+ verify(client).unlock(7L);
+ assertThatThrownBy(lock::ensureValid).hasMessageContaining("No Hive lock");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void releaseFailurePreservesOperationOutcome(boolean failOperation) throws Exception {
+ HiveCatalogLock lock = lock();
+ TException releaseFailure = new TException("Release failed");
+ RuntimeException operationFailure = new RuntimeException("Preparation failed");
+ doThrow(releaseFailure).when(client).unlock(7L);
+ java.util.concurrent.Callable<String> operation =
+ () -> {
+ if (failOperation) {
+ throw operationFailure;
+ }
+ return "published";
+ };
+ if (failOperation) {
+ assertThatThrownBy(() -> lock.runWithLock(operation))
+ .isSameAs(operationFailure)
+ .hasSuppressedException(releaseFailure);
+ } else {
+ assertThat(lock.runWithLock(operation)).isEqualTo("published");
+ }
+ verify(client).unlock(7L);
+ verify(future).cancel(true);
+ assertThatThrownBy(lock::ensureValid).hasMessageContaining("No Hive lock");
+ }
+
+ @Test
+ void heartbeatExtendsLeaseAndStopsAfterScope() throws Exception {
+ HiveCatalogLock lock = lock();
+ Runnable renewal =
+ lock.runWithLock(
+ () -> {
+ clock.set(TimeUnit.MILLISECONDS.toNanos(2000));
+ Runnable task = renewal();
+ task.run();
+ clock.set(TimeUnit.MILLISECONDS.toNanos(4000));
+ lock.ensureValid();
+ return task;
+ });
+ renewal.run();
+ verify(client, times(3)).heartbeat(0L, 7L);
+ verify(client).unlock(7L);
+ verify(future).cancel(true);
+ }
+
+ @Test
+ void heartbeatFailureInvalidatesScopeAndPreservesCause() throws Exception {
+ HiveCatalogLock lock = lock();
+ TException failure = new TException("Lost lock");
+ lock.runWithLock(
+ () -> {
+ doThrow(failure).when(client).heartbeat(0L, 7L);
+ renewal().run();
+ assertThatThrownBy(lock::ensureValid).hasCause(failure);
+ return null;
+ });
+ verify(client).unlock(7L);
+ }
+
+ @Test
+ void expiryWithoutHeartbeatPreventsFurtherUse() throws Exception {
+ HiveCatalogLock lock = lock();
+ lock.runWithLock(
+ () -> {
+ clock.set(TimeUnit.MILLISECONDS.toNanos(3000));
+ assertThatThrownBy(lock::ensureValid).hasMessageContaining("no longer usable");
+ renewal().run();
+ return null;
+ });
+ verify(client).heartbeat(0L, 7L);
+ verify(client).unlock(7L);
+ assertThatThrownBy(lock::ensureValid).hasMessageContaining("No Hive lock");
+ }
+
+ @Test
+ void preparationFailureReleasesScope() throws Exception {
+ HiveCatalogLock lock = lock();
+ RuntimeException failure = new RuntimeException("Preparation failed");
+ assertThatThrownBy(
+ () ->
+ lock.runWithLock(
+ () -> {
+ throw failure;
+ }))
+ .isSameAs(failure);
+ verify(client).unlock(7L);
+ verify(future).cancel(true);
+ }
+
+ @Test
+ void publicationChecksTheServerEvenWhenClientTimeoutIsLonger() throws Exception {
+ HiveCatalogLock lock = lock();
+ lock.runWithLock(
+ () -> {
+ NoSuchLockException missing = new NoSuchLockException("Expired on server");
+ doThrow(missing).when(client).heartbeat(0L, 7L);
+ // No scheduled renewal or local expiry has occurred yet.
+ assertThatThrownBy(lock::ensureValid).hasCause(missing);
+ return null;
+ });
+ verify(client).unlock(7L);
+ }
+
+ @Test
+ void olderHeartbeatResponseCannotShortenANewerDeadline() throws Exception {
+ HiveCatalogLock lock = lock();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CountDownLatch oldStarted = new CountDownLatch(1);
+ CountDownLatch oldResponse = new CountDownLatch(1);
+ AtomicInteger requests = new AtomicInteger();
+ try {
+ lock.runWithLock(
+ () -> {
+ doAnswer(
+ invocation -> {
+ if (requests.incrementAndGet() == 1) {
+ oldStarted.countDown();
+ assertThat(oldResponse.await(10, TimeUnit.SECONDS))
+ .isTrue();
+ }
+ return null;
+ })
+ .when(client)
+ .heartbeat(0L, 7L);
+ clock.set(TimeUnit.MILLISECONDS.toNanos(1000));
+ Future<?> old = executor.submit(renewal());
+ assertThat(oldStarted.await(10, TimeUnit.SECONDS)).isTrue();
+ clock.set(TimeUnit.MILLISECONDS.toNanos(2000));
+ lock.ensureValid();
+ oldResponse.countDown();
+ old.get(10, TimeUnit.SECONDS);
+ clock.set(TimeUnit.MILLISECONDS.toNanos(4000));
+ lock.ensureValid();
+ return null;
+ });
+ } finally {
+ oldResponse.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void blockedRenewalsDoNotPreventOtherCatalogsFromRenewing() throws Exception {
+ Set<Thread> renewalThreads = ConcurrentHashMap.newKeySet();
+ CountDownLatch blocked = new CountDownLatch(2);
+ CountDownLatch renewed = new CountDownLatch(4);
+ CountDownLatch releaseScopes = new CountDownLatch(1);
+ CountDownLatch releaseRenewals = new CountDownLatch(1);
+ ExecutorService writers = Executors.newFixedThreadPool(2);
+ List<Future<?>> scopes = new ArrayList<>();
+ try {
+ for (int i = 0; i < 2; i++) {
+ IMetaStoreClient slow = mock(IMetaStoreClient.class);
+ when(slow.lock(any())).thenReturn(new LockResponse(7L, LockState.ACQUIRED));
+ doAnswer(
+ invocation -> {
+ if (Thread.currentThread()
+ .getName()
+ .startsWith("hive-lock-renew")) {
+ renewalThreads.add(Thread.currentThread());
+ blocked.countDown();
+ assertThat(releaseRenewals.await(30, TimeUnit.SECONDS))
+ .isTrue();
+ }
+ return null;
+ })
+ .when(slow)
+ .heartbeat(0L, 7L);
+ HiveCatalogLock lock =
+ new HiveCatalogLock(
+ pool(slow), Identifier.create("db", "slow" + i), 10, 3000, 3000);
+ scopes.add(
+ writers.submit(
+ () ->
+ lock.runWithLock(
+ () -> {
+ assertThat(
+ releaseScopes.await(
+ 30, TimeUnit.SECONDS))
+ .isTrue();
+ return null;
+ })));
+ }
+ assertThat(blocked.await(10, TimeUnit.SECONDS)).isTrue();
+ when(client.lock(any())).thenReturn(new LockResponse(7L, LockState.ACQUIRED));
+ doAnswer(
+ invocation -> {
+ if (Thread.currentThread()
+ .getName()
+ .startsWith("hive-lock-renew")) {
+ renewalThreads.add(Thread.currentThread());
+ renewed.countDown();
+ }
+ return null;
+ })
+ .when(client)
+ .heartbeat(0L, 7L);
+ HiveCatalogLock healthy =
+ new HiveCatalogLock(pool, Identifier.create("db", "healthy"), 10, 3000, 3000);
+ assertThat(
+ healthy.runWithLock(
+ () -> {
+ assertThat(renewed.await(10, TimeUnit.SECONDS)).isTrue();
+ healthy.ensureValid();
+ return "published";
+ }))
+ .isEqualTo("published");
+ verify(client).unlock(7L);
+ } finally {
+ releaseScopes.countDown();
+ releaseRenewals.countDown();
+ try {
+ for (Future<?> scope : scopes) {
+ scope.get(10, TimeUnit.SECONDS);
+ }
+ } finally {
+ writers.shutdownNow();
+ }
+ }
+ for (Thread thread : renewalThreads) {
+ thread.join(10000);
+ assertThat(thread.isAlive()).isFalse();
+ }
+ }
+
+ private ClientPool<IMetaStoreClient, TException> pool(IMetaStoreClient client) {
+ return new ClientPool<IMetaStoreClient, TException>() {
+ @Override
+ public <R> R run(Action<R, IMetaStoreClient, TException> action) throws TException {
+ return action.run(client);
+ }
+
+ @Override
+ public void execute(ExecuteAction<IMetaStoreClient, TException> action)
+ throws TException {
+ action.run(client);
+ }
+ };
+ }
+
+ private HiveCatalogLock lock() throws Exception {
+ when(client.lock(any())).thenReturn(new LockResponse(7L, LockState.ACQUIRED));
+ when(renewer.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any()))
+ .thenAnswer(invocation -> future);
+ return new HiveCatalogLock(
+ pool, Identifier.create("db", "table"), 10, 3000, 3000, renewer, clock::get);
+ }
+
+ private Runnable renewal() {
+ ArgumentCaptor<Runnable> task = ArgumentCaptor.forClass(Runnable.class);
+ verify(renewer).scheduleWithFixedDelay(task.capture(), anyLong(), anyLong(), any());
+ return task.getValue();
+ }
+}
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
index 7276d9a251..279159eb48 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
@@ -1022,11 +1022,14 @@ public abstract class HiveCatalogITCaseBase {
new Thread(
() -> {
Lock lock =
- Lock.fromCatalog(
- catalogEnv
- .lockFactory()
- .createLock(catalogEnv.lockContext()),
- identifier);
+ catalogEnv
+ .lockFactory()
+ .createLock(
+ catalogEnv.lockContext(),
+ identifier,
+ null,
+ null,
+ new Options());
for (int j = 0; j < 10; j++) {
try {
lock.runWithLock(unsafeIncrement);
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveLocationTest.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveLocationTest.java
index 78a0333a51..f7f47f308c 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveLocationTest.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveLocationTest.java
@@ -87,7 +87,7 @@ public class HiveLocationTest {
options.set(CatalogOptions.WAREHOUSE, objectStorePath);
options.set(CatalogOptions.METASTORE, "hive");
options.set(CatalogOptions.URI, "");
- options.set(CatalogOptions.LOCK_ENABLED, false);
+ options.set(HiveCatalogOptions.LOCK_ENABLED, false);
options.set(HiveCatalogOptions.HIVE_CONF_DIR, hiveShell.getBaseDir() + HIVE_CONF);
options.set(HiveCatalogOptions.LOCATION_IN_PROPERTIES, true);