[YUNIKORN-2884] Task fail with post allocated but the pod will keep p…

pkg/cache/application.go

@@ -27,6 +27,7 @@
 	"github.com/looplab/fsm"
 	"go.uber.org/zap"
 	v1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/apache/yunikorn-k8shim/pkg/common"
@@ -636,9 +637,19 @@
 
 	if task, ok := app.taskMap[taskID]; ok {
 		task.setTaskTerminationType(terminationType)
-		err := task.DeleteTaskPod()
-		if err != nil {
-			log.Log(log.ShimCacheApplication).Error("failed to release allocation from application", zap.Error(err))
+		pod := task.GetTaskPod()
+		// Get the pod first to check if it exists
+		// For example, users manually killed the pod, so the pod already deleted before release.
+		_, err := task.context.apiProvider.GetAPIs().KubeClient.Get(pod.Namespace, pod.Name)
+		if err != nil && k8serrors.IsNotFound(err) {
+			// Pod does not exist, no need to delete
+			log.Log(log.ShimCacheApplication).Info("pod not found, it already deleted or released")
+		} else {
+			// Pod exists, delete it
+			err = task.DeleteTaskPod()
+			if err != nil {
+				log.Log(log.ShimCacheApplication).Error("failed to release allocation from application", zap.Error(err))
+			}
 		}
 		app.publishPlaceholderTimeoutEvents(task)
 	} else {

pkg/cache/scheduler_callback.go

@@ -93,12 +93,11 @@ func (callback *AsyncRMCallback) UpdateAllocation(response *si.AllocationRespons
 		// update cache
 		callback.context.ForgetPod(release.GetAllocationKey())
 
-		// TerminationType 0 mean STOPPED_BY_RM
-		if release.TerminationType != si.TerminationType_STOPPED_BY_RM {
-			// send release app allocation to application states machine
-			ev := NewReleaseAppAllocationEvent(release.ApplicationID, release.TerminationType, release.AllocationKey)
-			dispatcher.Dispatch(ev)
-		}
+		// TerminationType 0 mean STOPPED_BY_RM, but we also need to do the release when task failed,
+		// we also should send release event to application in case task failed but the pod is still pending.
+		// send release app allocation to application states machine
+		ev := NewReleaseAppAllocationEvent(release.ApplicationID, release.TerminationType, release.AllocationKey)
+		dispatcher.Dispatch(ev)
 	}
 
 	return nil

pkg/cache/scheduler_callback_test.go

@@ -215,7 +215,9 @@ func TestUpdateAllocation_AllocationReleased_StoppedByRM(t *testing.T) {
 	assert.NilError(t, err, "error updating allocation")
 	assert.Assert(t, !context.schedulerCache.IsAssumedPod(taskUID1))
 	err = utils.WaitForCondition(deleteCalled.Load, 10*time.Millisecond, 500*time.Millisecond)
-	assert.Error(t, err, "timeout waiting for condition") // pod is not expected to be deleted
+	// Pod should be deleted, because TerminationType_STOPPED_BY_RM will also be called when task fail.
+	// If we don't delete the pod, the pod will be stuck in pending state.
+	assert.NilError(t, err, "pod has not been deleted")
 }
 
 func TestUpdateApplication_Accepted(t *testing.T) {