hudi-agent commented on code in PR #19794:
URL: https://github.com/apache/hudi/pull/19794#discussion_r3945063250
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala:
##########
@@ -166,14 +170,32 @@ 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)) {
- logInfo("Auto config filesystem lock provider for metadata table")
- val props = FileSystemBasedLockProvider.getLockConfig(tablePath)
- props.stringPropertyNames.asScala
- .map(key => key -> props.getString(key))
- .toMap
+ /**
+ * FS lock options for the metadata table, or an empty map when the MDT is
absent, a lock provider
+ * is already set in {@code params}, or the table filesystem cannot support
atomic file creation
+ * (a hard requirement for the FS lock).
+ *
+ * @param metaClient the table meta client; the lock path and scheme are
derived from its base path
+ * @param params the already-merged write parameters, read for an
existing lock provider and
+ * the custom atomic-creation-support list ({@code
hoodie.fs.atomic_creation.support})
+ */
+ def getLockOptions(metaClient: HoodieTableMetaClient, params: Map[String,
String]): Map[String, String] = {
+ if (metaClient.getTableConfig.isMetadataTableAvailable
+ && !params.contains(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key)) {
+ val tablePath = metaClient.getBasePath
+ val scheme = tablePath.toUri.getScheme
+ val customSupportedFSs =
params.get(HoodieCommonConfig.HOODIE_FS_ATOMIC_CREATION_SUPPORT.key)
Review Comment:
🤖 nit: could we keep using the existing config helper here (e.g. wrap
`params` in a `TypedProperties` and call `getStringList(key, ",", emptyList)`)
instead of hand-rolling the split/trim/filter chain? Keeps the parsing
semantics identical to the rest of the codebase.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ArchiveCommitsProcedure.scala:
##########
@@ -89,13 +89,17 @@ class ArchiveCommitsProcedure extends BaseProcedure
HoodieMetadataConfig.ENABLE.defaultValue().toString)
val basePath = getBasePath(tableName, tablePath)
+ val metaClient = createMetaClient(jsc, basePath)
+ // Archival can run alongside a Flink streaming writer; auto-configure a
DFS lock so the two
+ // stay mutually exclusive on the shared metadata table.
+ val confsWithLock = confs ++ HoodieCLIUtils.getLockOptions(metaClient,
confs)
Review Comment:
🤖 Here `confs` is only the procedure's own options, so unlike the
write-client path (which checks the fully merged params incl. session confs and
hudi-defaults), a lock provider configured at the session level won't be
detected and we'd inject the FS lock anyway. If the concurrent writer uses e.g.
a ZK provider, the two locks aren't mutually exclusive — worth checking the
merged config here too?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCLIUtils.scala:
##########
@@ -60,13 +60,17 @@ object HoodieCLIUtils extends Logging {
}
// Priority: defaults < catalog props < table config < sparkSession conf <
specified conf
- val finalParameters = HoodieWriterUtils.parametersWithWriteDefaults(
+ val parameters = HoodieWriterUtils.parametersWithWriteDefaults(
(catalogProps ++
metaClient.getTableConfig.getProps.asScala.toMap ++
filterHoodieConfigs(sparkSession.sqlContext.getAllConfs) ++
conf).toMap
)
+ val finalParameters =
+ if (shouldDeriveLockConfig) parameters ++ getLockOptions(metaClient,
parameters)
Review Comment:
🤖 Could the derived provider still be dropped downstream when
`hoodie.auto.adjust.lock.configs=true`?
`HoodieWriteConfig.Builder.autoAdjustConfigsForConcurrencyMode` overwrites any
non-InProcess provider back to `InProcessLockProvider` for SINGLE_WRITER with
no async table services — which is exactly the COW clustering/clean case here.
It's off by default, but if a user has it on, this fix would silently no-op
again.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
