You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/05/27 05:54:08 UTC
(fluss) branch main updated: [spark] Overwrite paimon dlf authentication configs by spark job level properties
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 <ya...@alibaba-inc.com>
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)