Repository: kedacore/keda

Issue:
strange behaviour with job - scaler: accurate and pendingPodConditions
### Report

When using the accurate scaler and slow starting ML-containers, starting pods are not detected at all, and thus during the next pollingInterval a new job is created for the same message. With 15s pollingInterval, I usually end up with 4-6 jobs for a single message.

I tried defining the pendingPodConditions, but the pending pods are never detected - logs show 0 pending jobs, while running jobs are counted accurately.
```
scalingStrategy:
  strategy: "accurate"
  pendingPodConditions:
  - PodScheduled
  - Initialized
```

I dug into the pod conditions, which for my cluster are the following a fraction of a second after the pod has been scheduled on a node. So from this, the assumption was that PodSheduled and Initialized are good conditions to detect starting pods.
```
PodReadyToStartContainers:False (Last transition: 2025-04-23T14:28:09Z)
Initialized:True (Last transition: 2025-04-23T14:28:09Z)
Ready:False (Last transition: 2025-04-23T14:28:09Z)
ContainersReady:False (Last transition: 2025-04-23T14:28:0)
PodScheduled:True (Last transition: 2025-04-23T14:28:09Z)
```

Then I started looking at the code where I found the function [areAllPendingPodConditionsFulfilled](https://github.com/kedacore/keda/blob/b7f37a0fea67eb9126d40642c12534f60c298a06/pkg/scaling/executor/scale_jobs.go#L258):
```
	for _, pod := range pods.Items {
		for _, pendingConditionType := range pendingPodConditions {
			for _, podCondition := range pod.Status.Conditions {
				if string(podCondition.Type) == pendingConditionType && podCondition.Status == corev1.ConditionTrue {
					fulfilledConditionsCount++
				}
			}
		}
	}

	return len(pendingPodConditions) == fulfilledConditionsCount
```
This code seems to loop though a combination of all pods in the job and all conditions counting the aggregate matches, and not per pod. So, if I have two pending pods, with two pending states configured, I will get fulfilledConditionsCount == 4, which then wouldn't match len(pendingPodConditions) == 2 and thus this job would be considered to have no pending pods.



### Expected Behavior

Expected behaviour would be that any pod that fulfills any listed pendingPodConditions would be counted as a pending pod. 


### Actual Behavior

It seems to be hard if not impossible to get the scaler to accurately detect starting pods as pending.

### Steps to Reproduce the Problem

Any scaledjob with
```
scalingStrategy:
  strategy: "accurate"
  pendingPodConditions:
  - PodScheduled
  - Initialized
```

that spawns a pod with a startuptime that is longer than pollingInterval will trigger this.

### Logs from KEDA operator

```
2025-04-24T14:14:39Z	INFO	scaleexecutor	Scaling Jobs	{"scaledJob.Name": "some-job", "scaledJob.Namespace": "some-namespace", "Number of pending Jobs": 0}
```


### KEDA Version

2.16.0

### Kubernetes Version

1.31

### Platform

Other

### Scaler Details

Prometheus

### Anything else?

_No response_

Hints from issue comments:
@stale: This issue has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. Thank you for your contributions.
@stale: This issue has been automatically closed due to inactivity.
@rickbrouwer: Hi @diskis ,

I've made a fix. You can check it out in the PR mentioned above.
If you have the option and would like, you can create your own version of my branch to test it.
I can also create an image for you in my own registry if you'd like to verify that it works for you. If you'd like, please send me a message via Slack in the Kubernetes Slack.@diskis: Hi, first off thanks for taking this issue under work and sorry for late reply.

I built from your branch and deployed to our test environment. Now the job is detected as pending as it should, so from my side the fix looks good.@rickbrouwer: > Hi, first off thanks for taking this issue under work and sorry for late reply.
> 
> I built from your branch and deployed to our test environment. Now the job is detected as pending as it should, so from my side the fix looks good.

Great! Thanks for checking!

Solution patch:
diff --git a/.github/reviewers.yaml b/.github/reviewers.yml
similarity index 100%
rename from .github/reviewers.yaml
rename to .github/reviewers.yml
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f3a3f203e..3d28886a1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -83,6 +83,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 ### Fixes
 
 - **General**: Add missing omitempty json tags in the AuthPodIdentity struct ([#6779](https://github.com/kedacore/keda/issues/6779))
+- **General**: Correct pending pod condition logic for ScaledJobs ([#6727](https://github.com/kedacore/keda/issues/6727))
 - **General**: Fix prefixes on envFrom elements in a deployment spec aren't being interpreted and Environment variables are not prefixed with the prefix ([#6728](https://github.com/kedacore/keda/issues/6728))
 - **General**: Remove klogr dependency and replace with zap ([#5732](https://github.com/kedacore/keda/issues/5732))
 - **General**: Sets hpaName in Status when ScaledObject adopts/finds an existing HPA ([#6336](https://github.com/kedacore/keda/issues/6336))
diff --git a/pkg/scaling/executor/scale_jobs.go b/pkg/scaling/executor/scale_jobs.go
index 42d520c42..2223aa260 100644
--- a/pkg/scaling/executor/scale_jobs.go
+++ b/pkg/scaling/executor/scale_jobs.go
@@ -281,19 +281,29 @@ func (e *scaleExecutor) areAllPendingPodConditionsFulfilled(ctx context.Context,
 		return false
 	}
 
-	var fulfilledConditionsCount int
+	// Convert pendingPodConditions to a map for faster lookup
+	requiredConditions := make(map[string]struct{})
+	for _, condition := range pendingPodConditions {
+		requiredConditions[condition] = struct{}{}
+	}
 
+	// Check if any pod has all required conditions fulfilled
 	for _, pod := range pods.Items {
-		for _, pendingConditionType := range pendingPodConditions {
-			for _, podCondition := range pod.Status.Conditions {
-				if string(podCondition.Type) == pendingConditionType && podCondition.Status == corev1.ConditionTrue {
-					fulfilledConditionsCount++
-				}
+		fulfilledConditions := make(map[string]struct{})
+
+		for _, podCondition := range pod.Status.Conditions {
+			if _, isRequired := requiredConditions[string(podCondition.Type)]; isRequired && podCondition.Status == corev1.ConditionTrue {
+				fulfilledConditions[string(podCondition.Type)] = struct{}{}
 			}
 		}
+
+		// If this pod has all required conditions fulfilled, the job is no longer pending
+		if len(fulfilledConditions) == len(pendingPodConditions) {
+			return true
+		}
 	}
 
-	return len(pendingPodConditions) == fulfilledConditionsCount
+	return false
 }
 
 func (e *scaleExecutor) getPendingJobCount(ctx context.Context, scaledJob *kedav1alpha1.ScaledJob) int64 {



Test patch:
diff --git a/pkg/scaling/executor/scale_jobs_test.go b/pkg/scaling/executor/scale_jobs_test.go
index 9652abdc6..ea8aa5ff8 100644
--- a/pkg/scaling/executor/scale_jobs_test.go
+++ b/pkg/scaling/executor/scale_jobs_test.go
@@ -302,20 +302,112 @@ func TestGetPendingJobCount(t *testing.T) {
 		{PendingPodConditions: pendingPodConditions, PodStatus: v1.PodStatus{Conditions: []v1.PodCondition{readyCondition}}, PendingJobCount: 1},
 		{PendingPodConditions: pendingPodConditions, PodStatus: v1.PodStatus{Conditions: []v1.PodCondition{podScheduledCondition}}, PendingJobCount: 1},
 		{PendingPodConditions: pendingPodConditions, PodStatus: v1.PodStatus{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}}, PendingJobCount: 0},
+
+		// Test case: Job with 2 pods, both have all conditions - should NOT be pending
+		{
+			PendingPodConditions: pendingPodConditions,
+			PodStatuses: []v1.PodStatus{
+				{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}},
+				{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}},
+			},
+			PendingJobCount: 0,
+		},
+
+		// Test case: Job with 2 pods, 1 has all conditions, 1 doesn't - should NOT be pending
+		{
+			PendingPodConditions: pendingPodConditions,
+			PodStatuses: []v1.PodStatus{
+				{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}},
+				{Conditions: []v1.PodCondition{readyCondition}}, // missing PodScheduled
+			},
+			PendingJobCount: 0,
+		},
+
+		// Test case: Job with 2 pods, neither has all conditions - should be pending
+		{
+			PendingPodConditions: pendingPodConditions,
+			PodStatuses: []v1.PodStatus{
+				{Conditions: []v1.PodCondition{readyCondition}},        // missing PodScheduled
+				{Conditions: []v1.PodCondition{podScheduledCondition}}, // missing Ready
+			},
+			PendingJobCount: 1,
+		},
+
+		// Test case: Job with 3 pods, 1 has all conditions - should NOT be pending
+		{
+			PendingPodConditions: pendingPodConditions,
+			PodStatuses: []v1.PodStatus{
+				{Conditions: []v1.PodCondition{}},                                      // no conditions
+				{Conditions: []v1.PodCondition{readyCondition}},                        // partial conditions
+				{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}}, // all conditions
+			},
+			PendingJobCount: 0,
+		},
+
+		// Test case: Job with 3 pods, none have all conditions - should be pending
+		{
+			PendingPodConditions: pendingPodConditions,
+			PodStatuses: []v1.PodStatus{
+				{Conditions: []v1.PodCondition{}},                      // no conditions
+				{Conditions: []v1.PodCondition{readyCondition}},        // partial conditions
+				{Conditions: []v1.PodCondition{podScheduledCondition}}, // partial conditions
+			},
+			PendingJobCount: 1,
+		},
+		// Edge case: Empty conditions list
+		{
+			PendingPodConditions: []string{},
+			PodStatuses: []v1.PodStatus{
+				{Phase: v1.PodRunning, Conditions: []v1.PodCondition{readyCondition}},
+			},
+			PendingJobCount: 0, // No conditions to check, pod is running, so not pending
+		},
 	}
 
-	for _, testData := range testPendingJobTestData {
+	for i, testData := range testPendingJobTestData {
 		ctx := context.Background()
-		client := getMockClientForTestingPendingPods(t, ctrl, testData.PodStatus)
-		scaleExecutor := getMockScaleExecutor(client)
 
+		// Support both single pod and multiple pods test cases
+		var client *mock_client.MockClient
+		if len(testData.PodStatuses) > 0 {
+			client = getMockClientForTestingPendingPodsMultiple(t, ctrl, testData.PodStatuses)
+		} else {
+			client = getMockClientForTestingPendingPods(t, ctrl, testData.PodStatus)
+		}
+
+		scaleExecutor := getMockScaleExecutor(client)
 		scaledJob := getMockScaledJobWithPendingPodConditions(testData.PendingPodConditions)
 		result := scaleExecutor.getPendingJobCount(ctx, scaledJob)
 
-		assert.Equal(t, testData.PendingJobCount, result)
+		assert.Equal(t, testData.PendingJobCount, result, "Test case %d failed", i)
 	}
 }
 
+// Test to check issue #6727
+func TestAreAllPendingPodConditionsFulfilledBugScenario(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	pendingPodConditions := []string{"Ready", "PodScheduled"}
+	readyCondition := getPodCondition(v1.PodReady)
+	podScheduledCondition := getPodCondition(v1.PodScheduled)
+
+	podStatuses := []v1.PodStatus{
+		{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}},
+		{Conditions: []v1.PodCondition{readyCondition, podScheduledCondition}},
+	}
+
+	ctx := context.Background()
+	client := getMockClientForTestingPendingPodsMultiple(t, ctrl, podStatuses)
+	scaleExecutor := getMockScaleExecutor(client)
+	scaledJob := getMockScaledJobWithPendingPodConditions(pendingPodConditions)
+
+	result := scaleExecutor.getPendingJobCount(ctx, scaledJob)
+
+	// Should be 0, because both pods have all required conditions
+	assert.Equal(t, int64(0), result, "Job with pods having all conditions should not be pending")
+}
+
 func TestCreateJobs(t *testing.T) {
 	ctx := context.Background()
 	logger := logf.Log.WithName("CreateJobsTest")
@@ -382,7 +474,8 @@ type mockJobParameter struct {
 
 type pendingJobTestData struct {
 	PendingPodConditions []string
-	PodStatus            v1.PodStatus
+	PodStatus            v1.PodStatus   // For single pod tests
+	PodStatuses          []v1.PodStatus // For multiple pods tests
 	PendingJobCount      int64
 }
 
@@ -552,6 +645,45 @@ func getMockClientForTestingPendingPods(t *testing.T, ctrl *gomock.Controller, p
 	return client
 }
 
+// getMockClientForTestingPendingPodsMultiple returns mock client function for testing pending pod conditions with multiple pods in a job
+func getMockClientForTestingPendingPodsMultiple(t *testing.T, ctrl *gomock.Controller, podStatuses []v1.PodStatus) *mock_client.MockClient {
+	client := mock_client.NewMockClient(ctrl)
+	gomock.InOrder(
+		// listing jobs
+		client.EXPECT().
+			List(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(_ context.Context, list runtime.Object, _ ...runtimeclient.ListOption) {
+			j, ok := list.(*batchv1.JobList)
+
+			if !ok {
+				t.Error("Cast failed on batchv1.JobList at mocking client.List()")
+			}
+
+			if ok {
+				j.Items = append(j.Items, batchv1.Job{})
+			}
+		}).
+			Return(nil),
+		// listing pods - multiple pods this time
+		client.EXPECT().
+			List(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(_ context.Context, list runtime.Object, _ ...runtimeclient.ListOption) {
+			p, ok := list.(*v1.PodList)
+
+			if !ok {
+				t.Error("Cast failed on v1.PodList at mocking client.List()")
+			}
+
+			if ok {
+				for _, podStatus := range podStatuses {
+					p.Items = append(p.Items, v1.Pod{Status: podStatus})
+				}
+			}
+		}).
+			Return(nil),
+	)
+
+	return client
+}
+
 func getJob(t *testing.T, name string, completionTime string, jobConditionType batchv1.JobConditionType) *batchv1.Job {
 	parsedCompletionTime, err := time.Parse(time.RFC3339, completionTime)
 	completionTimeT := metav1.NewTime(parsedCompletionTime)