Repository: knative-sandbox/eventing-kafka-broker
Pull request #2793 (2022-11-17 22:57:08)

Retry Only If the Status Code Is Retryable
Signed-off-by: Vikram Vuppla naga.vicky@gmail.com

Fixes #

Why:
- Currently the retry is happening regardless of the response status code is retryable or not, remove the buggy retry logic

## Proposed Changes


- :bug: Fix bug


linux-foundation-easycla[bot]:  <a href="https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/25619798/273315692/2793/#/?version=2"><img src="https://s3.amazonaws.com/cla-project-logo-prod/cla-not-signed.svg" alt="CLA Not Signed" align="left" height="28" width="328"></a><br/><br /><ul><li><a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/25619798/273315692/2793/#/?version=2' target='_blank'>:x:</a> - login: @VikramVuppla / name: Vikram Vuppla . The commit (6e06d0b6cbb27461412121fa7253429ef623f64f) is not authorized under a signed CLA. <a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/25619798/273315692/2793/#/?version=2' target='_blank'>Please click here to be authorized</a>. For further assistance with EasyCLA, <a href='https://jira.linuxfoundation.org/servicedesk/customer/portal/4' target='_blank'>please submit a support request ticket</a>.</li></ul>
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2793#" title="Author self-approved">VikramVuppla</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign psschwei for approval by writing `/assign @psschwei` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["psschwei"]} -->
knative-prow[bot]: Hi @VikramVuppla. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

knative-prow[bot]: Welcome @VikramVuppla! It looks like this is your first PR to knative-sandbox/eventing-kafka-broker 🎉

Patch:
diff --git a/control-plane/pkg/autoscaler/autoscaler_api.go b/control-plane/pkg/autoscaler/autoscaler_api.go
new file mode 100644
index 000000000..6f686c88c
--- /dev/null
+++ b/control-plane/pkg/autoscaler/autoscaler_api.go
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package autoscaler
+
+const (
+	// Autoscaling refers to the autoscaling group.
+	Autoscaling = "autoscaling.eventing.knative.dev"
+
+	// AutoscalingClassAnnotation is the annotation for the explicit class of
+	// scaler that a particular resource has opted into.
+	AutoscalingClassAnnotation = Autoscaling + "/class"
+
+	// AutoscalingClassDisabledAnnotationValue is the value of the AutoscalingClassAnnotation with
+	// disabled autoscaling.
+	AutoscalingClassDisabledAnnotationValue = "disabled"
+
+	// AutoscalingMinScaleAnnotation is the annotation to specify the minimum number of replicas to scale down to.
+	AutoscalingMinScaleAnnotation = Autoscaling + "/min-scale"
+	// AutoscalingMaxScaleAnnotation is the annotation to specify the maximum number of replicas to scale out to.
+	AutoscalingMaxScaleAnnotation = Autoscaling + "/max-scale"
+
+	// AutoscalingPollingIntervalAnnotation is the annotation that refers to the interval in seconds the autoscaler
+	// uses to poll metrics in order to inform its scaling decisions.
+	AutoscalingPollingIntervalAnnotation = Autoscaling + "/polling-interval"
+
+	// AutoscalingCooldownPeriodAnnotation is the annotation that refers to the period autoscaler waits until it
+	// scales a ConsumerGroup down.
+	AutoscalingCooldownPeriodAnnotation = Autoscaling + "/cooldown-period"
+
+	// AutoscalingLagThreshold is the annotation that refers to the lag on the current
+	// consumer group that's used for scaling (1<->N)
+	AutoscalingLagThreshold = Autoscaling + "/lag-threshold"
+
+	// AutoscalingActivationLagThreshold is the annotation that refers to the lag on the
+	// current consumer group that's used for activation (0<->1)
+	AutoscalingActivationLagThreshold = Autoscaling + "/activation-lag-threshold"
+
+	// DefaultPollingInterval is the default value for AutoscalingPollingIntervalAnnotation.
+	DefaultPollingInterval = 30
+	// DefaultCooldownPeriod is the default value for AutoscalingCooldownPeriodAnnotation.
+	DefaultCooldownPeriod = 300
+	// DefaultMinReplicaCount is the default value for AutoscalingMinScaleAnnotation
+	DefaultMinReplicaCount = 0
+	// DefaultMaxReplicaCount is the default value for AutoscalingMaxScaleAnnotation.
+	DefaultMaxReplicaCount = 50
+	// DefaultLagThreshold is the default value for AutoscalingLagThreshold.
+	DefaultLagThreshold = 10
+	// DefaultActivationLagThreshold is the default value for AutoscalingActivationLagThreshold.
+	DefaultActivationLagThreshold = 1
+)
diff --git a/control-plane/pkg/keda/keda.go b/control-plane/pkg/autoscaler/keda/keda.go
similarity index 83%
rename from control-plane/pkg/keda/keda.go
rename to control-plane/pkg/autoscaler/keda/keda.go
index 469492d91..6760fde59 100644
--- a/control-plane/pkg/keda/keda.go
+++ b/control-plane/pkg/autoscaler/keda/keda.go
@@ -22,18 +22,20 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	"knative.dev/pkg/kmeta"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler"
+	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
+
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
 const (
-	defaultKafkaLagThreshold           = 10
-	defaultKafkaActivationLagThreshold = 0
+	// AutoscalerClass is the KEDA autoscaler class.
+	AutoscalerClass = "keda.autoscaling.knative.dev"
 )
 
 func GenerateScaleTarget(cg *kafkainternals.ConsumerGroup) *kedav1alpha1.ScaleTarget {
@@ -45,16 +47,16 @@ func GenerateScaleTarget(cg *kafkainternals.ConsumerGroup) *kedav1alpha1.ScaleTa
 }
 
 func GenerateScaleTriggers(cg *kafkainternals.ConsumerGroup, triggerAuthentication *kedav1alpha1.TriggerAuthentication) ([]kedav1alpha1.ScaleTriggers, error) {
-	triggers := []kedav1alpha1.ScaleTriggers{}
+	triggers := make([]kedav1alpha1.ScaleTriggers, 0, len(cg.Spec.Template.Spec.Topics))
 	bootstrapServers := cg.Spec.Template.Spec.Configs.Configs[kafka.BootstrapServersConfigMapKey]
 	consumerGroup := cg.Spec.Template.Spec.Configs.Configs[kafka.GroupIDConfigMapKey]
 
-	lagThreshold, err := GetInt32ValueFromMap(cg.Annotations, KedaAutoscalingKafkaLagThreshold, defaultKafkaLagThreshold)
+	lagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingLagThreshold, autoscaler.DefaultLagThreshold)
 	if err != nil {
 		return nil, err
 	}
 
-	activationLagThreshold, err := GetInt32ValueFromMap(cg.Annotations, KedaAutoscalingKafkaActivationLagThreshold, defaultKafkaActivationLagThreshold)
+	activationLagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingActivationLagThreshold, autoscaler.DefaultActivationLagThreshold)
 	if err != nil {
 		return nil, err
 	}
@@ -236,25 +238,36 @@ func addAuthSecretTargetRef(parameter string, secretKeyRef v1beta1.SecretValueFr
 	return secretTargetRefs
 }
 
-func SetAutoscalingAnnotations(objannotations map[string]string) map[string]string {
-	if objannotations != nil {
-		cgannotations := map[string]string{}
-		setAnnotation(objannotations, AutoscalingClassAnnotation, cgannotations)
-		setAnnotation(objannotations, AutoscalingMinScaleAnnotation, cgannotations)
-		setAnnotation(objannotations, AutoscalingMaxScaleAnnotation, cgannotations)
-		setAnnotation(objannotations, KedaAutoscalingPollingIntervalAnnotation, cgannotations)
-		setAnnotation(objannotations, KedaAutoscalingCooldownPeriodAnnotation, cgannotations)
-		setAnnotation(objannotations, KedaAutoscalingKafkaLagThreshold, cgannotations)
-		setAnnotation(objannotations, KedaAutoscalingKafkaActivationLagThreshold, cgannotations)
-		return cgannotations
+func SetAutoscalingAnnotations(objAnnotations map[string]string) map[string]string {
+	if objAnnotations != nil {
+		cgAnnotations := make(map[string]string, len(objAnnotations))
+		setAnnotation(objAnnotations, autoscaler.AutoscalingClassAnnotation, cgAnnotations)
+		setAnnotation(objAnnotations, autoscaler.AutoscalingMinScaleAnnotation, cgAnnotations)
+		setAnnotation(objAnnotations, autoscaler.AutoscalingMaxScaleAnnotation, cgAnnotations)
+		setAnnotation(objAnnotations, autoscaler.AutoscalingPollingIntervalAnnotation, cgAnnotations)
+		setAnnotation(objAnnotations, autoscaler.AutoscalingCooldownPeriodAnnotation, cgAnnotations)
+		setAnnotation(objAnnotations, autoscaler.AutoscalingLagThreshold, cgAnnotations)
+		setAnnotation(objAnnotations, autoscaler.AutoscalingActivationLagThreshold, cgAnnotations)
+		return cgAnnotations
 	}
 	return nil
 }
 
-func setAnnotation(objannotations map[string]string, key string, cgannotations map[string]string) {
-	value, ok := objannotations[key]
+func setAnnotation(objAnnotations map[string]string, key string, cgAnnotations map[string]string) {
+	value, ok := objAnnotations[key]
 	if !ok || value == "" {
 		return
 	}
-	cgannotations[key] = value
+	cgAnnotations[key] = value
+}
+
+func IsEnabled(cg *kafkainternals.ConsumerGroup) bool {
+	v, ok := cg.GetAnnotations()[autoscaler.AutoscalingClassAnnotation]
+	if ok && v == autoscaler.AutoscalingClassDisabledAnnotationValue {
+		return false
+	}
+	if ok && len(v) > 0 && v != AutoscalerClass {
+		return false
+	}
+	return true
 }
diff --git a/control-plane/pkg/keda/resources.go b/control-plane/pkg/autoscaler/keda/resources.go
similarity index 82%
rename from control-plane/pkg/keda/resources.go
rename to control-plane/pkg/autoscaler/keda/resources.go
index 55dd6775f..de2c24433 100644
--- a/control-plane/pkg/keda/resources.go
+++ b/control-plane/pkg/autoscaler/keda/resources.go
@@ -22,14 +22,9 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
-)
 
-const (
-	defaultPollingInterval = 30
-	defaultCooldownPeriod  = 300
-	defaultMinReplicaCount = 0
-	defaultMaxReplicaCount = 50
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler"
+	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
 )
 
 var (
@@ -38,26 +33,26 @@ var (
 
 func GenerateScaledObject(obj metav1.Object, gvk schema.GroupVersionKind, scaleTarget *kedav1alpha1.ScaleTarget, triggers []kedav1alpha1.ScaleTriggers) (*kedav1alpha1.ScaledObject, error) {
 
-	cooldownPeriod, err := GetInt32ValueFromMap(obj.GetAnnotations(), KedaAutoscalingCooldownPeriodAnnotation, defaultCooldownPeriod)
+	cooldownPeriod, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingCooldownPeriodAnnotation, autoscaler.DefaultCooldownPeriod)
 	if err != nil {
 		return nil, err
 	}
-	pollingInterval, err := GetInt32ValueFromMap(obj.GetAnnotations(), KedaAutoscalingPollingIntervalAnnotation, defaultPollingInterval)
+	pollingInterval, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingPollingIntervalAnnotation, autoscaler.DefaultPollingInterval)
 	if err != nil {
 		return nil, err
 	}
-	minReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), AutoscalingMinScaleAnnotation, defaultMinReplicaCount)
+	minReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMinScaleAnnotation, autoscaler.DefaultMinReplicaCount)
 	if err != nil {
 		return nil, err
 	}
-	maxReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), AutoscalingMaxScaleAnnotation, defaultMaxReplicaCount)
+	maxReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMaxScaleAnnotation, autoscaler.DefaultMaxReplicaCount)
 	if err != nil {
 		return nil, err
 	}
 
 	return &kedav1alpha1.ScaledObject{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      generateScaledObjectName(obj),
+			Name:      GenerateScaledObjectName(obj),
 			Namespace: obj.GetNamespace(),
 			OwnerReferences: []metav1.OwnerReference{
 				*metav1.NewControllerRef(obj, gvk),
@@ -74,7 +69,7 @@ func GenerateScaledObject(obj metav1.Object, gvk schema.GroupVersionKind, scaleT
 	}, nil
 }
 
-func generateScaledObjectName(obj metav1.Object) string {
+func GenerateScaledObjectName(obj metav1.Object) string {
 	return fmt.Sprintf("so-%s", string(obj.GetUID()))
 }
 
diff --git a/control-plane/pkg/keda/annotations.go b/control-plane/pkg/keda/annotations.go
deleted file mode 100644
index 30bb38100..000000000
--- a/control-plane/pkg/keda/annotations.go
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package keda
-
-const (
-	// Autoscaling refers to the autoscaling group.
-	Autoscaling = "autoscaling.knative.dev"
-
-	// AutoscalingClassAnnotation is the annotation for the explicit class of
-	// scaler that a particular resource has opted into.
-	AutoscalingClassAnnotation = Autoscaling + "/class"
-
-	// AutoscalingMinScaleAnnotation is the annotation to specify the minimum number of replicas to scale down to.
-	AutoscalingMinScaleAnnotation = Autoscaling + "/minScale"
-	// AutoscalingMaxScaleAnnotation is the annotation to specify the maximum number of replicas to scale out to.
-	AutoscalingMaxScaleAnnotation = Autoscaling + "/maxScale"
-
-	// KEDA is Keda autoscaler.
-	KEDA = "keda.autoscaling.knative.dev"
-
-	// KedaAutoscalingPollingIntervalAnnotation is the annotation that refers to the interval in seconds KEDA
-	// uses to poll metrics in order to inform its scaling decisions.
-	KedaAutoscalingPollingIntervalAnnotation = KEDA + "/pollingInterval"
-	// KedaAutoscalingCooldownPeriodAnnotation is the annotation that refers to the period KEDA waits until it
-	// scales a Deployment down.
-	KedaAutoscalingCooldownPeriodAnnotation = KEDA + "/cooldownPeriod"
-
-	// KedaAutoscalingKafkaLagThreshold is the annotation that refers to the lag on the current consumer group that's used for scaling (1<->N)
-	KedaAutoscalingKafkaLagThreshold = KEDA + "/kafkaLagThreshold"
-
-	// KedaAutoscalingKafkaActivationLagThreshold is the annotation that refers to the lag on the current consumer group that's used for activation (0<->1)
-	KedaAutoscalingKafkaActivationLagThreshold = KEDA + "/kafkaActivationLagThreshold"
-)
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
index df9483c30..833990fdd 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -49,6 +49,7 @@ import (
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/reconciler"
 
+	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
@@ -65,7 +66,6 @@ import (
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 
-	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 )
 
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 10f1fb36b..82a43af0d 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -44,13 +44,12 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 
-	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
 	kedaclientset "knative.dev/eventing-kafka-broker/third_party/pkg/client/clientset/versioned"
 )
@@ -99,7 +98,7 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 	}
 	cg.MarkScheduleSucceeded()
 
-	if r.isKEDAEnabled(ctx, cg.GetNamespace()) {
+	if r.isKEDAEnabled(ctx, cg) {
 		if err := r.reconcileKedaObjects(ctx, cg); err != nil {
 			return cg.MarkAutoscalerFailed("AutoscalerFailed", err)
 		}
@@ -455,18 +454,18 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 			return err
 		}
 
-		triggerAuthentication, secret, err = kedafunc.GenerateTriggerAuthentication(cg, secretData)
+		triggerAuthentication, secret, err = keda.GenerateTriggerAuthentication(cg, secretData)
 		if err != nil {
 			return err
 		}
 	}
 
-	triggers, err := kedafunc.GenerateScaleTriggers(cg, triggerAuthentication)
+	triggers, err := keda.GenerateScaleTriggers(cg, triggerAuthentication)
 	if err != nil {
 		return err
 	}
 
-	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), kedafunc.GenerateScaleTarget(cg), triggers)
+	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), keda.GenerateScaleTarget(cg), triggers)
 	if err != nil {
 		return err
 	}
@@ -581,14 +580,14 @@ func (r *Reconciler) reconcileSecret(ctx context.Context, expectedSecret *corev1
 	return nil
 }
 
-func (r *Reconciler) isKEDAEnabled(ctx context.Context, namespace string) bool {
+func (r *Reconciler) isKEDAEnabled(ctx context.Context, cg *kafkainternals.ConsumerGroup) bool {
 	// TODO: code below failing unit tests with err: "panic: interface conversion: testing.ActionImpl is not testing.GetAction: missing method GetName"
 	/*if err := discovery.ServerSupportsVersion(r.KubeClient.Discovery(), keda.KedaSchemeGroupVersion); err == nil {
 		 return true
 	 }*/
 
-	if r.KafkaFeatureFlags.IsControllerAutoscalerEnabled() {
-		if _, err := r.KedaClient.KedaV1alpha1().ScaledObjects(namespace).List(ctx, metav1.ListOptions{}); err != nil {
+	if r.KafkaFeatureFlags.IsControllerAutoscalerEnabled() && keda.IsEnabled(cg) {
+		if _, err := r.KedaClient.KedaV1alpha1().ScaledObjects(cg.GetNamespace()).List(ctx, metav1.ListOptions{}); err != nil {
 			logging.FromContext(ctx).Debug("KEDA not installed, failed to list ScaledObjects")
 			return false
 		}
@@ -599,4 +598,5 @@ func (r *Reconciler) isKEDAEnabled(ctx context.Context, namespace string) bool {
 
 var (
 	_ consumergroup.Interface = &Reconciler{}
+	_ consumergroup.Finalizer = &Reconciler{}
 )
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index e06005267..70f52b24e 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -35,10 +35,10 @@ import (
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned"
 	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 
-	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 	kedaclientset "knative.dev/eventing-kafka-broker/third_party/pkg/client/clientset/versioned"
 )
 
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index 1afc5960f..c6e8c6c9f 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -39,15 +39,14 @@ import (
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned"
 	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-
-	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
 const (
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index 7f2c39538..6f9ad9879 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -133,6 +133,7 @@ public final class WebClientCloudEventSender implements CloudEventSender {
           if (isRetryableStatusCode(response.statusCode()) && retryCounter < consumerVerticleContext.getEgressConfig().getRetry()) {
             return retry(retryCounter, event);
           }
+          return Future.failedFuture(cause);
         }
 
         if (retryCounter < consumerVerticleContext.getEgressConfig().getRetry()) {


Test patch:
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 091c6b4a3..8483db71b 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -27,7 +27,8 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 )
 
 const (
@@ -70,13 +71,13 @@ var (
 	}
 
 	ConsumerGroupAnnotations = map[string]string{
-		keda.AutoscalingClassAnnotation:                 keda.KEDA,
-		keda.AutoscalingMinScaleAnnotation:              "0",
-		keda.AutoscalingMaxScaleAnnotation:              "5",
-		keda.KedaAutoscalingPollingIntervalAnnotation:   "30",
-		keda.KedaAutoscalingCooldownPeriodAnnotation:    "300",
-		keda.KedaAutoscalingKafkaLagThreshold:           "10",
-		keda.KedaAutoscalingKafkaActivationLagThreshold: "1",
+		autoscaler.AutoscalingClassAnnotation:           keda.AutoscalerClass,
+		autoscaler.AutoscalingMinScaleAnnotation:        "0",
+		autoscaler.AutoscalingMaxScaleAnnotation:        "5",
+		autoscaler.AutoscalingPollingIntervalAnnotation: "30",
+		autoscaler.AutoscalingCooldownPeriodAnnotation:  "300",
+		autoscaler.AutoscalingLagThreshold:              "10",
+		autoscaler.AutoscalingActivationLagThreshold:    "1",
 	}
 )
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
index 4cb4fddbf..2d35c8607 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
@@ -134,6 +134,68 @@ public class WebClientCloudEventSenderTest {
 
     sender.close().onSuccess(v -> context.completeNow());
   }
+  
+  @Test
+  @Timeout(value = 20000)
+  public void shouldNotRetry(final Vertx vertx, final VertxTestContext context) throws ExecutionException, InterruptedException {
+
+    final var port = 12345;
+    final var retry = 5;
+    final var event = CloudEventBuilder.v1()
+      .withId(UUID.randomUUID().toString())
+      .withSource(URI.create("/api/v1/orders"))
+      .withType("dev.knative.eventing.created")
+      .build();
+
+    final var counter = new LongAdder();
+
+    vertx.createHttpServer()
+      .requestHandler(r -> {
+        if (r.getHeader("Ce-Id").equalsIgnoreCase(event.getId())) {
+          counter.increment();
+        }
+
+        r.response().setStatusCode(400).end();
+      })
+      .listen(port, "localhost")
+      .toCompletionStage()
+      .toCompletableFuture()
+      .get();
+
+    final var sender = new WebClientCloudEventSender(
+      vertx,
+      WebClient.create(vertx),
+      "http://localhost:" + port,
+      FakeConsumerVerticleContext.get(
+        FakeConsumerVerticleContext.get().getResource(),
+        DataPlaneContract.Egress.newBuilder(FakeConsumerVerticleContext.get().getEgress())
+          .setEgressConfig(
+            DataPlaneContract.EgressConfig.newBuilder()
+              .setBackoffDelay(100L)
+              .setTimeout(1000L)
+              .setBackoffPolicy(DataPlaneContract.BackoffPolicy.Linear)
+              .setRetry(retry)
+              .build()
+          )
+          .build()
+      )
+    );
+
+    final var success = new AtomicBoolean(true);
+
+    sender.send(event)
+      .onFailure(t -> success.set(false))
+      .onSuccess(v -> success.set(true));
+
+    await().untilFalse(success);
+    await().untilAdder(counter, is(equalTo(1L)));
+
+    // Verify that after some time counter is still equal to 1.
+    Thread.sleep(10000L);
+    await().untilAdder(counter, is(equalTo(1L)));
+
+    sender.close().onSuccess(v -> context.completeNow());
+  }
 
   @Test
   @Timeout(value = 20000)
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 3b92f2aae..b74a590d8 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -270,21 +270,21 @@ function delete_chaos() {
 }
 
 function apply_sacura() {
-  ko apply ${KO_FLAGS} -Rf ./test/config/sacura/resources || return $?
+  kubectl apply -Rf ./test/config/sacura/resources || return $?
 
   kubectl wait --for=condition=ready --timeout=3m -n sacura broker/broker || return $?
   kubectl wait --for=condition=ready --timeout=3m -n sacura trigger/trigger || return $?
 
-  ko apply ${KO_FLAGS} -Rf ./test/config/sacura || return $?
+  kubectl apply -Rf ./test/config/sacura || return $?
 }
 
 function apply_sacura_sink_source() {
-  ko apply ${KO_FLAGS} -Rf ./test/config/sacura-sink-source/resources || return $?
+  kubectl apply -Rf ./test/config/sacura-sink-source/resources || return $?
 
   kubectl wait --for=condition=ready --timeout=3m -n sacura-sink-source kafkasink/sink || return $?
   kubectl wait --for=condition=ready --timeout=3m -n sacura-sink-source kafkasource/source || return $?
 
-  ko apply ${KO_FLAGS} -Rf ./test/config/sacura-sink-source || return $?
+  kubectl apply -Rf ./test/config/sacura-sink-source || return $?
 }
 
 function delete_sacura() {