Copilot commented on code in PR #3719:
URL: https://github.com/apache/celeborn/pull/3719#discussion_r3759850467
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/tags/TagsManager.scala:
##########
@@ -19,112 +19,64 @@ package org.apache.celeborn.service.deploy.master.tags
import java.util
import java.util.{Collections, Set => JSet}
-import java.util.concurrent.ConcurrentHashMap
import java.util.function.Predicate
import java.util.stream.Collectors
-import scala.collection.JavaConverters.{asScalaIteratorConverter,
mapAsScalaConcurrentMapConverter}
+import scala.collection.JavaConverters._
import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.WorkerInfo
-import org.apache.celeborn.common.util.JavaUtils
import org.apache.celeborn.server.common.service.config.ConfigService
class TagsManager(configService: Option[ConfigService]) extends Logging {
- private val defaultTagStore = JavaUtils.newConcurrentHashMap[String,
JSet[String]]()
- private val addNewTagFunc =
- new util.function.Function[String, ConcurrentHashMap.KeySetView[String,
java.lang.Boolean]]() {
- override def apply(t: String): ConcurrentHashMap.KeySetView[String,
java.lang.Boolean] =
- ConcurrentHashMap.newKeySet[String]()
- }
-
- private def getTagStore: ConcurrentHashMap[String, JSet[String]] = {
+ private def tagStore: Option[util.Map[String, JSet[String]]] = {
configService match {
- case Some(cs) =>
- // TODO: Make configStore.getTags return ConcurrentMap
- JavaUtils.newConcurrentHashMap(cs.getSystemConfigFromCache.getTags)
- case _ =>
- defaultTagStore
+ case Some(cs) => Option(cs.getSystemConfigFromCache.getTags)
+ case _ => None
}
}
Review Comment:
Previously this code defensively copied the config-store tags into a
concurrent map; now it directly uses the underlying `getTags` map. If that
underlying map (or its value sets) can be updated concurrently by the config
service, `getTaggedWorkers`/`getTagsForWorker` can observe races or throw
(e.g., `ConcurrentModificationException`). Suggestion (mandatory for thread
safety): take a snapshot copy per call (e.g., into a new
`HashMap`/`ConcurrentHashMap` with copied sets) or reintroduce the prior
`JavaUtils.newConcurrentHashMap(...)` approach so reads are stable.
##########
master/src/test/scala/org/apache/celeborn/service/deploy/master/tags/TagsManagerSuite.scala:
##########
@@ -199,4 +130,37 @@ class TagsManagerSuite extends CelebornFunSuite {
assert(taggedWorkers.contains(WORKER3))
}
}
+
+ test("getTaggedWorkers matches workers tagged via either config store or
self-registration") {
+ tagsManager = new TagsManager(Option(configService))
+ val selfTaggedWorker = workerWithTags("host4", TAG1)
+ val all = List(WORKER1, WORKER2, WORKER3, selfTaggedWorker).asJava
+
+ val tagged = tagsManager.getTaggedWorkers(user, TAG1, all)
+ assert(tagged.size == 3) // WORKER1, WORKER2 (config store) + selfTagged
(self)
+ assert(tagged.contains(WORKER1))
+ assert(tagged.contains(WORKER2))
+ assert(!tagged.contains(WORKER3))
+ assert(tagged.contains(selfTaggedWorker))
+ }
+
+ test("getTaggedWorkers matches a worker whose tags span config store and
self-registration") {
+ tagsManager = new TagsManager(Option(configService))
+ // host1 already tagged via config service
+ val selfTaggedWorker = workerWithTags("host1", TAG2)
+ val all = List(WORKER1, WORKER2, WORKER3, selfTaggedWorker).asJava
Review Comment:
This test case specifically relies on the worker being the same identity as
the config-store entry ('host1 already tagged via config service'), but
`workerWithTags(\"host1\", ...)` likely produces a different `toUniqueId` (due
to different ports) than `WORKER1`. Adjust the test to ensure
`selfTaggedWorker.toUniqueId == WORKER1.toUniqueId` (e.g., create `WorkerInfo`
with `WORKER1`’s ports, then set `tags`).
##########
master/src/test/scala/org/apache/celeborn/service/deploy/master/tags/TagsManagerSuite.scala:
##########
@@ -38,127 +38,58 @@ class TagsManagerSuite extends CelebornFunSuite {
private val workers = List(WORKER1, WORKER2, WORKER3).asJava
private val user = UserIdentifier("tenant_01", "Jerry")
+ private var configService: ConfigService = _
override def beforeEach(): Unit = {
super.beforeEach()
DynamicConfigServiceFactory.reset()
- }
-
- test("test tags manager") {
- tagsManager = new TagsManager(Option(null))
-
- tagsManager.addTagToWorker(TAG1, WORKER1.toUniqueId)
- tagsManager.addTagToWorker(TAG1, WORKER2.toUniqueId)
-
- tagsManager.addTagToWorker(TAG2, WORKER2.toUniqueId)
- tagsManager.addTagToWorker(TAG2, WORKER3.toUniqueId)
- {
- val taggedWorkers = tagsManager.getTaggedWorkers(user, TAG1, workers)
- assert(taggedWorkers.size == 2)
- assert(taggedWorkers.contains(WORKER1))
- assert(taggedWorkers.contains(WORKER2))
- assert(!taggedWorkers.contains(WORKER3))
- }
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.DYNAMIC_CONFIG_STORE_BACKEND, "FS")
+ conf.set(
+ CelebornConf.DYNAMIC_CONFIG_STORE_FS_PATH.key,
+ getTestResourceFile("dynamicConfig-tags.yaml").getPath)
+ configService = DynamicConfigServiceFactory.getConfigService(conf)
+ }
- {
- val taggedWorkers = tagsManager.getTaggedWorkers(user, TAG2, workers)
- assert(taggedWorkers.size == 2)
- assert(!taggedWorkers.contains(WORKER1))
- assert(taggedWorkers.contains(WORKER2))
- assert(taggedWorkers.contains(WORKER3))
- }
+ private def workerWithTags(host: String, tags: String*): WorkerInfo = {
+ val w = new WorkerInfo(host, 111, 112, 113, 114, 115)
+ w.tags = new java.util.HashSet[String](tags.asJava)
+ w
+ }
Review Comment:
`workerWithTags` hard-codes ports, but tag lookup against the config-store
is done via `worker.toUniqueId`. In tests that intend to merge config-store
tags with self-registration for an existing worker, creating a `WorkerInfo`
with the same host but different ports will likely produce a different
`toUniqueId`, so the config-store tags won’t apply. Fix by constructing the
tagged worker using the exact same host/port tuple as the target worker (e.g.,
derive ports from `WORKER1`/`WORKER2`), or mutate `tags` on an existing
`WorkerInfo` instance used in the test.
--
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]
