Repository: knative-sandbox/eventing-kafka-broker
Pull request #2797 (2022-11-21 13:15:51)

Decouple autoscaling annotations from KEDA
Our annotations expose KEDA, I think we should avoid doing that, so
I'm removing keda from the prefix of our annotations.

In addition, it is recommended that Knative annotations follow the kebab-case
convention instead of camel case

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2797#" title="Author self-approved">pierDipi</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2797](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a03ff1d) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/204aa033a4091ad1e008bd50c6fc25e0f9cef404?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (204aa03) will **decrease** coverage by `6.16%`.
> The diff coverage is `42.85%`.

```diff
@@             Coverage Diff              @@
##               main    #2797      +/-   ##
============================================
- Coverage     61.62%   55.45%   -6.17%     
============================================
  Files           152       75      -77     
  Lines         10710     8093    -2617     
  Branches        231        0     -231     
============================================
- Hits           6600     4488    -2112     
+ Misses         3647     3281     -366     
+ Partials        463      324     -139     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ntrol-plane/pkg/reconciler/channel/v2/channelv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NoYW5uZWx2Mi5nbw==) | `70.84% <ø> (ø)` | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | `87.07% <ø> (ø)` | |
| [...ntrol-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL3RyaWdnZXJ2Mi5nbw==) | `85.95% <ø> (ø)` | |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `51.85% <42.85%> (ø)` | |
| [.../kafka/broker/dispatcher/main/ConsumerFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lckZhY3RvcnkuamF2YQ==) | | |
| [...ve/eventing/kafka/broker/core/metrics/Metrics.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvbWV0cmljcy9NZXRyaWNzLmphdmE=) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [...r/dispatcher/impl/ResponseToKafkaTopicHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZVRvS2Fma2FUb3BpY0hhbmRsZXIuamF2YQ==) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| ... and [71 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2797/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests-namespaced-broker_eventing-kafka-broker_main | deca082ad857bfd2cf1d45e31f48f24d89923c2e | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2797/reconciler-tests-namespaced-broker_eventing-kafka-broker_main/1594684399780106240) | true | `/test reconciler-tests-namespaced-broker`

[Your PR dashboard](https://prow.knative.dev/pr).

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: /test reconciler-tests-namespaced-broker
pierDipi: /cc @aavarghese 
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2797#pullrequestreview-1188627155" title="LGTM">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2797#" title="Author self-approved">pierDipi</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->

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 (


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",
 	}
 )