fix: rps should correctly ignore other rps job msgs
the fair share resource pool (rp) actor would fail on receiving an unsupported moveJob message
which happens because those msgs are broadcast to all rps for an
agent rm.
@@ -565,18 +565,21 @@ func (rp *ResourcePool) moveJob(
anchorID model.JobID,
aheadOf bool,
) error {
- if rp.config.Scheduler.GetType() != config.PriorityScheduling {
- return fmt.Errorf("unable to perform operation on resource pool with %s",
- rp.config.Scheduler.GetType())
- }
if anchorID == "" || jobID == "" || anchorID == jobID {
return nil
}
+ // check whether the msg belongs to this resource pool or not.
+ // job messages to agent rm are forwarded to all resource pools.
if _, ok := rp.queuePositions[jobID]; !ok {
return nil
}
+ if rp.config.Scheduler.GetType() != config.PriorityScheduling {
+ return fmt.Errorf("unable to perform operation on resource pool with %s",
+ rp.config.Scheduler.GetType())
+ }
+
groupAddr, ok := rp.IDToGroupActor[jobID]
if !ok {
return sproto.ErrJobNotFound(jobID)
@@ -647,10 +650,6 @@ func (rp *ResourcePool) receiveJobQueueMsg(ctx *actor.Context) error {
ctx.Respond(rp.scheduler.JobQInfo(rp))
case sproto.MoveJob:
- if rp.config.Scheduler.GetType() != config.PriorityScheduling {
- return fmt.Errorf("unable to perform operation on resource pool with %s",
- rp.config.Scheduler.GetType())
- }
err := rp.moveJob(ctx, msg.ID, msg.Anchor, msg.Ahead)
ctx.Respond(err)
