Copilot commented on code in PR #3803:
URL: https://github.com/apache/celeborn/pull/3803#discussion_r3845255604
##########
master/src/main/java/org/apache/celeborn/service/deploy/master/slotsalloc/SlotsAllocator.java:
##########
@@ -458,6 +458,7 @@ && satisfyRackAware(
replicaIndex,
index ->
!(sameWorkerCandidates && index == selectedPrimaryIndex)
+ && canAssign(null, replicaWorkers.get(index),
availableStorageTypes)
&& satisfyRackAware(true, primaryWorker,
replicaWorkers.get(index)));
Review Comment:
`canAssign` is called with `null` for the first argument in the replica
selection predicate, even though this code path appears to have `slotBudgets`
in scope later when building `StorageInfo`. If `canAssign` is intended to
validate feasibility under `slotBudgets` (or similar constraints), passing
`null` can cause the selector to accept a worker that later can’t actually be
assigned, leading to allocation failure or inconsistent behavior between
selection and assignment. Consider passing the same budgeting/context object
used for `buildStorageInfo(...)` (e.g., `slotBudgets`) into `canAssign`, or
introduce a dedicated `canAssignWorker(WorkerInfo, availableStorageTypes, ...)`
that matches what the selection step truly needs.
##########
master/src/test/java/org/apache/celeborn/service/deploy/master/slotsalloc/SlotsAllocatorRackAwareSuiteJ.java:
##########
@@ -183,6 +183,95 @@ public void accept(WorkerInfo workerInfo) {
return workers;
}
+ @Test
+ public void offerSlotsRackAwareFallbackRequiresReplicaDisk() {
+ List<WorkerInfo> workers =
+ Arrays.asList(
+ prepareWorker("disk-worker", "/rack/r1", "/mnt/disk1"),
+ prepareWorker("diskless-worker", "/rack/r2", null));
+
+ Map<WorkerInfo, Tuple2<List<PartitionLocation>, List<PartitionLocation>>>
slots =
+ SlotsAllocator.offerSlots(
+ workers,
+ Collections.singletonList(0),
+ true,
+ true,
+ StorageInfo.LOCAL_DISK_MASK,
+ false,
+ 0,
+ SLOTS_ASSIGN_STRATEGY);
+
+ Assert.assertTrue(slots.isEmpty());
+ }
+
+ @Test
+ public void offerSlotsRackAwareFallbackSkipsDisklessReplica() {
+ WorkerInfo primaryWorker = prepareWorker("primary-worker", "/rack/r1",
"/mnt/disk1");
+ WorkerInfo disklessReplica = prepareWorker("diskless-replica", "/rack/r2",
null);
+ WorkerInfo diskReplica = prepareWorker("disk-replica", "/rack/r2",
"/mnt/disk2");
+ // Use separate candidate lists and two partitions so the diskless replica
is scanned regardless
+ // of the random initial replica index.
+ disklessReplica.nextInterruptionNotice_$eq(1L);
+ diskReplica.nextInterruptionNotice_$eq(2L);
+
+ Map<WorkerInfo, Tuple2<List<PartitionLocation>, List<PartitionLocation>>>
slots =
+ SlotsAllocator.offerSlots(
+ Arrays.asList(primaryWorker, disklessReplica, diskReplica),
+ Arrays.asList(0, 1),
+ true,
+ true,
+ StorageInfo.LOCAL_DISK_MASK,
+ true,
+ 0,
+ SLOTS_ASSIGN_STRATEGY);
+
+ Assert.assertFalse(slots.containsKey(disklessReplica));
+ Assert.assertEquals(2, slots.get(primaryWorker)._1.size());
+ Assert.assertEquals(2, slots.get(diskReplica)._2.size());
+ }
+
+ @Test
+ public void offerSlotsRackAwareFallbackUsesReplicaStorageInfo() {
+ List<WorkerInfo> workers =
+ Arrays.asList(
+ prepareWorker("worker1", "/rack/r1", "/mnt/disk1"),
+ prepareWorker("worker2", "/rack/r2", "/mnt/disk2"));
+
+ Map<WorkerInfo, Tuple2<List<PartitionLocation>, List<PartitionLocation>>>
slots =
+ SlotsAllocator.offerSlots(
+ workers,
+ Collections.singletonList(0),
+ true,
+ true,
+ StorageInfo.LOCAL_DISK_MASK,
+ false,
+ 0,
+ SLOTS_ASSIGN_STRATEGY);
+
+ Assert.assertEquals(2, slots.size());
+ workers.forEach(
+ worker -> {
+ List<PartitionLocation> locations = new ArrayList<>();
+ locations.addAll(slots.get(worker)._1);
+ locations.addAll(slots.get(worker)._2);
+ Assert.assertEquals(1, locations.size());
+ Assert.assertEquals(
+ worker.diskInfos().keySet().iterator().next(),
+ locations.get(0).getStorageInfo().getMountPoint());
+ });
+ }
+
+ private static WorkerInfo prepareWorker(String host, String rack, String
mountPoint) {
+ Map<String, DiskInfo> diskInfos = new HashMap<>();
+ if (mountPoint != null) {
+ // Keep availableSlots at zero to force allocation through the
best-effort fallback.
+ diskInfos.put(mountPoint, new DiskInfo(mountPoint, 1024L, 1, 1, 0));
+ }
+ WorkerInfo worker = new WorkerInfo(host, 1, 2, 3, 4, 5, diskInfos, null);
+ worker.networkLocation_$eq(rack);
+ return worker;
+ }
Review Comment:
The test helper uses a set of positional “magic numbers” in the `WorkerInfo`
constructor (`1, 2, 3, 4, 5`) that makes the setup harder to understand and
more fragile if the constructor semantics change. Consider replacing these with
named constants in the test (e.g., `RPC_PORT`, `PUSH_PORT`, etc.), or using a
builder/factory that makes the intent explicit while keeping the test focused
on disk/mount behavior.
--
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]
