Repository: knative-sandbox/eventing-kafka-broker
Pull request #2924 (2023-01-26 09:09:35)

Enforce owner references on contract ConfigMap creation
When contract configmaps are created, we don't enforce
owner references to be set when they are create which
could lead to leftovers ConfigMaps when:

- KafkaNamespaced brokers are deleted
- kafka-source-dispatcher pods are deleted (for source v2)

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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/2924#" 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":[]} -->
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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 210c445f1b9b4c100c1d2465c6d072ff6ca0be39 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2924/unit-tests_eventing-kafka-broker_main/1618536626265264128) | true | `/test unit-tests`

[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 unit-tests
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2924](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (210c445) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/fe880a192cbc8127d4a324d79a80e1390e918006?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (fe880a1) will **decrease** coverage by `5.26%`.
> The diff coverage is `0.00%`.

```diff
@@             Coverage Diff              @@
##               main    #2924      +/-   ##
============================================
- Coverage     63.85%   58.60%   -5.26%     
============================================
  Files           155       78      -77     
  Lines         10931     8269    -2662     
  Branches        231        0     -231     
============================================
- Hits           6980     4846    -2134     
+ Misses         3449     3057     -392     
+ Partials        502      366     -136     
```

| 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/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | `60.96% <0.00%> (-0.82%)` | :arrow_down: |
| [...security/secrets\_provider\_legacy\_channel\_secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0c19wcm92aWRlcl9sZWdhY3lfY2hhbm5lbF9zZWNyZXQuZ28=) | `25.00% <0.00%> (-11.93%)` | :arrow_down: |
| [...receiver/impl/handler/MethodNotAllowedHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvTWV0aG9kTm90QWxsb3dlZEhhbmRsZXIuamF2YQ==) | | |
| [...r/dispatcher/impl/ResponseToKafkaTopicHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZVRvS2Fma2FUb3BpY0hhbmRsZXIuamF2YQ==) | | |
| [...fka/broker/receiver/impl/handler/ProbeHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvUHJvYmVIYW5kbGVyLmphdmE=) | | |
| [...fka/broker/core/security/PlaintextCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvUGxhaW50ZXh0Q3JlZGVudGlhbHMuamF2YQ==) | | |
| [...cher/impl/filter/subscriptionsapi/CeSqlFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9DZVNxbEZpbHRlci5qYXZh) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...oker/dispatcher/impl/ResponseFailureException.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZUZhaWx1cmVFeGNlcHRpb24uamF2YQ==) | | |
| ... and [71 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2924?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]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2924#pullrequestreview-1270731792" title="Approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2924#" 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)~~ [aliok,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/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 60238956e..40ab5f03b 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -112,7 +112,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "source-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return source.NewController(ctx)
+				return source.NewController(ctx, watcher)
 			},
 		},
 
diff --git a/control-plane/cmd/kafka-source-controller/main.go b/control-plane/cmd/kafka-source-controller/main.go
index 0df4b6225..1a35929b8 100644
--- a/control-plane/cmd/kafka-source-controller/main.go
+++ b/control-plane/cmd/kafka-source-controller/main.go
@@ -43,7 +43,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "source-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return source.NewController(ctx)
+				return source.NewController(ctx, watcher)
 			},
 		},
 
diff --git a/control-plane/pkg/autoscaler/keda/keda.go b/control-plane/pkg/autoscaler/keda/keda.go
index 7e6cae432..a1b3f932b 100644
--- a/control-plane/pkg/autoscaler/keda/keda.go
+++ b/control-plane/pkg/autoscaler/keda/keda.go
@@ -17,14 +17,19 @@
 package keda
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
 	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
+	"knative.dev/eventing-kafka-broker/third_party/pkg/client/clientset/versioned"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler"
 	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
@@ -260,13 +265,28 @@ func setAnnotation(objAnnotations map[string]string, key string, cgAnnotations m
 	cgAnnotations[key] = value
 }
 
-func IsEnabled(cg *kafkainternals.ConsumerGroup) bool {
-	v, ok := cg.GetAnnotations()[autoscaler.AutoscalingClassAnnotation]
+func IsEnabled(ctx context.Context, features *config.KafkaFeatureFlags, client versioned.Interface, object metav1.Object) bool {
+	if !features.IsControllerAutoscalerEnabled() {
+		return false
+	}
+
+	v, ok := object.GetAnnotations()[autoscaler.AutoscalingClassAnnotation]
 	if ok && v == autoscaler.AutoscalingClassDisabledAnnotationValue {
 		return false
 	}
 	if ok && len(v) > 0 && v != AutoscalerClass {
 		return false
 	}
+
+	// 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 _, err := client.KedaV1alpha1().ScaledObjects(object.GetNamespace()).List(ctx, metav1.ListOptions{}); err != nil {
+		logging.FromContext(ctx).Debug("KEDA not installed, failed to list ScaledObjects")
+		return false
+	}
+
 	return true
 }
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 338b575c6..b4d94620e 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -153,6 +153,10 @@ func (r *Reconciler) createDataPlaneConfigMap(ctx context.Context) (*corev1.Conf
 		},
 	}
 
+	if r.DataPlaneConfigMapTransformer != nil {
+		r.DataPlaneConfigMapTransformer(cm)
+	}
+
 	return r.KubeClient.CoreV1().ConfigMaps(r.DataPlaneConfigMapNamespace).Create(ctx, cm, metav1.CreateOptions{})
 }
 
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index e9dedcb5f..f7cef59a5 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -34,12 +34,13 @@ import (
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/utils/pointer"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
 
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+
 	"knative.dev/eventing/pkg/scheduler"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
@@ -112,7 +113,7 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 	}
 	cg.MarkScheduleSucceeded()
 
-	if r.isKEDAEnabled(ctx, cg) {
+	if keda.IsEnabled(ctx, r.KafkaFeatureFlags, r.KedaClient, cg) {
 		if err := r.reconcileKedaObjects(ctx, cg); err != nil {
 			return cg.MarkAutoscalerFailed("AutoscalerFailed", err)
 		}
@@ -619,22 +620,6 @@ func (r *Reconciler) reconcileSecret(ctx context.Context, expectedSecret *corev1
 	return nil
 }
 
-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() && 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
-		}
-		return true
-	}
-	return false
-}
-
 func (r Reconciler) ensureContractConfigmapsExist(ctx context.Context, scheduler Scheduler) error {
 	selector := labels.SelectorFromSet(map[string]string{"app": scheduler.StatefulSetName})
 	pods, err := r.PodLister.
@@ -677,7 +662,7 @@ func (r Reconciler) ensureContractConfigMapExists(ctx context.Context, p *corev1
 		DataPlaneConfigMapTransformer: base.PodOwnerReference(p),
 	}
 
-	if _, err := b.GetOrCreateDataPlaneConfigMap(ctx); err != nil && apierrors.IsAlreadyExists(err) {
+	if _, err := b.GetOrCreateDataPlaneConfigMap(ctx); err != nil && !apierrors.IsAlreadyExists(err) {
 		return fmt.Errorf("failed to create ConfigMap %s/%s: %w", r.SystemNamespace, name, err)
 	}
 	return nil
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor.go b/control-plane/pkg/reconciler/consumergroup/evictor.go
index 9ca0a573e..185801a42 100644
--- a/control-plane/pkg/reconciler/consumergroup/evictor.go
+++ b/control-plane/pkg/reconciler/consumergroup/evictor.go
@@ -134,7 +134,7 @@ func (e *evictor) disablePodScheduling(logger *zap.Logger, pod *corev1.Pod) erro
 }
 
 func removePlacement(before []eventingduckv1alpha1.Placement, toRemove *eventingduckv1alpha1.Placement) []eventingduckv1alpha1.Placement {
-	after := make([]eventingduckv1alpha1.Placement, 0, len(before)-1)
+	after := make([]eventingduckv1alpha1.Placement, 0, len(before))
 	for _, p := range before {
 		if p.PodName != toRemove.PodName {
 			after = append(after, p)
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
index 70ff59a87..fb8fb503e 100644
--- a/control-plane/pkg/reconciler/source/controller.go
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -20,8 +20,10 @@ import (
 	"context"
 
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
 
@@ -34,7 +36,7 @@ import (
 	kedaclient "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client"
 )
 
-func NewController(ctx context.Context) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
 
 	kafkaInformer := kafkainformer.Get(ctx)
 	consumerGroupInformer := consumergroupinformer.Get(ctx)
@@ -45,12 +47,19 @@ func NewController(ctx context.Context) *controller.Impl {
 		ConsumerGroupLister: consumerGroupInformer.Lister(),
 		InternalsClient:     consumergroupclient.Get(ctx),
 		KedaClient:          kedaclient.Get(ctx),
+		KafkaFeatureFlags:   config.DefaultFeaturesConfig(),
 	}
 
 	impl := kafkasource.NewImpl(ctx, r)
 
 	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
+	configStore := config.NewStore(ctx, func(name string, value *config.KafkaFeatureFlags) {
+		r.KafkaFeatureFlags.Reset(value)
+		impl.GlobalResync(kafkaInformer.Informer())
+	})
+	configStore.WatchConfigs(watcher)
+
 	// ConsumerGroup changes and enqueue associated KafkaSource
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: consumergroup.Filter("kafkasource"),
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index be8aff379..f93e20ee9 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -26,16 +26,18 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/utils/pointer"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/reconciler"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+
 	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"
+	"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"
 
@@ -59,6 +61,7 @@ type Reconciler struct {
 	ConsumerGroupLister internalslst.ConsumerGroupLister
 	InternalsClient     internalsclient.Interface
 	KedaClient          kedaclientset.Interface
+	KafkaFeatureFlags   *config.KafkaFeatureFlags
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
@@ -111,6 +114,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 			},
 		},
 		Spec: internalscg.ConsumerGroupSpec{
+			Replicas: ks.Spec.Consumers,
 			Template: internalscg.ConsumerTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
@@ -153,14 +157,18 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 	}
 
 	// TODO: make keda annotation values configurable and maybe unexposed
-	expectedCg.Annotations = kedafunc.SetAutoscalingAnnotations(ks.Annotations)
+	expectedCg.Annotations = keda.SetAutoscalingAnnotations(ks.Annotations)
+
+	// If KEDA is enabled, then we ignore KafkaSource replicas setting
+	if keda.IsEnabled(ctx, r.KafkaFeatureFlags, r.KedaClient, ks) {
+		expectedCg.Spec.Replicas = nil
+	}
 
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(ks.GetNamespace()).Get(string(ks.UID)) //Get by consumer group id
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, err
 	}
 	if apierrors.IsNotFound(err) {
-		expectedCg.Spec.Replicas = ks.Spec.Consumers
 		cg, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(expectedCg.GetNamespace()).Create(ctx, expectedCg, metav1.CreateOptions{})
 		if err != nil && !apierrors.IsAlreadyExists(err) {
 			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", expectedCg.GetNamespace(), expectedCg.GetName(), err)
@@ -180,11 +188,6 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 	}
 	newCg.Annotations = expectedCg.Annotations
 
-	// If KEDA is not enabled, then we must update the ConsumerGroup replicas to match KafkaSource consumers
-	if !r.isKEDAEnabled(ctx, cg.GetNamespace()) {
-		newCg.Spec.Replicas = ks.Spec.Consumers
-	}
-
 	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
 		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
 	}
@@ -217,15 +220,3 @@ func propagateConsumerGroupStatus(cg *internalscg.ConsumerGroup, ks *sources.Kaf
 		ks.Status.Consumers = *cg.Status.Replicas
 	}
 }
-
-func (r *Reconciler) isKEDAEnabled(ctx context.Context, namespace string) 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 _, err := r.KedaClient.KedaV1alpha1().ScaledObjects(namespace).List(ctx, metav1.ListOptions{}); err == nil {
-		return true
-	}
-	return false
-}


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
index 1aabe56b8..848f90259 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/utils/pointer"
+
 	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
@@ -163,7 +164,17 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					WithNamespacedBrokerOwnerRef,
 					WithNamespacedLabel,
 				),
-				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil,
+					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
+					WithConfigmapOwnerRef(&metav1.OwnerReference{
+						APIVersion:         eventing.SchemeGroupVersion.String(),
+						Kind:               "Broker",
+						Name:               BrokerName,
+						UID:                BrokerUUID,
+						Controller:         pointer.Bool(false),
+						BlockOwnerDeletion: pointer.Bool(true),
+					}),
+				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(BrokerNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat,
@@ -262,7 +273,17 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil,
+					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
+					WithConfigmapOwnerRef(&metav1.OwnerReference{
+						APIVersion:         eventing.SchemeGroupVersion.String(),
+						Kind:               "Broker",
+						Name:               BrokerName,
+						UID:                BrokerUUID,
+						Controller:         pointer.Bool(false),
+						BlockOwnerDeletion: pointer.Bool(true),
+					}),
+				),
 			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
diff --git a/control-plane/pkg/reconciler/consumer/consumer_test.go b/control-plane/pkg/reconciler/consumer/consumer_test.go
index 6f3e1e9b2..5c4703be9 100644
--- a/control-plane/pkg/reconciler/consumer/consumer_test.go
+++ b/control-plane/pkg/reconciler/consumer/consumer_test.go
@@ -93,7 +93,7 @@ func TestReconcileKind(t *testing.T) {
 				patchFinalizers(),
 			},
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte("")),
+				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte(""), DispatcherPodAsOwnerReference("p1")),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(SystemNamespace, "p1", base.Json, &contract.Contract{
@@ -195,7 +195,7 @@ func TestReconcileKind(t *testing.T) {
 				patchFinalizers(),
 			},
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte("")),
+				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte(""), DispatcherPodAsOwnerReference("p1")),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(SystemNamespace, "p1", base.Json, &contract.Contract{
@@ -297,7 +297,7 @@ func TestReconcileKind(t *testing.T) {
 				patchFinalizers(),
 			},
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte("")),
+				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte(""), DispatcherPodAsOwnerReference("p1")),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(SystemNamespace, "p1", base.Json, &contract.Contract{
@@ -405,7 +405,7 @@ func TestReconcileKind(t *testing.T) {
 				patchFinalizers(),
 			},
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte("")),
+				NewConfigMapWithBinaryData(SystemNamespace, "p1", []byte(""), DispatcherPodAsOwnerReference("p1")),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(SystemNamespace, "p1", base.Json, &contract.Contract{
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 8c439edde..53053d973 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -29,11 +29,12 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
 	"k8s.io/utils/pointer"
-	bindings "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	"knative.dev/pkg/apis"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 
+	bindings "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
@@ -190,8 +191,8 @@ func TestReconcileKind(t *testing.T) {
 				}),
 			},
 			WantCreates: []runtime.Object{
-				NewConfigMapWithBinaryData(systemNamespace, "p1", nil),
-				NewConfigMapWithBinaryData(systemNamespace, "p2", nil),
+				NewConfigMapWithBinaryData(systemNamespace, "p1", nil, DispatcherPodAsOwnerReference("p1")),
+				NewConfigMapWithBinaryData(systemNamespace, "p2", nil, DispatcherPodAsOwnerReference("p2")),
 				NewConsumer(1,
 					ConsumerSpec(NewConsumerSpec(
 						ConsumerTopics("t1", "t2"),
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor_test.go b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
index f587e4f6a..98fa5d114 100644
--- a/control-plane/pkg/reconciler/consumergroup/evictor_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
@@ -76,6 +76,40 @@ func TestEvictorEvictSuccess(t *testing.T) {
 	require.Nil(t, err)
 }
 
+func TestEvictorNoEvictionEmptyPlacement(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+
+	pod1 := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns1", Name: "name1"},
+	}
+
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements:          []eventingduckv1alpha1.Placement{},
+			}},
+		},
+	}
+	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
+	cg.MarkScheduleSucceeded()
+
+	placement := &eventingduckv1alpha1.Placement{PodName: pod1.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.Nil(t, err)
+}
+
 func TestEvictorNoEviction(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
diff --git a/control-plane/pkg/reconciler/source/controller_test.go b/control-plane/pkg/reconciler/source/controller_test.go
index a3deb62dd..3ccfefb94 100644
--- a/control-plane/pkg/reconciler/source/controller_test.go
+++ b/control-plane/pkg/reconciler/source/controller_test.go
@@ -30,6 +30,7 @@ import (
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	"knative.dev/pkg/configmap"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
@@ -64,7 +65,11 @@ func TestNewController(t *testing.T) {
 
 	dynamicclient.With(ctx, dynamicScheme)
 
-	controller := NewController(ctx)
+	controller := NewController(ctx, configmap.NewStaticWatcher(&corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "config-kafka-features",
+		},
+	}))
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 3827a6a60..066e2b1ce 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -26,11 +26,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgotesting "k8s.io/client-go/testing"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
-	bindings "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	cm "knative.dev/pkg/configmap/testing"
 	"knative.dev/pkg/kmeta"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	bindings "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	configapis "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
+
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -959,17 +962,152 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - update cg replicas",
+			Objects: []runtime.Object{
+				NewSource(WithSourceConsumers(3)),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT0.3S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(2),
+					ConsumerGroupReplicasStatus(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(SourceUUID),
+						WithConsumerGroupNamespace(SourceNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+						WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+						WithConsumerGroupLabels(ConsumerSourceLabel),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(SourceConsumerGroup),
+								ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							),
+							ConsumerAuth(NewConsumerSpecAuth()),
+							ConsumerDelivery(
+								NewConsumerSpecDelivery(
+									internals.Ordered,
+									NewConsumerTimeout("PT600S"),
+									NewConsumerRetry(10),
+									NewConsumerBackoffDelay("PT0.3S"),
+									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								),
+							),
+							ConsumerSubscriber(NewSourceSinkReference()),
+							ConsumerReply(ConsumerNoReply()),
+						)),
+						ConsumerGroupReplicas(3),
+						ConsumerGroupReplicasStatus(1),
+						ConsumerGroupReady,
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithSourceConsumers(3),
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+						StatusSourceConsumerGroupReplicas(1),
+						StatusSourceSelector(),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - ignore source replicas when KEDA is enabled",
+			Objects: []runtime.Object{
+				NewSource(WithSourceConsumers(3)),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT0.3S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(2),
+					ConsumerGroupReplicasStatus(1),
+					ConsumerGroupReady,
+				),
+			},
+			WithReactors: []clientgotesting.ReactionFunc{
+				ReactorKEDAEnabled(),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithSourceConsumers(3),
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+						StatusSourceConsumerGroupReplicas(1),
+						StatusSourceSelector(),
+					),
+				},
+			},
+		},
 	}
 
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
-		ctx, _ = kedaclient.With(ctx)
+		store := configapis.NewStore(ctx)
+		_, exampleConfig := cm.ConfigMapsFromTestFile(t, configapis.FlagsConfigName)
+		store.OnConfigChanged(exampleConfig)
 
 		reconciler := &Reconciler{
 			ConsumerGroupLister: listers.GetConsumerGroupLister(),
 			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
 			KedaClient:          kedaclient.Get(ctx),
+			KafkaFeatureFlags:   configapis.DefaultFeaturesConfig(),
 		}
 
+		reconciler.KafkaFeatureFlags = configapis.FromContext(store.ToContext(ctx))
+
 		r := eventingkafkasourcereconciler.NewReconciler(
 			ctx,
 			logging.FromContext(ctx),
diff --git a/control-plane/pkg/reconciler/source/testdata/config-kafka-features.yaml b/control-plane/pkg/reconciler/source/testdata/config-kafka-features.yaml
new file mode 100644
index 000000000..d0057f72b
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/testdata/config-kafka-features.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-features
+  namespace: knative-eventing
+data:
+  _example: |
+    controller.autoscaler: "enabled"
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index b27fafaaf..5237ab9f5 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -39,6 +39,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 
 	fakekafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	fakekeda "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client/fake"
 )
 
 const (
@@ -60,6 +61,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 		ctx, eventingKafkaBrokerClient := fakeeventingkafkabrokerclient.With(ctx, listers.GetEventingKafkaBrokerObjects()...)
 		ctx, kubeClient := fakekubeclient.With(ctx, listers.GetKubeObjects()...)
 		ctx, kafkaInternalsClient := fakekafkainternals.With(ctx, listers.GetKafkaInternalsObjects()...)
+		ctx, kedaClient := fakekeda.With(ctx, []runtime.Object{}...)
 
 		ctx, dynamicClient := fakedynamicclient.With(ctx,
 			newScheme(),
@@ -95,6 +97,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 			eventingClient.PrependReactor("*", "*", reactor)
 			eventingKafkaBrokerClient.PrependReactor("*", "*", reactor)
 			kafkaInternalsClient.PrependReactor("*", "*", reactor)
+			kedaClient.PrependReactor("*", "*", reactor)
 		}
 
 		actionRecorderList := ActionRecorderList{
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 34e391102..88605d227 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -25,9 +25,10 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/types"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	corev1 "k8s.io/api/core/v1"
@@ -686,3 +687,12 @@ func DataPlaneConfigInitialOffset(key string, offset sources.Offset) reconcilert
 		cm.Data[key] = props.String()
 	}
 }
+
+func ReactorKEDAEnabled() clientgotesting.ReactionFunc {
+	return func(action clientgotesting.Action) (handled bool, ret runtime.Object, err error) {
+		if action.Matches("list", "scaledobjects") {
+			return true, nil, nil
+		}
+		return false, nil, nil
+	}
+}
diff --git a/test/e2e_new/kafka_source_test.go b/test/e2e_new/kafka_source_test.go
index 01b81d533..f502bc7df 100644
--- a/test/e2e_new/kafka_source_test.go
+++ b/test/e2e_new/kafka_source_test.go
@@ -62,3 +62,19 @@ func TestKafkaSourceDeletedFromContractConfigMaps(t *testing.T) {
 	env.Test(ctx, t, features.SetupAndCleanupKafkaSources("x-kafka-source-", 42))
 	env.Test(ctx, t, features.KafkaSourcesAreNotPresentInContractConfigMaps("x-kafka-source-"))
 }
+
+func TestKafkaSourceScale(t *testing.T) {
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+	t.Cleanup(env.Finish)
+
+	env.Test(ctx, t, features.ScaleKafkaSource())
+}
diff --git a/test/rekt/features/kafka_source.go b/test/rekt/features/kafka_source.go
index 5bf0530e8..c3a8cade2 100644
--- a/test/rekt/features/kafka_source.go
+++ b/test/rekt/features/kafka_source.go
@@ -24,8 +24,6 @@ import (
 
 	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
-	kafkaclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/reconciler-test/pkg/environment"
@@ -33,6 +31,9 @@ import (
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+	kafkaclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
+
 	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasource"
 	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
 
@@ -145,3 +146,32 @@ func (r ResourceError) Error() string {
 	bytes, _ := json.Marshal(r)
 	return string(bytes)
 }
+
+func ScaleKafkaSource() *feature.Feature {
+	f := feature.NewFeatureNamed("scale KafkaSource")
+
+	replicas := int32(3)
+	source := feature.MakeRandomK8sName("kafkasource")
+	topicName := feature.MakeRandomK8sName("scale-topic")
+	sink := feature.MakeRandomK8sName("sink")
+
+	f.Setup("install a sink", svc.Install(sink, "app", "rekt"))
+	f.Setup("install kafka topic", kafkatopic.Install(topicName))
+	f.Setup("scale kafkasource", kafkasource.Install(source,
+		kafkasource.WithBootstrapServers(testingpkg.BootstrapServersPlaintextArr),
+		kafkasource.WithTopics([]string{topicName}),
+		kafkasource.WithSink(svc.AsKReference(sink), ""),
+		kafkasource.WithAnnotations(map[string]string{
+			// Disable autoscaling for this KafkaSource since we want to have the expected replicas
+			// in the status reflected without the autoscaler intervention.
+			"autoscaling.eventing.knative.dev/class": "disabled",
+		}),
+		kafkasource.WithConsumers(replicas),
+	))
+
+	f.Requirement("kafkasource is ready", kafkasource.IsReady(source))
+
+	f.Assert("kafkasource is scaled", kafkasource.VerifyScale(source, replicas))
+
+	return f
+}
diff --git a/test/rekt/resources/kafkasource/kafkasource.go b/test/rekt/resources/kafkasource/kafkasource.go
index 1f44c5a42..01ad8a920 100644
--- a/test/rekt/resources/kafkasource/kafkasource.go
+++ b/test/rekt/resources/kafkasource/kafkasource.go
@@ -19,13 +19,21 @@ package kafkasource
 import (
 	"context"
 	"embed"
+	"encoding/json"
+	"fmt"
 	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
+
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+	kafkaclientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
 )
 
 //go:embed *.yaml
@@ -51,6 +59,46 @@ func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	}
 }
 
+func VerifyScale(name string, replicas int32) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		interval, timeout := environment.PollTimingsFromContext(ctx)
+		last := &sources.KafkaSource{}
+		err := wait.PollImmediate(interval, timeout, func() (done bool, err error) {
+			ks, err := kafkaclientset.Get(ctx).
+				SourcesV1beta1().
+				KafkaSources(environment.FromContext(ctx).Namespace()).
+				Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			last = ks
+
+			if *ks.Spec.Consumers != replicas {
+				return false, fmt.Errorf("spec.consumers wanted %d, got %d", replicas, *ks.Spec.Consumers)
+			}
+
+			if ks.Status.Consumers != replicas {
+				return false, nil
+			}
+
+			count := int32(0)
+			for _, p := range ks.Status.Placements {
+				count += p.VReplicas
+			}
+
+			if count != replicas {
+				return false, nil
+			}
+
+			return true, nil
+		})
+		if err != nil {
+			bytes, _ := json.MarshalIndent(last, "", "  ")
+			t.Errorf("failed to verify kafkasource scale: %w, last state:\n%s\n", err, string(bytes))
+		}
+	}
+}
+
 // IsReady tests to see if a KafkaSource becomes ready within the time given.
 func IsReady(name string, timings ...time.Duration) feature.StepFn {
 	return k8s.IsReady(GVR(), name, timings...)
@@ -65,6 +113,13 @@ func WithVersion(version string) manifest.CfgFn {
 	}
 }
 
+// WithConsumers adds consumers to a KafkaSource spec.
+func WithConsumers(consumers int32) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["consumers"] = consumers
+	}
+}
+
 // WithAnnotations adds annotation to a KafkaSource metadata.
 func WithAnnotations(annotations map[string]string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
diff --git a/test/rekt/resources/kafkasource/kafkasource.yaml b/test/rekt/resources/kafkasource/kafkasource.yaml
index 2c8673cfd..9813d2d1c 100644
--- a/test/rekt/resources/kafkasource/kafkasource.yaml
+++ b/test/rekt/resources/kafkasource/kafkasource.yaml
@@ -33,6 +33,9 @@ spec:
   {{ if .consumerGroup }}
   consumerGroup: {{ .consumerGroup }}
   {{ end }}
+  {{ if .consumers }}
+  consumers: {{ .consumers }}
+  {{ end }}
   {{ if .topics }}
   topics:
     {{ range $value := .topics }}