This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 75678093b3eb fix(lock): fix lock config derivation for Spark procedure
(#19794)
75678093b3eb is described below
commit 75678093b3eb6af46b8dcfa1c1e41ccc384a9afd
Author: Joy <[email protected]>
AuthorDate: Tue Sep 8 16:20:32 2026 +0800
fix(lock): fix lock config derivation for Spark procedure (#19794)
* fix(lock): unify default FS lock path and make CLI lock config take effect
Two related defects in the metadata-table auto-lock path:
1. Lock-path inconsistency: FileSystemBasedLockProvider.getLockConfig used
the .aux folder as the default lock path, while the constructor fell back
to the .hoodie meta folder. The two defaults could diverge, silently
placing the lock file at different paths and breaking mutual exclusion
across engines/tasks. Unify both on the table metadata path (.hoodie) by
reusing a single defaultLockPath() helper.
2. Lock config never applied: RunClustering/RunCompaction appended lock
options to `confs` AFTER the write client was already built, so they had
no effect. Move the auto-lock injection down into
HoodieCLIUtils.createHoodieWriteClient, applied after the final
parameters
are merged and before the client is built, guarded on the fully-merged
params so any explicitly-configured lock provider is respected. This also
extends the coverage from just clustering/compaction to all write
procedures that go through createHoodieWriteClient.
Remove the now-dead lock-injection blocks and unused imports from the two
procedures. Add/extend tests covering the unified path and getLockOptions.
* fix(lock): keep metadata lock derivation in clustering and compaction
---------
Co-authored-by: jiangyu84 <[email protected]>
Co-authored-by: danny0405 <[email protected]>
---
.../lock/FileSystemBasedLockProvider.java | 5 +-
.../lock/TestFileSystemBasedLockProvider.java | 38 +++++++++--
.../scala/org/apache/hudi/HoodieCLIUtils.scala | 34 +++++++---
.../scala/org/apache/hudi/TestHoodieCLIUtils.scala | 72 +++++++++++++++++++++
.../procedures/RunClusteringProcedure.scala | 13 ++--
.../procedures/RunCompactionProcedure.scala | 14 ++--
.../procedure/TestTableServiceLockConfig.scala | 74 ++++++++++++++++++++++
7 files changed, 222 insertions(+), 28 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
index 0988cf4a7054..da3279e32013 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
@@ -24,7 +24,6 @@ import org.apache.hudi.common.config.LockConfiguration;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.lock.LockProvider;
import org.apache.hudi.common.lock.LockState;
-import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
@@ -115,8 +114,8 @@ public class FileSystemBasedLockProvider implements
LockProvider<String>, Serial
this.lockConfiguration = lockConfiguration;
String lockDirectory =
lockConfiguration.getConfig().getString(FILESYSTEM_LOCK_PATH_PROP_KEY, null);
if (StringUtils.isNullOrEmpty(lockDirectory)) {
- lockDirectory =
lockConfiguration.getConfig().getString(HoodieWriteConfig.BASE_PATH.key())
- + StoragePath.SEPARATOR + HoodieTableMetaClient.METAFOLDER_NAME;
+ // Match getLockConfig() so writers using either default share the same
lock.
+ lockDirectory =
defaultLockPath(lockConfiguration.getConfig().getString(HoodieWriteConfig.BASE_PATH.key()));
}
this.lockTimeoutMinutes =
lockConfiguration.getConfig().getInteger(FILESYSTEM_LOCK_EXPIRE_PROP_KEY);
this.lockFile = new StoragePath(lockDirectory + StoragePath.SEPARATOR +
LOCK_FILE_NAME);
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestFileSystemBasedLockProvider.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestFileSystemBasedLockProvider.java
index e2ca7a8a02cd..1fd1eaa69071 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestFileSystemBasedLockProvider.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestFileSystemBasedLockProvider.java
@@ -433,8 +433,8 @@ public class TestFileSystemBasedLockProvider {
public void testGetLockConfigProducesUsableProperties() {
String tablePath = tempDir.resolve("table").toString();
TypedProperties props =
FileSystemBasedLockProvider.getLockConfig(tablePath);
- // The generated config points the lock provider at the table's auxiliary
folder.
-
assertTrue(props.getString(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key()).startsWith(tablePath));
+ assertEquals(tablePath + StoragePath.SEPARATOR +
HoodieTableMetaClient.AUXILIARYFOLDER_NAME,
+ props.getString(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key()));
assertEquals(FileSystemBasedLockProvider.class.getName(),
props.getString(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key()));
@@ -450,7 +450,35 @@ public class TestFileSystemBasedLockProvider {
}
@Test
- public void testLockPathDefaultsToMetafolderFromBasePath() {
+ public void testExplicitLockConfigAndBasePathFallbackResolveToSamePath() {
+ // Regression guard for the cross-engine lock-path divergence:
getLockConfig()'s explicit default
+ // and the constructor's BASE_PATH-derived fallback must point at the very
same lock file, or a
+ // writer relying on one would not be mutually excluded from a writer
relying on the other.
+ StorageConfiguration<?> storageConf =
HoodieTestUtils.getDefaultStorageConf();
+ String tablePath = tempDir.resolve("sharedtable").toString();
+
+ // 1) Provider built from getLockConfig() (explicit FILESYSTEM_LOCK_PATH).
+ FileSystemBasedLockProvider fromLockConfig =
+ new FileSystemBasedLockProvider(
+ new
LockConfiguration(FileSystemBasedLockProvider.getLockConfig(tablePath)),
storageConf);
+
+ // 2) Provider built with only BASE_PATH set, exercising the constructor
fallback.
+ Properties fallbackProps = new Properties();
+ fallbackProps.setProperty(HoodieWriteConfig.BASE_PATH.key(), tablePath);
+ fallbackProps.setProperty(FILESYSTEM_LOCK_EXPIRE_PROP_KEY, "0");
+ FileSystemBasedLockProvider fromFallback =
+ new FileSystemBasedLockProvider(new LockConfiguration(fallbackProps),
storageConf);
+
+ // getLock() is a pure accessor of the resolved lock-file path (no storage
I/O), so comparing it
+ // is enough and needs no acquire/release.
+ assertEquals(fromLockConfig.getLock(), fromFallback.getLock(),
+ "explicit default and BASE_PATH fallback must resolve to the same lock
file");
+ assertTrue(fromLockConfig.getLock()
+ .endsWith(HoodieTableMetaClient.AUXILIARYFOLDER_NAME +
StoragePath.SEPARATOR + "lock"));
+ }
+
+ @Test
+ public void testLockPathDefaultsToAuxiliaryFolderFromBasePath() {
StorageConfiguration<?> storageConf =
HoodieTestUtils.getDefaultStorageConf();
Properties props = new Properties();
props.setProperty(HoodieWriteConfig.BASE_PATH.key(),
lockDir("defaultpath"));
@@ -460,9 +488,9 @@ public class TestFileSystemBasedLockProvider {
try {
assertTrue(provider.tryLock(1, TimeUnit.SECONDS),
"lock acquisition must work without an explicit lock path");
- // Without an explicit lock path the provider locks under the table
metafolder.
+ // Without an explicit lock path the provider locks under the table
auxiliary folder.
assertTrue(provider.getLock().endsWith(
- HoodieTableMetaClient.METAFOLDER_NAME + StoragePath.SEPARATOR +
"lock"));
+ HoodieTableMetaClient.AUXILIARYFOLDER_NAME + StoragePath.SEPARATOR +
"lock"));
} finally {
provider.unlock();
provider.close();
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
index c6e37f5bb6dd..c46a0595899d 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala
@@ -37,7 +37,7 @@ import
org.apache.spark.sql.catalyst.catalog.HoodieCatalogTable
import org.apache.spark.sql.hudi.HoodieOptionConfig
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.filterHoodieConfigs
-import java.util.ArrayList
+import java.util.Collections
import scala.collection.JavaConverters.{collectionAsScalaIterableConverter,
mapAsJavaMapConverter, propertiesAsScalaMapConverter}
@@ -52,6 +52,16 @@ object HoodieCLIUtils extends Logging {
val schemaUtil = new TableSchemaResolver(metaClient)
val schemaStr = schemaUtil.getTableSchema(false).toString
+ val finalParameters = getWriteParameters(sparkSession, metaClient, conf,
tableName)
+ val jsc = new JavaSparkContext(sparkSession.sparkContext)
+ DataSourceUtils.createHoodieClient(jsc, schemaStr, basePath,
+ metaClient.getTableConfig.getTableName, finalParameters.asJava)
+ }
+
+ def getWriteParameters(sparkSession: SparkSession,
+ metaClient: HoodieTableMetaClient,
+ conf: Map[String, String],
+ tableName: Option[String]): Map[String, String] = {
// If tableName is provided, we need to add catalog props
val catalogProps = tableName match {
case Some(value) =>
HoodieOptionConfig.mapSqlOptionsToDataSourceWriteConfigs(
@@ -60,16 +70,12 @@ object HoodieCLIUtils extends Logging {
}
// Priority: defaults < catalog props < table config < sparkSession conf <
specified conf
- val finalParameters = HoodieWriterUtils.parametersWithWriteDefaults(
+ HoodieWriterUtils.parametersWithWriteDefaults(
(catalogProps ++
metaClient.getTableConfig.getProps.asScala.toMap ++
filterHoodieConfigs(sparkSession.sqlContext.getAllConfs) ++
conf).toMap
)
-
- val jsc = new JavaSparkContext(sparkSession.sparkContext)
- DataSourceUtils.createHoodieClient(jsc, schemaStr, basePath,
- metaClient.getTableConfig.getTableName, finalParameters.asJava)
}
def extractPartitions(clusteringGroups: Seq[HoodieClusteringGroup]): String
= {
@@ -166,9 +172,19 @@ object HoodieCLIUtils extends Logging {
key -> value
}
- def getLockOptions(tablePath: String, schema: String, lockConfig:
TypedProperties): Map[String, String] = {
- val customSupportedFSs =
lockConfig.getStringList(HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key,
",", new ArrayList[String])
- if (schema == null || customSupportedFSs.contains(schema) ||
StorageSchemes.isAtomicCreationSupported(schema)) {
+ /**
+ * Builds the filesystem-based lock configuration for the metadata table, or
an empty map when the
+ * table's filesystem cannot support atomic file creation (a hard
requirement for the FS lock).
+ *
+ * @param tablePath the table base path, used to derive the shared lock file
location
+ * @param scheme the table filesystem scheme; {@code null} is treated as
supported
+ * @param params the already-merged write parameters, read only for the
custom
+ * atomic-creation-support list ({@code
hoodie.fs.atomic_creation.support})
+ */
+ def getLockOptions(tablePath: String, scheme: String, params: Map[String,
String]): Map[String, String] = {
+ val customSupportedFSs = TypedProperties.fromMap(params.asJava)
+ .getStringList(HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key,
",", Collections.emptyList[String]())
+ if (scheme == null || customSupportedFSs.contains(scheme) ||
StorageSchemes.isAtomicCreationSupported(scheme)) {
logInfo("Auto config filesystem lock provider for metadata table")
val props = FileSystemBasedLockProvider.getLockConfig(tablePath)
props.stringPropertyNames.asScala
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestHoodieCLIUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestHoodieCLIUtils.scala
index ca4869286c26..2818a0bce190 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestHoodieCLIUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestHoodieCLIUtils.scala
@@ -19,8 +19,15 @@
package org.apache.hudi
+import org.apache.hudi.client.transaction.lock.FileSystemBasedLockProvider
+import org.apache.hudi.common.config.{HoodieCommonConfig, TypedProperties}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import org.apache.hudi.config.HoodieLockConfig
+
+import org.apache.spark.sql.{SparkSession, SQLContext}
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows,
assertTrue}
import org.junit.jupiter.api.Test
+import org.mockito.Mockito.{mock, when}
class TestHoodieCLIUtils {
@@ -101,4 +108,69 @@ class TestHoodieCLIUtils {
classOf[IllegalArgumentException],
() => HoodieCLIUtils.extractOptions(" =v"))
}
+
+ @Test
+ def testGetLockOptionsSupportedSchemeReturnsFsLockConfig(): Unit = {
+ val tablePath = "/tmp/hudi/some_table"
+ // A null scheme is treated as supported; the FS lock provider must be
auto-configured.
+ val opts = HoodieCLIUtils.getLockOptions(tablePath, null, Map.empty)
+ assertEquals(classOf[FileSystemBasedLockProvider].getName,
+ opts(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key))
+ assertEquals(s"$tablePath/${HoodieTableMetaClient.AUXILIARYFOLDER_NAME}",
+ opts(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key))
+ }
+
+ @Test
+ def testGetLockOptionsUnsupportedSchemeReturnsEmpty(): Unit = {
+ // s3 is a known scheme without atomic-creation support, so no FS lock can
be configured.
+ assertTrue(HoodieCLIUtils.getLockOptions("s3://bucket/table", "s3",
Map.empty).isEmpty)
+ }
+
+ @Test
+ def testGetLockOptionsCustomAtomicSupportEnablesScheme(): Unit = {
+ // Opting s3 into hoodie.fs.atomic_creation.support makes the FS lock
provider eligible again.
+ val params = Map(HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key
-> " hdfs, s3 ")
+ val opts = HoodieCLIUtils.getLockOptions("s3://bucket/table", "s3", params)
+ assertEquals(classOf[FileSystemBasedLockProvider].getName,
+ opts(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key))
+ }
+
+ @Test
+ def testWriteParametersPreserveLockConfigPrecedence(): Unit = {
+ val sparkSession = mock(classOf[SparkSession])
+ val sqlContext = mock(classOf[SQLContext])
+ val metaClient = mock(classOf[HoodieTableMetaClient])
+ val tableConfig = mock(classOf[HoodieTableConfig])
+ val tableProps = new TypedProperties()
+ val providerKey = HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key
+ val atomicSupportKey =
HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key
+ tableProps.setProperty(providerKey, "table.provider")
+ tableProps.setProperty(atomicSupportKey, "hdfs")
+ when(sparkSession.sqlContext).thenReturn(sqlContext)
+ when(metaClient.getTableConfig).thenReturn(tableConfig)
+ when(tableConfig.getProps).thenReturn(tableProps)
+ when(sqlContext.getAllConfs).thenReturn(Map.empty[String, String])
+
+ val fromTable = HoodieCLIUtils.getWriteParameters(sparkSession,
metaClient, Map.empty, None)
+ assertEquals("table.provider", fromTable(providerKey))
+ assertEquals("hdfs", fromTable(atomicSupportKey))
+
+ when(sqlContext.getAllConfs).thenReturn(Map(providerKey ->
"session.provider", atomicSupportKey -> "s3"))
+ val fromSession = HoodieCLIUtils.getWriteParameters(sparkSession,
metaClient, Map.empty, None)
+ assertEquals("session.provider", fromSession(providerKey))
+ assertEquals("s3", fromSession(atomicSupportKey))
+
+ val fromOptions = HoodieCLIUtils.getWriteParameters(sparkSession,
metaClient,
+ Map(providerKey -> "options.provider", atomicSupportKey -> "file"), None)
+ assertEquals("options.provider", fromOptions(providerKey))
+ assertEquals("file", fromOptions(atomicSupportKey))
+ assertEquals(fromOptions,
+ HoodieCLIUtils.getWriteParameters(sparkSession, metaClient, fromOptions,
None))
+
+ tableProps.clear()
+ when(sqlContext.getAllConfs).thenReturn(Map.empty[String, String])
+ val withoutLock = HoodieCLIUtils.getWriteParameters(sparkSession,
metaClient, Map.empty, None)
+ assertTrue(!withoutLock.contains(providerKey))
+ }
+
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala
index e3248f337ffc..10c4c7b5ad70 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala
@@ -181,6 +181,14 @@ class RunClusteringProcedure extends BaseProcedure
var (filteredPendingClusteringInstants, operation) =
HoodieProcedureUtils.filterPendingInstantsAndGetOperation(
pendingClusteringInstants,
specificInstants.asInstanceOf[Option[String]], op.asInstanceOf[Option[String]],
limit.asInstanceOf[Option[Int]])
+ confs = HoodieCLIUtils.getWriteParameters(sparkSession, metaClient, confs,
+ tableName.asInstanceOf[Option[String]])
+ // Apply the lock options before constructing the write client.
+ if (metaClient.getTableConfig.isMetadataTableAvailable
+ && !confs.contains(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key)) {
+ confs = HoodieCLIUtils.getLockOptions(basePath,
metaClient.getBasePath.toUri.getScheme, confs) ++ confs
+ }
+
var client: SparkRDDWriteClient[_] = null
try {
client = HoodieCLIUtils.createHoodieWriteClient(sparkSession, basePath,
confs,
@@ -197,11 +205,6 @@ class RunClusteringProcedure extends BaseProcedure
SpatialCurveSortPartitionerBase.validateOrderByColumns(orderColumns,
tableSchema, strategy)
}
}
- if (metaClient.getTableConfig.isMetadataTableAvailable) {
- if (!confs.contains(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key)) {
- confs = confs ++ HoodieCLIUtils.getLockOptions(basePath,
metaClient.getBasePath.toUri.getScheme,
client.getConfig.getCommonConfig.getProps())
- }
- }
if (operation.isSchedule) {
val instantTime = client.scheduleClustering(HOption.empty())
instantTime.ifPresent(instant => {
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
index 772d8be1b5b7..f709b85a331b 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
@@ -94,17 +94,19 @@ class RunCompactionProcedure extends BaseProcedure with
ProcedureBuilder with Sp
var (filteredPendingCompactionInstants, operation) =
HoodieProcedureUtils.filterPendingInstantsAndGetOperation(
pendingCompactionInstants,
specificInstants.asInstanceOf[Option[String]], Option(op), limit)
+ confs = HoodieCLIUtils.getWriteParameters(sparkSession, metaClient, confs,
+ tableName.asInstanceOf[Option[String]])
+ // Apply the lock options before constructing the write client.
+ if (metaClient.getTableConfig.isMetadataTableAvailable
+ && !confs.contains(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key)) {
+ confs = HoodieCLIUtils.getLockOptions(basePath,
metaClient.getBasePath.toUri.getScheme, confs) ++ confs
+ }
+
var client: SparkRDDWriteClient[_] = null
try {
client = HoodieCLIUtils.createHoodieWriteClient(sparkSession, basePath,
confs,
tableName.asInstanceOf[Option[String]])
- if (metaClient.getTableConfig.isMetadataTableAvailable) {
- if (!confs.contains(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key)) {
- confs = confs ++ HoodieCLIUtils.getLockOptions(basePath,
metaClient.getBasePath.toUri.getScheme,
client.getConfig.getCommonConfig.getProps())
- }
- }
-
if (operation.isSchedule) {
val instantTime =
client.scheduleCompaction(HOption.empty[java.util.Map[String, String]])
instantTime.ifPresent(instant => {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestTableServiceLockConfig.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestTableServiceLockConfig.scala
new file mode 100644
index 000000000000..ddcabdde533e
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestTableServiceLockConfig.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.spark.sql.hudi.procedure
+
+import org.apache.hudi.HoodieCLIUtils
+import org.apache.hudi.client.transaction.lock.FileSystemBasedLockProvider
+import org.apache.hudi.common.config.LockConfiguration
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.config.HoodieLockConfig
+import org.apache.hudi.exception.HoodieLockException
+
+import java.util.concurrent.TimeUnit
+
+class TestTableServiceLockConfig extends HoodieSparkProcedureTestBase {
+ Seq("run_clustering" -> "cow", "run_compaction" -> "mor").foreach { case
(procedure, tableType) =>
+ test(s"$procedure acquires the derived metadata table lock") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val basePath = tmp.getCanonicalPath
+ spark.sql(
+ s"""create table $tableName (id int, ts long) using hudi
+ |location '$basePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts', type =
'$tableType',
+ | 'hoodie.metadata.enable' = 'true')""".stripMargin)
+ spark.sql(s"insert into $tableName values (1, 1)")
+
assert(HoodieTestUtils.createMetaClient(basePath).getTableConfig.isMetadataTableAvailable)
+
+ val client = HoodieCLIUtils.createHoodieWriteClient(spark, basePath,
Map.empty, Some(tableName))
+ try {
+ assert(client.getConfig.getLockProviderClass !=
classOf[FileSystemBasedLockProvider].getName)
+ } finally {
+ client.close()
+ }
+
+ val lock = new FileSystemBasedLockProvider(
+ new
LockConfiguration(FileSystemBasedLockProvider.getLockConfig(basePath)),
+ HoodieTestUtils.getDefaultStorageConf)
+ try {
+ assert(lock.tryLock(1, TimeUnit.SECONDS))
+ val options = Seq(
+ HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.key -> "1",
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.key -> "0"
+ ).map { case (key, value) => s"$key=$value" }.mkString(",")
+ val error = intercept[Exception] {
+ spark.sql(s"call $procedure(op => 'schedule', table =>
'$tableName', options => '$options')").collect()
+ }
+ assert(Iterator.iterate[Throwable](error)(_.getCause).takeWhile(_ !=
null)
+ .exists(_.isInstanceOf[HoodieLockException]))
+ } finally {
+ lock.unlock()
+ lock.close()
+ }
+ spark.sql(s"call $procedure(op => 'schedule', table =>
'$tableName')").collect()
+ }
+ }
+ }
+}
