Copilot commented on code in PR #3775:
URL: https://github.com/apache/celeborn/pull/3775#discussion_r3754918663
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1062,34 +1068,112 @@ private[celeborn] class Master(
s"extraSlots=$offerSlotsExtraSize"))
if (authEnabled) {
- pushApplicationMetaToWorkers(requestSlots, slots)
+ pushApplicationMetaToWorkers(requestSlots.applicationId, slots.keySet())
}
context.reply(RequestSlotsResponse(
StatusCode.SUCCESS,
slots.asInstanceOf[WorkerResource],
requestSlots.packed))
}
+ private def selectWorkersForRequest(
+ requestWorkers: PbRequestWorkers,
+ availableWorkers: util.List[WorkerInfo]): util.List[WorkerInfo] = {
+ val maxWorkers =
+ if (requestWorkers.getMaxWorkers <= 0) splitSlotAssignMaxWorkers
+ else Math.min(splitSlotAssignMaxWorkers, requestWorkers.getMaxWorkers)
+ val eligibleWorkers = new util.ArrayList[WorkerInfo]()
+ availableWorkers.asScala
+ .filter { worker =>
+
!StorageInfo.localDiskAvailable(requestWorkers.getAvailableStorageTypes) ||
+ worker.haveDisk
+ }
+ .foreach(eligibleWorkers.add)
+ if (eligibleWorkers.isEmpty) {
+ return Collections.emptyList()
+ }
+
+ val minWorkers = if (requestWorkers.getShouldReplicate) 2 else 1
+ val selectedWorkerCount =
+ Math.min(Math.max(minWorkers, maxWorkers), eligibleWorkers.size)
+ val startIndex = ThreadLocalRandom.current().nextInt(eligibleWorkers.size)
Review Comment:
`selectWorkersForRequest` can return SUCCESS with only 1 worker even when
`shouldReplicate=true` (when `eligibleWorkers.size == 1`). This contradicts the
documented/expected replication minimum and can cause the client to treat an
under-replicated candidate list as valid. Consider returning an empty selection
(so the caller responds with SLOT_NOT_AVAILABLE) when replication is requested
but fewer than 2 eligible workers exist.
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala:
##########
@@ -100,7 +104,8 @@ class ChangePartitionManagerUpdateWorkersSuite extends
WithShuffleClientSuite
setUpWorkers(workerConfForAdding, 2)
assert(workerInfos.size == 3)
- 0 until 10 foreach { partitionId: Int =>
+ var partitionId = 0
+ eventually(timeout(10.seconds), interval(0.milliseconds)) {
Review Comment:
`eventually(..., interval(0.milliseconds))` creates a tight retry loop that
can busy-spin the CPU and hammer RPCs, making this IT test flaky under load/CI.
Use a small non-zero polling interval (for example 100ms) to reduce churn while
still keeping the test responsive.
This issue also appears on line 211 of the same file.
--
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]
