kingamarton commented on a change in pull request #244:
URL:
https://github.com/apache/incubator-yunikorn-k8shim/pull/244#discussion_r596976942
##########
File path: pkg/callback/scheduler_callback.go
##########
@@ -88,6 +88,27 @@ func (callback *AsyncRMCallback) RecvUpdateResponse(response
*si.UpdateResponse)
}
}
+ // handle status changes
+ for _, updated := range response.UpdatedApplications {
+ log.Logger().Debug("status update callback received",
+ zap.String("appId", updated.ApplicationID),
+ zap.String("new status", updated.State))
+ // delete application from context
+ if updated.State == events.States().Application.Completed {
+ err :=
callback.context.RemoveApplicationInternal(updated.ApplicationID)
+ if err != nil {
+ log.Logger().Error("failed to delete
application", zap.Error(err))
+ }
+ } else {
+ if updated.State == "Failing" || updated.State ==
events.States().Application.Failed {
Review comment:
Yes, we need this change. Without this check the shim will continue to
send the real pods for scheduling. Even if we would handle it correctly in the
core side, it is wrong. We should not send any new requests when the app is
failing or failed. Also the app in the shim and core side should be in sync. We
should not keep the app running in the shim side while it is already Failed in
the core side.
For now I think we can keep this string check for the "Failing" state, but I
opened a Jira for introducing a new state in the shim side as well: YUNIKORN-585
##########
File path: pkg/callback/scheduler_callback.go
##########
@@ -88,6 +88,27 @@ func (callback *AsyncRMCallback) RecvUpdateResponse(response
*si.UpdateResponse)
}
}
+ // handle status changes
+ for _, updated := range response.UpdatedApplications {
+ log.Logger().Debug("status update callback received",
+ zap.String("appId", updated.ApplicationID),
+ zap.String("new status", updated.State))
+ // delete application from context
+ if updated.State == events.States().Application.Completed {
+ err :=
callback.context.RemoveApplicationInternal(updated.ApplicationID)
+ if err != nil {
+ log.Logger().Error("failed to delete
application", zap.Error(err))
+ }
+ } else {
+ if updated.State == "Failing" || updated.State ==
events.States().Application.Failed {
+ ev :=
cache.NewFailApplicationEvent(updated.ApplicationID)
+ dispatcher.Dispatch(ev)
+ }
+ // handle status update
+
dispatcher.Dispatch(cache.NewApplicationStatusChangeEvent(updated.ApplicationID,
events.AppStateChange, updated.State))
+ }
+ }
+
Review comment:
moved it back
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
