You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by pb...@apache.org on 2024/05/22 21:07:49 UTC
(yunikorn-k8shim) branch master updated: [YUNIKORN-2618]Update Streamline AsyncRMCallback UpdateAllocation (#846)
This is an automated email from the ASF dual-hosted git repository.
pbacsko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 08a9b93d [YUNIKORN-2618]Update Streamline AsyncRMCallback UpdateAllocation (#846)
08a9b93d is described below
commit 08a9b93dcc308f1e70bbf9a6596664877977e1bd
Author: YUN SUN <yu...@gmail.com>
AuthorDate: Wed May 22 23:07:37 2024 +0200
[YUNIKORN-2618]Update Streamline AsyncRMCallback UpdateAllocation (#846)
Closes: #846
Signed-off-by: Peter Bacsko <pb...@cloudera.com>
---
pkg/cache/scheduler_callback.go | 45 +++++++++++++++++------------------------
1 file changed, 19 insertions(+), 26 deletions(-)
diff --git a/pkg/cache/scheduler_callback.go b/pkg/cache/scheduler_callback.go
index dd147ff0..8ed487c4 100644
--- a/pkg/cache/scheduler_callback.go
+++ b/pkg/cache/scheduler_callback.go
@@ -56,27 +56,24 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons
// update cache
task := callback.context.getTask(alloc.ApplicationID, alloc.AllocationKey)
- if task != nil {
- task.setAllocationKey(alloc.AllocationKey)
- } else {
+ if task == nil {
log.Log(log.ShimRMCallback).Warn("Unable to get task", zap.String("taskID", alloc.AllocationKey))
+ continue
}
+
+ task.setAllocationKey(alloc.AllocationKey)
+
if err := callback.context.AssumePod(alloc.AllocationKey, alloc.NodeID); err != nil {
- if task != nil {
- task.failWithEvent(err.Error(), "AssumePodError")
- }
+ task.failWithEvent(err.Error(), "AssumePodError")
return err
}
- if app := callback.context.GetApplication(alloc.ApplicationID); app != nil {
- if task != nil {
- if utils.IsAssignedPod(task.GetTaskPod()) {
- // task is already bound, fixup state and continue
- task.MarkPreviouslyAllocated(alloc.AllocationKey, alloc.NodeID)
- } else {
- ev := NewAllocateTaskEvent(app.GetApplicationID(), task.taskID, alloc.AllocationKey, alloc.NodeID)
- dispatcher.Dispatch(ev)
- }
- }
+
+ if utils.IsAssignedPod(task.GetTaskPod()) {
+ // task is already bound, fixup state and continue
+ task.MarkPreviouslyAllocated(alloc.AllocationKey, alloc.NodeID)
+ } else {
+ ev := NewAllocateTaskEvent(alloc.ApplicationID, task.taskID, alloc.AllocationKey, alloc.NodeID)
+ dispatcher.Dispatch(ev)
}
}
@@ -84,22 +81,18 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons
// request rejected by the scheduler, put it back and try scheduling again
log.Log(log.ShimRMCallback).Debug("callback: response to rejected ask",
zap.String("allocationKey", reject.AllocationKey))
- if app := callback.context.GetApplication(reject.ApplicationID); app != nil {
- dispatcher.Dispatch(NewRejectTaskEvent(app.GetApplicationID(), reject.AllocationKey,
- fmt.Sprintf("task %s ask from application %s is rejected by scheduler",
- reject.AllocationKey, reject.ApplicationID)))
- }
+ dispatcher.Dispatch(NewRejectTaskEvent(reject.ApplicationID, reject.AllocationKey,
+ fmt.Sprintf("task %s ask from application %s is rejected by scheduler",
+ reject.AllocationKey, reject.ApplicationID)))
}
for _, reject := range response.RejectedAllocations {
// request rejected by the scheduler, reject it
log.Log(log.ShimRMCallback).Debug("callback: response to rejected allocation",
zap.String("allocationKey", reject.AllocationKey))
- if app := callback.context.GetApplication(reject.ApplicationID); app != nil {
- dispatcher.Dispatch(NewRejectTaskEvent(app.GetApplicationID(), reject.AllocationKey,
- fmt.Sprintf("task %s allocation from application %s is rejected by scheduler",
- reject.AllocationKey, reject.ApplicationID)))
- }
+ dispatcher.Dispatch(NewRejectTaskEvent(reject.ApplicationID, reject.AllocationKey,
+ fmt.Sprintf("task %s allocation from application %s is rejected by scheduler",
+ reject.AllocationKey, reject.ApplicationID)))
}
for _, release := range response.Released {
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org