This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 32f66f3ce [spark] Overwrite paimon dlf authentication configs by spark
job level properties
32f66f3ce is described below
commit 32f66f3cea181670d98dc8da29c51f8bd0127483
Author: Yang Zhang <[email protected]>
AuthorDate: Wed May 27 13:54:03 2026 +0800
[spark] Overwrite paimon dlf authentication configs by spark job level
properties
---
.../fluss/spark/read/lake/FlussLakeUtils.scala | 26 +++++++++++++++++-----
1 file changed, 21 insertions(+), 5 deletions(-)
diff --git
a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala
b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala
index 62a6b6d98..0bdcb8721 100644
---
a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala
+++
b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/lake/FlussLakeUtils.scala
@@ -29,7 +29,20 @@ import scala.collection.JavaConverters._
object FlussLakeUtils {
- private val SPARK_CATALOG_PREFIX = "spark.sql.catalog."
+ /**
+ * Paimon DLF supports multiple authentication schemes (see
+ * https://paimon.apache.org/docs/1.3/concepts/rest/dlf/), allowing clients
to freely choose the
+ * appropriate method based on their runtime environment. However, the Fluss
server does not
+ * propagate auth configs for all scenarios. These keys are therefore
overridden with the values
+ * from the Catalog configuration, replacing whatever the Fluss server
provides, so that clients
+ * can use their preferred authentication method.
+ */
+ private val REMOVAL_KEYS = Set(
+ "dlf.access-key-id",
+ "dlf.access-key-secret",
+ "dlf.security-token",
+ "dlf.token-path",
+ "dlf.token-loader")
def createLakeSource(
catalogProperties: util.Map[String, String],
@@ -39,11 +52,14 @@ object FlussLakeUtils {
val datalakeFormat = tableConfig.get(ConfigOptions.TABLE_DATALAKE_FORMAT)
val dataLakePrefix = "table.datalake." + datalakeFormat + "."
+ val rewriteProperties =
+ PropertiesUtils.extractAndRemovePrefix(tableProperties,
dataLakePrefix).asScala.filterNot {
+ case (k, _) => REMOVAL_KEYS(k)
+ } ++
+ catalogProperties.asScala.filter { case (k, _) => REMOVAL_KEYS(k) }
+
val lakeConfig =
-
Configuration.fromMap(PropertiesUtils.extractAndRemovePrefix(tableProperties,
dataLakePrefix))
- catalogProperties.asScala.foreach {
- case (k, v) => lakeConfig.setString(s"$SPARK_CATALOG_PREFIX$k", v)
- }
+ Configuration.fromMap(rewriteProperties.asJava)
val lakeStoragePlugin =
LakeStoragePluginSetUp.fromDataLakeFormat(datalakeFormat.toString, null)
val lakeStorage = lakeStoragePlugin.createLakeStorage(lakeConfig)
