rm unnecessary code & update state transition

go/tasks/plugins/array/core/state.go

@@ -246,23 +246,22 @@ func SummaryToPhase(ctx context.Context, minSuccesses int64, summary arraystatus
 		}
 	}
 
-	if totalWaitingForResources > 0 {
-		logger.Infof(ctx, "Array is still running and waiting for resources totalWaitingForResources[%v]", totalWaitingForResources)
-		return PhaseWaitingForResources
-	}
-
 	if totalCount < minSuccesses {
 		logger.Infof(ctx, "Array failed because totalCount[%v] < minSuccesses[%v]", totalCount, minSuccesses)
 		return PhaseWriteToDiscoveryThenFail
 	}
 
 	// No chance to reach the required success numbers.
-	if totalRunning+totalSuccesses < minSuccesses {
-		logger.Infof(ctx, "Array failed early because totalRunning[%v] + totalSuccesses[%v] < minSuccesses[%v]",
-			totalRunning, totalSuccesses, minSuccesses)
+	if totalRunning+totalSuccesses+totalWaitingForResources < minSuccesses {
+		logger.Infof(ctx, "Array failed early because totalRunning[%v] + totalSuccesses[%v] + totalWaitingForResource[%v] < minSuccesses[%v]",
+			totalRunning, totalSuccesses, totalWaitingForResources, minSuccesses)
 		return PhaseWriteToDiscoveryThenFail
 	}
 
+	if totalWaitingForResources > 0 {
+		logger.Infof(ctx, "Array is still running and waiting for resources totalWaitingForResources[%v]", totalWaitingForResources)
+		return PhaseWaitingForResources
+	}
 	if totalSuccesses >= minSuccesses && totalRunning == 0 {
 		logger.Infof(ctx, "Array succeeded because totalSuccesses[%v] >= minSuccesses[%v]", totalSuccesses, minSuccesses)
 		return PhaseWriteToDiscovery

go/tasks/plugins/array/k8s/launcher.go

@@ -21,7 +21,6 @@ const (
 	ErrSubmitJob              errors2.ErrorCode = "SUBMIT_JOB_FAILED"
 	JobIndexVarName           string            = "BATCH_JOB_ARRAY_INDEX_VAR_NAME"
 	FlyteK8sArrayIndexVarName string            = "FLYTE_K8S_ARRAY_INDEX"
-	ResourcesPrimaryLabel     string            = "token"
 )
 
 var arrayJobEnvVars = []corev1.EnvVar{

go/tasks/plugins/array/k8s/task.go

@@ -163,15 +163,9 @@ func (t Task) Abort(ctx context.Context, tCtx core.TaskExecutionContext, kubeCli
 	err := kubeClient.GetClient().Delete(ctx, pod)
 	if err != nil {
 		if k8serrors.IsNotFound(err) {
-			// Deallocate Resource
-			err = deallocateResource(ctx, tCtx, t.Config, t.ChildIdx)
-			if err != nil {
-				logger.Errorf(ctx, "Error releasing allocation token [%s] in Finalize [%s]", podName, err)
-				return err
-			}
+
 			return nil
 		}
-
 		return err
 	}