You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by re...@apache.org on 2024/12/04 01:28:51 UTC
(celeborn) branch main updated: [CELEBORN-1748] Deprecate identity provider configs tied with quota
This is an automated email from the ASF dual-hosted git repository.
rexxiong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 782393af0 [CELEBORN-1748] Deprecate identity provider configs tied with quota
782393af0 is described below
commit 782393af05a6567505e7be541ab91a9817c29c66
Author: Sanskar Modi <sa...@gmail.com>
AuthorDate: Wed Dec 4 09:28:40 2024 +0800
[CELEBORN-1748] Deprecate identity provider configs tied with quota
### What changes were proposed in this pull request?
Deprecate identity configs related with quota –
```
"celeborn.quota.identity.provider"
"celeborn.quota.identity.user-specific.tenant"
"celeborn.quota.identity.user-specific.userName"
```
In favour of identity configs independent of quota
```
"celeborn.identity.provider"
"celeborn.identity.user-specific.tenant"
"celeborn.identity.user-specific.userName"
```
### Why are the changes needed?
Current identity configs are tied with quota but identity should be free of quota because other pieces like tags are also using it. In future other new components can also make use of identity.
### Does this PR introduce _any_ user-facing change?
NA
### How was this patch tested?
Existing UTs
Closes #2952 from s0nskar/fix_identity.
Authored-by: Sanskar Modi <sa...@gmail.com>
Signed-off-by: Shuang <lv...@alibaba-inc.com>
---
.../hadoop/mapred/CelebornMapOutputCollector.java | 3 +-
.../task/reduce/CelebornShuffleConsumer.java | 2 +-
.../org/apache/celeborn/common/CelebornConf.scala | 85 +++++++++++++---------
.../common/identity/DefaultIdentityProvider.scala | 4 +-
.../common/identity/IdentityProvider.scala | 2 +-
.../apache/celeborn/common/util/UtilsSuite.scala | 2 +-
docs/configuration/client.md | 6 +-
docs/configuration/quota.md | 3 -
8 files changed, 61 insertions(+), 46 deletions(-)
diff --git a/client-mr/mr/src/main/java/org/apache/hadoop/mapred/CelebornMapOutputCollector.java b/client-mr/mr/src/main/java/org/apache/hadoop/mapred/CelebornMapOutputCollector.java
index 1c2f57809..334636818 100644
--- a/client-mr/mr/src/main/java/org/apache/hadoop/mapred/CelebornMapOutputCollector.java
+++ b/client-mr/mr/src/main/java/org/apache/hadoop/mapred/CelebornMapOutputCollector.java
@@ -66,8 +66,7 @@ public class CelebornMapOutputCollector<K extends Object, V extends Object>
applicationAttemptId,
IOBufferSize);
UserIdentifier userIdentifier =
- new UserIdentifier(
- celebornConf.quotaUserSpecificTenant(), celebornConf.quotaUserSpecificUserName());
+ new UserIdentifier(celebornConf.userSpecificTenant(), celebornConf.userSpecificUserName());
final float spiller = jobConf.getFloat(JobContext.MAP_SORT_SPILL_PERCENT, (float) 0.8);
int pushSize = (int) ((IOBufferSize << 20) * spiller);
diff --git a/client-mr/mr/src/main/java/org/apache/hadoop/mapreduce/task/reduce/CelebornShuffleConsumer.java b/client-mr/mr/src/main/java/org/apache/hadoop/mapreduce/task/reduce/CelebornShuffleConsumer.java
index 052b340ab..778e243fa 100644
--- a/client-mr/mr/src/main/java/org/apache/hadoop/mapreduce/task/reduce/CelebornShuffleConsumer.java
+++ b/client-mr/mr/src/main/java/org/apache/hadoop/mapreduce/task/reduce/CelebornShuffleConsumer.java
@@ -80,7 +80,7 @@ public class CelebornShuffleConsumer<K, V>
lmPort,
celebornConf,
new UserIdentifier(
- celebornConf.quotaUserSpecificTenant(), celebornConf.quotaUserSpecificUserName()));
+ celebornConf.userSpecificTenant(), celebornConf.userSpecificUserName()));
this.merger =
new MergeManagerImpl<>(
reduceId,
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 32cf2ebfa..cbc66c404 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -28,7 +28,7 @@ import scala.util.Try
import scala.util.matching.Regex
import org.apache.celeborn.common.authentication.AnonymousAuthenticationProviderImpl
-import org.apache.celeborn.common.identity.{DefaultIdentityProvider, IdentityProvider}
+import org.apache.celeborn.common.identity.{DefaultIdentityProvider, HadoopBasedIdentityProvider, IdentityProvider}
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.internal.config._
import org.apache.celeborn.common.network.util.ByteUnit
@@ -884,11 +884,15 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
// Quota //
// //////////////////////////////////////////////////////
def quotaEnabled: Boolean = get(QUOTA_ENABLED)
- def quotaIdentityProviderClass: String = get(QUOTA_IDENTITY_PROVIDER)
- def quotaUserSpecificTenant: String = get(QUOTA_USER_SPECIFIC_TENANT)
- def quotaUserSpecificUserName: String = get(QUOTA_USER_SPECIFIC_USERNAME)
def quotaInterruptShuffleEnabled: Boolean = get(QUOTA_INTERRUPT_SHUFFLE_ENABLED)
+ // //////////////////////////////////////////////////////
+ // Identity //
+ // //////////////////////////////////////////////////////
+ def identityProviderClass: String = get(IDENTITY_PROVIDER)
+ def userSpecificTenant: String = get(USER_SPECIFIC_TENANT)
+ def userSpecificUserName: String = get(USER_SPECIFIC_USERNAME)
+
// //////////////////////////////////////////////////////
// Client //
// //////////////////////////////////////////////////////
@@ -1589,7 +1593,19 @@ object CelebornConf extends Logging {
DeprecatedConfig(
"celeborn.worker.congestionControl.high.watermark",
"0.6.0",
- "Please use celeborn.worker.congestionControl.diskBuffer.high.watermark"))
+ "Please use celeborn.worker.congestionControl.diskBuffer.high.watermark"),
+ DeprecatedConfig(
+ "celeborn.quota.identity.provider",
+ "0.6.0",
+ "Please use celeborn.identity.provider"),
+ DeprecatedConfig(
+ "celeborn.quota.identity.user-specific.tenant",
+ "0.6.0",
+ "Please use celeborn.identity.user-specific.tenant"),
+ DeprecatedConfig(
+ "celeborn.quota.identity.user-specific.userName",
+ "0.6.0",
+ "Please use celeborn.identity.user-specific.userName"))
Map(configs.map { cfg => (cfg.key -> cfg) }: _*)
}
@@ -5324,6 +5340,37 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(true)
+ val IDENTITY_PROVIDER: ConfigEntry[String] =
+ buildConf("celeborn.identity.provider")
+ .withAlternative("celeborn.quota.identity.provider")
+ .categories("client")
+ .doc(s"IdentityProvider class name. Default class is " +
+ s"`${classOf[DefaultIdentityProvider].getName}`. " +
+ s"Optional values: " +
+ s"${classOf[HadoopBasedIdentityProvider].getName} user name will be obtained by UserGroupInformation.getUserName; " +
+ s"${classOf[DefaultIdentityProvider].getName} user name and tenant id are default values or user-specific values.")
+ .version("0.6.0")
+ .stringConf
+ .createWithDefault(classOf[DefaultIdentityProvider].getName)
+
+ val USER_SPECIFIC_TENANT: ConfigEntry[String] =
+ buildConf("celeborn.identity.user-specific.tenant")
+ .withAlternative("celeborn.quota.identity.user-specific.tenant")
+ .categories("client")
+ .doc(s"Tenant id if ${IDENTITY_PROVIDER.key} is ${classOf[DefaultIdentityProvider].getName}.")
+ .version("0.6.0")
+ .stringConf
+ .createWithDefault(IdentityProvider.DEFAULT_TENANT_ID)
+
+ val USER_SPECIFIC_USERNAME: ConfigEntry[String] =
+ buildConf("celeborn.identity.user-specific.userName")
+ .withAlternative("celeborn.quota.identity.user-specific.userName")
+ .categories("client")
+ .doc(s"User name if ${IDENTITY_PROVIDER.key} is ${classOf[DefaultIdentityProvider].getName}.")
+ .version("0.6.0")
+ .stringConf
+ .createWithDefault(IdentityProvider.DEFAULT_USERNAME)
+
val QUOTA_ENABLED: ConfigEntry[Boolean] =
buildConf("celeborn.quota.enabled")
.categories("quota", "master", "client")
@@ -5337,18 +5384,6 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(true)
- val QUOTA_IDENTITY_PROVIDER: ConfigEntry[String] =
- buildConf("celeborn.quota.identity.provider")
- .categories("quota", "client")
- .doc(s"IdentityProvider class name. Default class is " +
- s"`${classOf[DefaultIdentityProvider].getName}`. " +
- s"Optional values: " +
- s"org.apache.celeborn.common.identity.HadoopBasedIdentityProvider user name will be obtained by UserGroupInformation.getUserName; " +
- s"org.apache.celeborn.common.identity.DefaultIdentityProvider user name and tenant id are default values or user-specific values.")
- .version("0.2.0")
- .stringConf
- .createWithDefault(classOf[DefaultIdentityProvider].getName)
-
val CONTAINER_INFO_PROVIDER: ConfigEntry[String] =
buildConf("celeborn.container.info.provider")
.categories("master", "worker")
@@ -5358,22 +5393,6 @@ object CelebornConf extends Logging {
.stringConf
.createWithDefault("org.apache.celeborn.server.common.container.DefaultContainerInfoProvider")
- val QUOTA_USER_SPECIFIC_TENANT: ConfigEntry[String] =
- buildConf("celeborn.quota.identity.user-specific.tenant")
- .categories("quota", "client")
- .doc(s"Tenant id if celeborn.quota.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider.")
- .version("0.3.0")
- .stringConf
- .createWithDefault(IdentityProvider.DEFAULT_TENANT_ID)
-
- val QUOTA_USER_SPECIFIC_USERNAME: ConfigEntry[String] =
- buildConf("celeborn.quota.identity.user-specific.userName")
- .categories("quota", "client")
- .doc(s"User name if celeborn.quota.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider.")
- .version("0.3.0")
- .stringConf
- .createWithDefault(IdentityProvider.DEFAULT_USERNAME)
-
val QUOTA_DISK_BYTES_WRITTEN: ConfigEntry[Long] =
buildConf("celeborn.quota.tenant.diskBytesWritten")
.categories("quota")
diff --git a/common/src/main/scala/org/apache/celeborn/common/identity/DefaultIdentityProvider.scala b/common/src/main/scala/org/apache/celeborn/common/identity/DefaultIdentityProvider.scala
index 76b89b331..d9f1a10ad 100644
--- a/common/src/main/scala/org/apache/celeborn/common/identity/DefaultIdentityProvider.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/identity/DefaultIdentityProvider.scala
@@ -23,7 +23,7 @@ class DefaultIdentityProvider extends IdentityProvider {
override def provide(): UserIdentifier = {
val conf = new CelebornConf()
UserIdentifier(
- conf.quotaUserSpecificTenant,
- conf.quotaUserSpecificUserName)
+ conf.userSpecificTenant,
+ conf.userSpecificUserName)
}
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/identity/IdentityProvider.scala b/common/src/main/scala/org/apache/celeborn/common/identity/IdentityProvider.scala
index f734ffffe..5d4541840 100644
--- a/common/src/main/scala/org/apache/celeborn/common/identity/IdentityProvider.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/identity/IdentityProvider.scala
@@ -30,6 +30,6 @@ object IdentityProvider extends Logging {
val DEFAULT_USERNAME = "default"
def instantiate(conf: CelebornConf): IdentityProvider = {
- Utils.instantiate[IdentityProvider](conf.quotaIdentityProviderClass)
+ Utils.instantiate[IdentityProvider](conf.identityProviderClass)
}
}
diff --git a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
index 2c0000fd9..afc74707f 100644
--- a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
+++ b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
@@ -248,7 +248,7 @@ class UtilsSuite extends CelebornFunSuite {
test("test instantiate") {
val celebornConf = new CelebornConf()
- assert(Utils.instantiate[DefaultIdentityProvider](celebornConf.quotaIdentityProviderClass)
+ assert(Utils.instantiate[DefaultIdentityProvider](celebornConf.identityProviderClass)
.isInstanceOf[DefaultIdentityProvider])
}
}
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index 606c216b0..f035713eb 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -121,12 +121,12 @@ license: |
| celeborn.client.spark.shuffle.forceFallback.enabled | false | false | Always use spark built-in shuffle implementation. This configuration is deprecated, consider configuring `celeborn.client.spark.shuffle.fallback.policy` instead. | 0.3.0 | celeborn.shuffle.forceFallback.enabled |
| celeborn.client.spark.shuffle.writer | HASH | false | Celeborn supports the following kind of shuffle writers. 1. hash: hash-based shuffle writer works fine when shuffle partition count is normal; 2. sort: sort-based shuffle writer works fine when memory pressure is high or shuffle partition count is huge. This configuration only takes effect when celeborn.client.spark.push.dynamicWriteMode.enabled is false. | 0.3.0 | celeborn.shuffle.writer |
| celeborn.client.spark.stageRerun.enabled | true | false | Whether to enable stage rerun. If true, client throws FetchFailedException instead of CelebornIOException. | 0.4.0 | celeborn.client.spark.fetch.throwsFetchFailure |
+| celeborn.identity.provider | org.apache.celeborn.common.identity.DefaultIdentityProvider | false | IdentityProvider class name. Default class is `org.apache.celeborn.common.identity.DefaultIdentityProvider`. Optional values: org.apache.celeborn.common.identity.HadoopBasedIdentityProvider user name will be obtained by UserGroupInformation.getUserName; org.apache.celeborn.common.identity.DefaultIdentityProvider user name and tenant id are default values or user-specific values. | 0.6.0 | [...]
+| celeborn.identity.user-specific.tenant | default | false | Tenant id if celeborn.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider. | 0.6.0 | celeborn.quota.identity.user-specific.tenant |
+| celeborn.identity.user-specific.userName | default | false | User name if celeborn.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider. | 0.6.0 | celeborn.quota.identity.user-specific.userName |
| celeborn.master.endpoints | <localhost>:9097 | false | Endpoints of master nodes for celeborn clients to connect. Client uses resolver provided by celeborn.master.endpoints.resolver to resolve the master endpoints. By default Celeborn uses `org.apache.celeborn.common.client.StaticMasterEndpointResolver` which take static master endpoints as input. Allowed pattern: `<host1>:<port1>[,<host2>:<port2>]*`, e.g. `clb1:9097,clb2:9098,clb3:9099`. If the port is omitted, 9097 will be used [...]
| celeborn.master.endpoints.resolver | org.apache.celeborn.common.client.StaticMasterEndpointResolver | false | Resolver class that can be used for discovering and updating the master endpoints. This allows users to provide a custom master endpoint resolver implementation. This is useful in environments where the master nodes might change due to scaling operations or infrastructure updates. Clients need to ensure that provided resolver class should be present in the classpath. | 0.6.0 | |
| celeborn.quota.enabled | true | false | When Master side sets to true, the master will enable to check the quota via QuotaManager. When Client side sets to true, LifecycleManager will request Master side to check whether the current user has enough quota before registration of shuffle. Fallback to the default shuffle service when Master side checks that there is no enough quota for current user. | 0.2.0 | |
-| celeborn.quota.identity.provider | org.apache.celeborn.common.identity.DefaultIdentityProvider | false | IdentityProvider class name. Default class is `org.apache.celeborn.common.identity.DefaultIdentityProvider`. Optional values: org.apache.celeborn.common.identity.HadoopBasedIdentityProvider user name will be obtained by UserGroupInformation.getUserName; org.apache.celeborn.common.identity.DefaultIdentityProvider user name and tenant id are default values or user-specific values. | 0 [...]
-| celeborn.quota.identity.user-specific.tenant | default | false | Tenant id if celeborn.quota.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider. | 0.3.0 | |
-| celeborn.quota.identity.user-specific.userName | default | false | User name if celeborn.quota.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider. | 0.3.0 | |
| celeborn.quota.interruptShuffle.enabled | false | false | Whether to enable interrupt shuffle when quota exceeds. | 0.6.0 | |
| celeborn.storage.availableTypes | HDD | false | Enabled storages. Available options: MEMORY,HDD,SSD,HDFS,S3. Note: HDD and SSD would be treated as identical. | 0.3.0 | celeborn.storage.activeTypes |
| celeborn.storage.hdfs.dir | <undefined> | false | HDFS base directory for Celeborn to store shuffle data. | 0.2.0 | |
diff --git a/docs/configuration/quota.md b/docs/configuration/quota.md
index 9e362ae7e..b1fa4f940 100644
--- a/docs/configuration/quota.md
+++ b/docs/configuration/quota.md
@@ -20,9 +20,6 @@ license: |
| Key | Default | isDynamic | Description | Since | Deprecated |
| --- | ------- | --------- | ----------- | ----- | ---------- |
| celeborn.quota.enabled | true | false | When Master side sets to true, the master will enable to check the quota via QuotaManager. When Client side sets to true, LifecycleManager will request Master side to check whether the current user has enough quota before registration of shuffle. Fallback to the default shuffle service when Master side checks that there is no enough quota for current user. | 0.2.0 | |
-| celeborn.quota.identity.provider | org.apache.celeborn.common.identity.DefaultIdentityProvider | false | IdentityProvider class name. Default class is `org.apache.celeborn.common.identity.DefaultIdentityProvider`. Optional values: org.apache.celeborn.common.identity.HadoopBasedIdentityProvider user name will be obtained by UserGroupInformation.getUserName; org.apache.celeborn.common.identity.DefaultIdentityProvider user name and tenant id are default values or user-specific values. | 0 [...]
-| celeborn.quota.identity.user-specific.tenant | default | false | Tenant id if celeborn.quota.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider. | 0.3.0 | |
-| celeborn.quota.identity.user-specific.userName | default | false | User name if celeborn.quota.identity.provider is org.apache.celeborn.common.identity.DefaultIdentityProvider. | 0.3.0 | |
| celeborn.quota.interruptShuffle.enabled | false | false | Whether to enable interrupt shuffle when quota exceeds. | 0.6.0 | |
| celeborn.quota.tenant.diskBytesWritten | 9223372036854775807 | true | Quota dynamic configuration for written disk bytes. | 0.5.0 | |
| celeborn.quota.tenant.diskFileCount | 9223372036854775807 | true | Quota dynamic configuration for written disk file count. | 0.5.0 | |