Repository: kedacore/keda

Issue:
DataDog Scaler datadogNamespace parameter is not respected
### Report

I believe that in the DataDog scaler the `datadogNamespace` parameter is not being correctly used to reference the DataDog namespace an therefor the scaler only works on the ClusterAgentProxy mode when all the DatadogMetrics objects are on the same namespace as the Datadog metrics service.

As you can see in [here](https://github.com/kedacore/keda/blob/dcfae290507c3b5936abe7dc9c75969ceaa785f9/pkg/scalers/datadog_scaler.go#L255). 


### Expected Behavior

When the URL for the metric query is being built, the correct should be to use `meta.DatadogNamespace` in here, however, `meta.DatadogMetricNamespace` is being used.
 
The URLs shoul come out like this:

https://datadog-cluster-agent-metrics-api.datadog:8443/apis/external.metrics.k8s.io/v1beta1/namespaces/kong/datadogmetric@datadog:kong-request-per-s


### Actual Behavior

They are coming out like this:

https://datadog-cluster-agent-metrics-api.kong:8443/apis/external.metrics.k8s.io/v1beta1/namespaces/kong/datadogmetric@datadog:kong-request-per-s

### Steps to Reproduce the Problem

1. Create a trigger of type datadog
2. Set datadogMetricNamespace to any value
3. Set datadogNamespace to any value that is not equal to step 2
4. Check that the URL is built wrongly


### Logs from KEDA operator

```
{"level":"error","ts":"2025-05-09T20:55:27Z","logger":"scale_handler","msg":"error getting scale decision","scaledObject.Namespace":"kong-staging","scaledObject.Name":"kong-staging-test","scaler":"datadogScaler","error":"error generating http request: %!w(<nil>)","stacktrace":"github.com/kedacore/keda/v2/pkg/scaling.(*scaleHandler).getScalerState\n\t/workspace/pkg/scaling/scale_handler.go:781\ngithub.com/kedacore/keda/v2/pkg/scaling.(*scaleHandler).getScaledObjectState.func1\n\t/workspace/pkg/scaling/scale_handler.go:634"}
```


### KEDA Version

2.17.0

### Kubernetes Version

1.31

### Platform

Amazon Web Services

### Scaler Details

Datadog

### Anything else?

_No response_

Hints from issue comments:

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d995b4282..ef6b74216 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -71,6 +71,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 
 - **General**: Improve Events emitted from ScaledObject controller ([#6802](https://github.com/kedacore/keda/issues/6802))
 - **General**: Internal gRPC connection's certificates are hot reloaded ([#6756](https://github.com/kedacore/keda/pull/6756))
+- **Datadog Scaler**: Fix bug with datadogNamespace config ([#6828](https://github.com/kedacore/keda/pull/6828))
 - **Metrics API**: Support multiple auth methods simultaneously in Metrics API scaler ([#6642](https://github.com/kedacore/keda/issues/6642))
 - **Temporal Scaler**: Support custom tlsServerName ([#6820](https://github.com/kedacore/keda/pull/6820))
 
diff --git a/pkg/scalers/datadog_scaler.go b/pkg/scalers/datadog_scaler.go
index 84e6f9eb9..d04098a67 100644
--- a/pkg/scalers/datadog_scaler.go
+++ b/pkg/scalers/datadog_scaler.go
@@ -252,7 +252,7 @@ func parseDatadogClusterAgentMetadata(config *scalersconfig.ScalerConfig, logger
 	}
 	meta.HpaMetricName = "datadogmetric@" + meta.DatadogMetricNamespace + ":" + meta.DatadogMetricName
 
-	meta.DatadogMetricServiceURL = buildClusterAgentURL(meta.DatadogMetricsService, meta.DatadogMetricNamespace, meta.DatadogMetricsServicePort)
+	meta.DatadogMetricServiceURL = buildClusterAgentURL(meta.DatadogMetricsService, meta.DatadogNamespace, meta.DatadogMetricsServicePort)
 
 	return meta, nil
 }



Test patch:
diff --git a/pkg/scalers/datadog_scaler_test.go b/pkg/scalers/datadog_scaler_test.go
index 52421d8f1..a3e16903d 100644
--- a/pkg/scalers/datadog_scaler_test.go
+++ b/pkg/scalers/datadog_scaler_test.go
@@ -2,7 +2,9 @@ package scalers
 
 import (
 	"context"
+	"fmt"
 	"slices"
+	"strings"
 	"testing"
 
 	"github.com/go-logr/logr"
@@ -120,6 +122,10 @@ var testDatadogClusterAgentMetadata = []datadogAuthMetadataTestData{
 	{"", map[string]string{"useClusterAgentProxy": "true", "datadogMetricName": "nginx-hits", "datadogMetricNamespace": "default", "targetValue": "notanint", "type": "global"}, map[string]string{"token": "token", "datadogNamespace": "datadog", "datadogMetricsService": "datadog-cluster-agent-metrics-api", "datadogMetricsServicePort": "8080", "unsafeSsl": "true", "authMode": "bearer"}, true},
 	// wrong type
 	{"", map[string]string{"useClusterAgentProxy": "true", "datadogMetricName": "nginx-hits", "datadogMetricNamespace": "default", "targetValue": "2", "type": "notatype"}, map[string]string{"token": "token", "datadogNamespace": "datadog", "datadogMetricsService": "datadog-cluster-agent-metrics-api", "datadogMetricsServicePort": "8080", "unsafeSsl": "true", "authMode": "bearer"}, true},
+	// Test case with different datadogNamespace and datadogMetricNamespace to ensure the correct namespace is used in URL
+	{"", map[string]string{"useClusterAgentProxy": "true", "datadogMetricName": "test-metric", "datadogMetricNamespace": "application-metrics", "targetValue": "10"}, map[string]string{"token": "test-token", "datadogNamespace": "datadog-system", "datadogMetricsService": "datadog-cluster-agent-metrics-api", "datadogMetricsServicePort": "8443", "authMode": "bearer"}, false},
+	// Test case with custom service name and port to verify URL building
+	{"", map[string]string{"useClusterAgentProxy": "true", "datadogMetricName": "custom-metric", "datadogMetricNamespace": "prod-metrics", "targetValue": "5"}, map[string]string{"token": "test-token", "datadogNamespace": "monitoring", "datadogMetricsService": "custom-datadog-service", "datadogMetricsServicePort": "9443", "authMode": "bearer"}, false},
 }
 
 var testDatadogAPIMetadata = []datadogAuthMetadataTestData{
@@ -182,7 +188,7 @@ func TestDatadogScalerAPIAuthParams(t *testing.T) {
 
 func TestDatadogScalerClusterAgentAuthParams(t *testing.T) {
 	for idx, testData := range testDatadogClusterAgentMetadata {
-		_, err := parseDatadogClusterAgentMetadata(&scalersconfig.ScalerConfig{TriggerMetadata: testData.metadata, AuthParams: testData.authParams, MetricType: testData.metricType}, logr.Discard())
+		meta, err := parseDatadogClusterAgentMetadata(&scalersconfig.ScalerConfig{TriggerMetadata: testData.metadata, AuthParams: testData.authParams, MetricType: testData.metricType}, logr.Discard())
 
 		if err != nil && !testData.isError {
 			t.Errorf("Expected success but got error: %s for test case %d", err, idx)
@@ -190,6 +196,32 @@ func TestDatadogScalerClusterAgentAuthParams(t *testing.T) {
 		if testData.isError && err == nil {
 			t.Errorf("Expected error but got success for test case %d", idx)
 		}
+
+		// Additional validation for URL building when we have valid metadata
+		// This validates that datadogNamespace is used correctly in URL building (issue #6769)
+		if !testData.isError && meta != nil {
+			datadogNamespace := testData.authParams["datadogNamespace"]
+			datadogMetricNamespace := testData.metadata["datadogMetricNamespace"]
+
+			if datadogNamespace != "" && datadogMetricNamespace != "" {
+				// Verify that the URL contains the service namespace (datadogNamespace), not the metric namespace
+				if !strings.Contains(meta.DatadogMetricServiceURL, datadogNamespace) {
+					t.Errorf("Test case %d: DatadogMetricServiceURL should contain datadogNamespace '%s', but got %s", idx, datadogNamespace, meta.DatadogMetricServiceURL)
+				}
+				// When namespaces are different, ensure metric namespace is NOT used in the service URL
+				if datadogNamespace != datadogMetricNamespace {
+					datadogMetricsService := testData.authParams["datadogMetricsService"]
+					datadogMetricsServicePort := testData.authParams["datadogMetricsServicePort"]
+
+					incorrectURL := fmt.Sprintf("https://%s.%s:%s/apis/external.metrics.k8s.io/v1beta1",
+						datadogMetricsService, datadogMetricNamespace, datadogMetricsServicePort)
+
+					if meta.DatadogMetricServiceURL == incorrectURL {
+						t.Errorf("Test case %d: Bug detected - DatadogMetricServiceURL incorrectly uses datadogMetricNamespace instead of datadogNamespace. Got %s", idx, meta.DatadogMetricServiceURL)
+					}
+				}
+			}
+		}
 	}
 }
 
diff --git a/tests/helper/helper.go b/tests/helper/helper.go
index c0262fa36..90d12f6f2 100644
--- a/tests/helper/helper.go
+++ b/tests/helper/helper.go
@@ -32,6 +32,9 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
@@ -149,6 +152,14 @@ func ExecuteCommandWithDir(cmdWithArgs, dir string) ([]byte, error) {
 }
 
 func ExecCommandOnSpecificPod(t *testing.T, podName string, namespace string, command string) (string, string, error) {
+	return executeCommandOnPod(t, podName, namespace, command, true)
+}
+
+func ExecCommandOnSpecificPodWithoutTTY(t *testing.T, podName string, namespace string, command string) (string, string, error) {
+	return executeCommandOnPod(t, podName, namespace, command, false)
+}
+
+func executeCommandOnPod(t *testing.T, podName string, namespace string, command string, tty bool) (string, string, error) {
 	cmd := []string{
 		"sh",
 		"-c",
@@ -164,7 +175,7 @@ func ExecCommandOnSpecificPod(t *testing.T, podName string, namespace string, co
 			Stdin:   false,
 			Stdout:  true,
 			Stderr:  true,
-			TTY:     true,
+			TTY:     tty,
 		}, scheme.ParameterCodec)
 	exec, err := remotecommand.NewSPDYExecutor(KubeConfig, "POST", request.URL())
 	assert.NoErrorf(t, err, "cannot execute command - %s", err)
@@ -913,3 +924,224 @@ func CheckKubectlGetResult(t *testing.T, kind string, name string, namespace str
 	unqoutedOutput := strings.ReplaceAll(string(output), "\"", "")
 	assert.Equal(t, expected, unqoutedOutput)
 }
+
+// KedaConsistently checks if the provided conditionFunc consistently returns true
+// (and no error) for the entire duration specified by the context's deadline.
+// It polls the conditionFunc at the given interval.
+func KedaConsistently(ctx context.Context, conditionFunc wait.ConditionWithContextFunc, interval time.Duration) error {
+	if interval <= 0 {
+		return fmt.Errorf("polling interval must be positive, got %v", interval)
+	}
+
+	ok, err := conditionFunc(ctx)
+	if err != nil {
+		return fmt.Errorf("consistency check failed on initial check: %w", err)
+	}
+	if !ok {
+		return fmt.Errorf("consistency check failed: condition was false on initial check")
+	}
+
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+
+		case <-ticker.C:
+			ok, err := conditionFunc(ctx)
+			if err != nil {
+				return fmt.Errorf("consistency check failed during polling: %w", err)
+			}
+			if !ok {
+				return fmt.Errorf("consistency check failed: condition returned false during polling period")
+			}
+		}
+	}
+}
+
+// EventWatchResult holds the outcome of the event watch.
+type EventWatchResult struct {
+	Event corev1.Event
+	Err   error
+}
+
+// StartEventWatch starts a watch for Kubernetes events matching the specified criteria.
+// It returns a channel that will receive the first matching event or an error if the watch fails.
+func StartEventWatch(
+	ctx context.Context,
+	t *testing.T,
+	clientset *kubernetes.Clientset,
+	namespace string,
+	involvedObjectName string,
+	involvedObjectKind string,
+	expectedReason string,
+	expectedType string,
+	expectedMessages []string,
+	resourceVersion string,
+) <-chan EventWatchResult {
+	t.Helper()
+	resultChan := make(chan EventWatchResult, 1)
+
+	fieldSelectorMap := fields.Set{
+		"involvedObject.name": involvedObjectName,
+		"involvedObject.kind": involvedObjectKind,
+		"reason":              expectedReason,
+		"type":                expectedType,
+	}
+	fieldSelector := fieldSelectorMap.String()
+
+	listOptions := metav1.ListOptions{
+		FieldSelector:   fieldSelector,
+		ResourceVersion: resourceVersion,
+	}
+
+	t.Logf("Starting event watch goroutine: namespace=%s, selector='%s', messages=%v, resourceVersion=%s",
+		namespace, fieldSelector, expectedMessages, resourceVersion)
+
+	watcher, err := clientset.CoreV1().Events(namespace).Watch(ctx, listOptions)
+	assert.NoErrorf(t, err, "failed to start watch for events with selector '%s' and rv %s: %s", fieldSelector, resourceVersion, err)
+	if err != nil {
+		resultChan <- EventWatchResult{Err: err}
+		close(resultChan)
+		return resultChan
+	}
+
+	go func() {
+		defer close(resultChan)
+		defer watcher.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				err := fmt.Errorf("watch context cancelled or timed out while waiting for event: %w", ctx.Err())
+				resultChan <- EventWatchResult{Err: err}
+				return
+
+			case watchEvent, ok := <-watcher.ResultChan():
+				if !ok {
+					err := fmt.Errorf("watcher channel closed unexpectedly for selector '%s'", fieldSelector)
+					if ctx.Err() != nil {
+						err = fmt.Errorf("watcher channel closed, likely due to context timeout: %w", ctx.Err())
+					}
+					resultChan <- EventWatchResult{Err: err}
+					return
+				}
+
+				switch watchEvent.Type {
+				case watch.Added, watch.Modified:
+					event, ok := watchEvent.Object.(*corev1.Event)
+					if !ok {
+						t.Logf("Received non-Event object type (%T) from watch, skipping", watchEvent.Object)
+						continue
+					}
+
+					t.Logf("Watch goroutine received event: %s/%s - %s %s (ts: %s): %s",
+						event.InvolvedObject.Kind, event.InvolvedObject.Name, event.Type, event.Reason,
+						event.LastTimestamp.Time.Format(time.RFC3339), event.Message)
+
+					messageMatched := false
+					for _, expectedMsg := range expectedMessages {
+						if strings.Contains(event.Message, expectedMsg) {
+							messageMatched = true
+							break
+						}
+					}
+					if !messageMatched {
+						t.Logf(" -> Skipping event: Message '%s' did not match any of expected %v", event.Message, expectedMessages)
+						continue
+					}
+
+					t.Logf("Watch goroutine found matching event: %s/%s - %s %s: %s",
+						event.InvolvedObject.Kind, event.InvolvedObject.Name, event.Type, event.Reason, event.Message)
+					resultChan <- EventWatchResult{Event: *event}
+					return
+
+				case watch.Error:
+					status, ok := watchEvent.Object.(*metav1.Status)
+					var errMsg string
+					if ok {
+						errMsg = fmt.Sprintf("received watch.Error status: %s", status.Message)
+					} else {
+						errMsg = fmt.Sprintf("received watch.Error with unexpected object type %T: %v", watchEvent.Object, watchEvent.Object)
+					}
+					resultChan <- EventWatchResult{Err: fmt.Errorf("%s", errMsg)}
+					return
+
+				case watch.Deleted, watch.Bookmark:
+					continue
+				}
+			}
+		}
+	}()
+
+	return resultChan
+}
+
+// WatchForEventAfterTrigger starts watching for a specific Kubernetes event,
+// executes a trigger function, and then waits for the event to appear.
+func WatchForEventAfterTrigger(
+	t *testing.T,
+	kc *kubernetes.Clientset,
+	namespace string,
+	involvedObjectName string,
+	involvedObjectKind string,
+	expectedReason string,
+	expectedType string,
+	expectedMessages []string, // matches any of the messages
+	watchTimeout time.Duration,
+	triggerAction func() error,
+) {
+	t.Helper()
+
+	watchCtx, watchCancel := context.WithTimeout(context.Background(), watchTimeout)
+	defer watchCancel()
+
+	initialListOptions := metav1.ListOptions{
+		FieldSelector: fields.Set{
+			"involvedObject.name": involvedObjectName,
+			"involvedObject.kind": involvedObjectKind,
+			"reason":              expectedReason,
+			"type":                expectedType,
+		}.String(),
+		Limit: 1,
+	}
+
+	eventsList, err := kc.CoreV1().Events(namespace).List(watchCtx, initialListOptions)
+	require.NoError(t, err, "failed to get initial resource version for watch")
+
+	// Use the resource version from the initial list to start the watch
+	// This ensures we only get events that occur after the initial list
+	initialResourceVersion := eventsList.ListMeta.ResourceVersion
+
+	t.Logf("Waiting up to %s for Kubernetes event '%s' via watch...", watchTimeout, expectedReason)
+
+	resultChan := StartEventWatch(
+		watchCtx,
+		t,
+		kc,
+		namespace,
+		involvedObjectName,
+		involvedObjectKind,
+		expectedReason,
+		expectedType,
+		expectedMessages,
+		initialResourceVersion,
+	)
+
+	err = triggerAction()
+	require.NoError(t, err, "Trigger action failed")
+
+	select {
+	case result := <-resultChan:
+		assert.NoError(t, result.Err, "Event watch failed for %s", expectedReason)
+		if result.Err == nil {
+			t.Logf("Received expected Kubernetes event via watch: %s/%s - %s %s: %s",
+				result.Event.InvolvedObject.Kind, result.Event.InvolvedObject.Name, result.Event.Type, result.Event.Reason, result.Event.Message)
+		}
+
+	case <-time.After(watchTimeout + 5*time.Second):
+		assert.Fail(t, "Timed out waiting for result from Kubernetes event watch channel")
+	}
+}
diff --git a/tests/internals/cloudevent_source/cloudevent_source_test.go b/tests/internals/cloudevent_source/cloudevent_source_test.go
index e16f964fc..e6b7b6b68 100644
--- a/tests/internals/cloudevent_source/cloudevent_source_test.go
+++ b/tests/internals/cloudevent_source/cloudevent_source_test.go
@@ -4,6 +4,7 @@
 package trigger_update_so_test
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"strings"
@@ -13,8 +14,12 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/joho/godotenv"
 	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes"
 
+	cloudevent_types "github.com/kedacore/keda/v2/apis/eventing/v1alpha1"
+	message "github.com/kedacore/keda/v2/pkg/common/message"
+	eventreason "github.com/kedacore/keda/v2/pkg/eventreason"
 	. "github.com/kedacore/keda/v2/tests/helper"
 )
 
@@ -351,6 +356,10 @@ func TestScaledObjectGeneral(t *testing.T) {
 	// Create kubernetes resources
 	kc := GetKubernetesClient(t)
 	data, templates := getTemplateData()
+	t.Cleanup(func() {
+		t.Log("--- cleaning up ---")
+		DeleteKubernetesResources(t, namespace, data, templates)
+	})
 	CreateKubernetesResources(t, kc, namespace, data, templates)
 
 	assert.True(t, WaitForAllPodRunningInNamespace(t, kc, namespace, 5, 20), "all pods should be running")
@@ -365,12 +374,10 @@ func TestScaledObjectGeneral(t *testing.T) {
 	testErrEventSourceExcludeValue(t, kc, data, false)
 	testErrEventSourceIncludeValue(t, kc, data, false)
 	testErrEventSourceCreation(t, kc, data, false)
-
-	DeleteKubernetesResources(t, namespace, data, templates)
 }
 
 // tests error events emitted
-func testErrEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data templateData, isClusterScope bool) {
+func testErrEventSourceEmitValue(t *testing.T, kc *kubernetes.Clientset, data templateData, isClusterScope bool) {
 	ceTemplate := ""
 	if isClusterScope {
 		ceTemplate = clusterCloudEventSourceTemplate
@@ -378,16 +385,33 @@ func testErrEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data tem
 		ceTemplate = cloudEventSourceTemplate
 	}
 
-	t.Log("--- test emitting eventsource about scaledobject err---")
+	t.Logf("--- test emitting eventsource about scaledobject err --- [isClusterScope: %t]", isClusterScope)
 	KubectlApplyWithTemplate(t, data, "cloudEventSourceTemplate", ceTemplate)
-	KubectlApplyWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
-
-	// wait 15 seconds to ensure event propagation
-	time.Sleep(5 * time.Second)
+	defer KubectlDeleteWithTemplate(t, data, "cloudEventSourceTemplate", ceTemplate)
+
+	WatchForEventAfterTrigger(
+		t,
+		kc,
+		namespace,
+		scaledObjectName,
+		"ScaledObject",
+		eventreason.ScaledObjectCheckFailed,
+		corev1.EventTypeWarning,
+		[]string{message.ScaleTargetErrMsg, message.ScaleTargetNotFoundMsg},
+		60*time.Second,
+		func() error {
+			KubectlApplyWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
+			return nil
+		},
+	)
+
+	// in order to satisfy lastCloudEventTime, we cannot defer this deletion because the ExecCommand would then retrieve a stale list of current events
+	// i.e. more ScaledObjectCheckFailed events will happen between the ExecCommand and the deletion of the ScaledObject
+	// hence, we have to delete it here after an arbitrary few seconds, in order to stop all events
+	// then we will calculate lastCloudEventTime which will be updated with the ACTUAL last event without any surprises
 	KubectlDeleteWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
-	time.Sleep(10 * time.Second)
 
-	out, outErr, err := ExecCommandOnSpecificPod(t, clientName, namespace, fmt.Sprintf("curl -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectCheckFailed"))
+	out, outErr, err := ExecCommandOnSpecificPodWithoutTTY(t, clientName, namespace, fmt.Sprintf("curl -s -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectCheckFailed"))
 	assert.NotEmpty(t, out)
 	assert.Empty(t, outErr)
 	assert.NoError(t, err, "dont expect error requesting ")
@@ -405,17 +429,17 @@ func testErrEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data tem
 			foundEvents = append(foundEvents, cloudEvent)
 			data := map[string]string{}
 			err := cloudEvent.DataAs(&data)
-			t.Log("--- test emitting eventsource about scaledobject err---", "message", data["message"])
+			t.Log("--- test emitting eventsource about scaledobject err ---", "message", data["message"])
 
 			assert.NoError(t, err)
 			assert.Condition(t, func() bool {
-				if strings.Contains(data["message"], "ScaledObject doesn't have correct scaleTargetRef specification") || strings.Contains(data["message"], "Target resource doesn't exist") {
+				if strings.Contains(data["message"], message.ScaleTargetErrMsg) || strings.Contains(data["message"], message.ScaleTargetNotFoundMsg) {
 					return true
 				}
 				return false
 			}, "get filtered event")
 
-			assert.Equal(t, cloudEvent.Type(), "keda.scaledobject.failed.v1")
+			assert.Equal(t, cloudEvent.Type(), string(cloudevent_types.ScaledObjectFailedType))
 			assert.Equal(t, cloudEvent.Source(), expectedSource)
 			assert.Equal(t, cloudEvent.DataContentType(), "application/json")
 
@@ -424,22 +448,33 @@ func testErrEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data tem
 			}
 		}
 	}
+
 	assert.NotEmpty(t, foundEvents)
-	KubectlDeleteWithTemplate(t, data, "cloudEventSourceTemplate", ceTemplate)
-	t.Log("--- testErrEventSourceEmitValuetestErrEventSourceEmitValuer---", "cloud event time", lastCloudEventTime)
 }
 
-func testEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data templateData) {
-	t.Log("--- test emitting eventsource about scaledobject removed---")
+func testEventSourceEmitValue(t *testing.T, kc *kubernetes.Clientset, data templateData) {
+	t.Log("--- test emitting eventsource about scaledobject removed ---")
 	KubectlApplyWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
 	KubectlApplyWithTemplate(t, data, "deploymentTemplate", deploymentTemplate)
-
-	// wait 15 seconds to ensure event propagation
-	time.Sleep(5 * time.Second)
-	KubectlDeleteWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
-	time.Sleep(10 * time.Second)
-
-	out, outErr, err := ExecCommandOnSpecificPod(t, clientName, namespace, fmt.Sprintf("curl -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectDeleted"))
+	defer KubectlDeleteWithTemplate(t, data, "deploymentTemplate", deploymentTemplate)
+
+	WatchForEventAfterTrigger(
+		t,
+		kc,
+		namespace,
+		scaledObjectName,
+		"ScaledObject",
+		eventreason.ScaledObjectDeleted,
+		corev1.EventTypeWarning,
+		[]string{message.ScaledObjectRemoved},
+		60*time.Second,
+		func() error {
+			KubectlDeleteWithTemplate(t, data, "scaledObjectTemplate", scaledObjectTemplate)
+			return nil
+		},
+	)
+
+	out, outErr, err := ExecCommandOnSpecificPodWithoutTTY(t, clientName, namespace, fmt.Sprintf("curl -s -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectDeleted"))
 	assert.NotEmpty(t, out)
 	assert.Empty(t, outErr)
 	assert.NoError(t, err, "dont expect error requesting ")
@@ -459,8 +494,8 @@ func testEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data templa
 			err := cloudEvent.DataAs(&data)
 
 			assert.NoError(t, err)
-			assert.Equal(t, data["message"], "ScaledObject was deleted")
-			assert.Equal(t, cloudEvent.Type(), "keda.scaledobject.removed.v1")
+			assert.Equal(t, data["message"], message.ScaledObjectRemoved)
+			assert.Equal(t, cloudEvent.Type(), string(cloudevent_types.ScaledObjectRemovedType))
 			assert.Equal(t, cloudEvent.Source(), expectedSource)
 			assert.Equal(t, cloudEvent.DataContentType(), "application/json")
 
@@ -469,12 +504,13 @@ func testEventSourceEmitValue(t *testing.T, _ *kubernetes.Clientset, data templa
 			}
 		}
 	}
+
 	assert.NotEmpty(t, foundEvents)
 }
 
 // tests error events not emitted by
 func testErrEventSourceExcludeValue(t *testing.T, _ *kubernetes.Clientset, data templateData, isClusterScope bool) {
-	t.Log("--- test emitting eventsource about scaledobject err with exclude filter---", "cloud event time", lastCloudEventTime)
+	t.Logf("--- test emitting eventsource about scaledobject err with exclude filter --- [isClusterScope: %t]", isClusterScope)
 
 	ceTemplate := ""
 	if isClusterScope {
@@ -484,38 +520,56 @@ func testErrEventSourceExcludeValue(t *testing.T, _ *kubernetes.Clientset, data
 	}
 
 	KubectlApplyWithTemplate(t, data, "cloudEventSourceWithExcludeTemplate", ceTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "cloudEventSourceWithExcludeTemplate", ceTemplate)
 	KubectlApplyWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
 
-	// wait 15 seconds to ensure event propagation
-	time.Sleep(15 * time.Second)
+	consistencyDuration := 30 * time.Second
+	pollingInterval := 5 * time.Second
 
-	out, outErr, err := ExecCommandOnSpecificPod(t, clientName, namespace, fmt.Sprintf("curl -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectCheckFailed"))
-	assert.NotEmpty(t, out)
-	assert.Empty(t, outErr)
-	assert.NoError(t, err, "dont expect error requesting ")
+	t.Logf("Checking consistently every %v for %v that the excluded CloudEvent does not get emitted", pollingInterval, consistencyDuration)
+	conditionFunc := func(ctx context.Context) (bool, error) {
+		out, outErr, err := ExecCommandOnSpecificPodWithoutTTY(t, clientName, namespace, fmt.Sprintf("curl -s -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectCheckFailed"))
 
-	cloudEvents := []cloudevents.Event{}
-	err = json.Unmarshal([]byte(out), &cloudEvents)
+		if err != nil {
+			return false, fmt.Errorf("command execution error: %w (stderr: %s)", err, outErr)
+		}
+		if outErr != "" {
+			return false, fmt.Errorf("command execution error: %s", outErr)
+		}
+		if out == "" {
+			return false, fmt.Errorf("command execution returned empty output")
+		}
 
-	assert.NoError(t, err, "dont expect error unmarshaling the cloudEvents")
+		var cloudEvents []cloudevents.Event
+		err = json.Unmarshal([]byte(out), &cloudEvents)
+		if err != nil {
+			return false, fmt.Errorf("cannot unmarshal cloudevents JSON '%s': %w", out, err)
+		}
 
-	for _, cloudEvent := range cloudEvents {
-		assert.Condition(t, func() bool {
+		for _, cloudEvent := range cloudEvents {
 			if cloudEvent.Subject() == expectedSubject &&
-				cloudEvent.Time().After(lastCloudEventTime) &&
-				cloudEvent.Type() == "keda.scaledobject.failed.v1" {
-				return false
+				cloudEvent.Type() == string(cloudevent_types.ScaledObjectFailedType) &&
+				cloudEvent.Time().After(lastCloudEventTime) {
+				t.Logf("found excluded event: subject=%q, type=%q, time=%s, lastCloudEventTime=%s", cloudEvent.Subject(), cloudEvent.Type(), cloudEvent.Time().Format(time.RFC3339), lastCloudEventTime.Format(time.RFC3339))
+				return false, nil
 			}
-			return true
-		}, "get filtered event")
+		}
+
+		t.Log("no excluded event found... continuing to validate consistency")
+		return true, nil
 	}
 
-	KubectlDeleteWithTemplate(t, data, "cloudEventSourceWithExcludeTemplate", ceTemplate)
+	ctx, cancel := context.WithTimeout(context.Background(), consistencyDuration)
+	defer cancel()
+
+	err := KedaConsistently(ctx, conditionFunc, pollingInterval)
+	assert.NoError(t, err, "KedaConsistently check failed: An excluded event was likely found, or a check error occurred.")
 }
 
 // tests error events in include filter
-func testErrEventSourceIncludeValue(t *testing.T, _ *kubernetes.Clientset, data templateData, isClusterScope bool) {
-	t.Log("--- test emitting eventsource about scaledobject err with include filter---")
+func testErrEventSourceIncludeValue(t *testing.T, kc *kubernetes.Clientset, data templateData, isClusterScope bool) {
+	t.Logf("--- test emitting eventsource about scaledobject err with include filter --- [isClusterScope: %t]", isClusterScope)
 
 	ceTemplate := ""
 	if isClusterScope {
@@ -523,38 +577,50 @@ func testErrEventSourceIncludeValue(t *testing.T, _ *kubernetes.Clientset, data
 	} else {
 		ceTemplate = cloudEventSourceWithIncludeTemplate
 	}
-
 	KubectlApplyWithTemplate(t, data, "cloudEventSourceWithIncludeTemplate", ceTemplate)
-	KubectlApplyWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
-
-	// wait 15 seconds to ensure event propagation
-	time.Sleep(15 * time.Second)
-
-	out, outErr, err := ExecCommandOnSpecificPod(t, clientName, namespace, fmt.Sprintf("curl -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectCheckFailed"))
+	defer KubectlDeleteWithTemplate(t, data, "cloudEventSourceWithIncludeTemplate", ceTemplate)
+	defer KubectlDeleteWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
+
+	WatchForEventAfterTrigger(
+		t,
+		kc,
+		namespace,
+		scaledObjectName,
+		"ScaledObject",
+		eventreason.ScaledObjectCheckFailed,
+		corev1.EventTypeWarning,
+		[]string{message.ScaleTargetErrMsg, message.ScaleTargetNotFoundMsg},
+		60*time.Second,
+		func() error {
+			KubectlApplyWithTemplate(t, data, "scaledObjectErrTemplate", scaledObjectErrTemplate)
+			return nil
+		},
+	)
+
+	out, outErr, err := ExecCommandOnSpecificPodWithoutTTY(t, clientName, namespace, fmt.Sprintf("curl -s -X GET %s/getCloudEvent/%s", cloudEventHTTPServiceURL, "ScaledObjectCheckFailed"))
 	assert.NotEmpty(t, out)
 	assert.Empty(t, outErr)
 	assert.NoError(t, err, "dont expect error requesting ")
 
 	cloudEvents := []cloudevents.Event{}
 	err = json.Unmarshal([]byte(out), &cloudEvents)
-
 	assert.NoError(t, err, "dont expect error unmarshaling the cloudEvents")
 
-	foundEvents := []cloudevents.Event{}
+	foundCloudEvents := []cloudevents.Event{}
 	for _, cloudEvent := range cloudEvents {
 		if cloudEvent.Subject() == expectedSubject &&
 			cloudEvent.Time().After(lastCloudEventTime) &&
-			cloudEvent.Type() == "keda.scaledobject.failed.v1" {
-			foundEvents = append(foundEvents, cloudEvent)
+			cloudEvent.Type() == string(cloudevent_types.ScaledObjectFailedType) {
+			foundCloudEvents = append(foundCloudEvents, cloudEvent)
 		}
 	}
-	assert.NotEmpty(t, foundEvents)
-	KubectlDeleteWithTemplate(t, data, "cloudEventSourceWithIncludeTemplate", ceTemplate)
+
+	assert.NotEmpty(t, foundCloudEvents)
 }
 
 // tests error event type when creation
 func testErrEventSourceCreation(t *testing.T, _ *kubernetes.Clientset, data templateData, isClusterScope bool) {
-	t.Log("--- test emitting eventsource about scaledobject err with include filter---")
+	t.Logf("--- test emitting eventsource about scaledobject err with include filter --- [isClusterScope: %t]", isClusterScope)
 
 	ceErrTemplate := ""
 	ceErrTemplate2 := ""
@@ -566,8 +632,6 @@ func testErrEventSourceCreation(t *testing.T, _ *kubernetes.Clientset, data temp
 		ceErrTemplate2 = cloudEventSourceWithErrTypeTemplate2
 	}
 
-	// KubectlDeleteWithTemplate(t, data, "cloudEventSourceTemplate", cloudEventSourceTemplate)
-
 	err := KubectlApplyWithErrors(t, data, "cloudEventSourceWithErrTypeTemplate", ceErrTemplate)
 	if isClusterScope {
 		assert.ErrorContains(t, err, `The ClusterCloudEventSource "eventsource-test-cce-err" is invalid:`)