Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
kafka-controller panic: interface conversion: interface {} is cache.DeletedFinalStateUnknown, not *v1alpha1.Consumer
Observed a panic in kafka-controller

```
E0126 08:17:37.130870       1 runtime.go:79] Observed a panic: &runtime.TypeAssertionError{_interface:(*runtime._type)(0x1d15700), concrete:(*runtime._type)(0x1e458a0), asserted:(*runtime._type)(0x20493c0), missingMethod:""} (interface conversion: interface {} is cache.DeletedFinalStateUnknown, not *v1alpha1.Consumer)
goroutine 650 [running]:
k8s.io/apimachinery/pkg/util/runtime.logPanic({0x1d96260?, 0xc00a7161e0})
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/runtime/runtime.go:75 +0x99
k8s.io/apimachinery/pkg/util/runtime.HandleCrash({0x0, 0x0, 0xc007f15f18?})
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/runtime/runtime.go:49 +0x75
panic({0x1d96260, 0xc00a7161e0})
	/usr/lib/golang/src/runtime/panic.go:838 +0x207
knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup.enqueueConsumerGroupFromConsumer.func1({0x1e458a0?, 0xc00a46aa60?})
	/remote-source/knative-kafka-broker/app/control-plane/pkg/reconciler/consumergroup/controller.go:168 +0x85
k8s.io/client-go/tools/cache.ResourceEventHandlerFuncs.OnDelete(...)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/client-go/tools/cache/controller.go:246
k8s.io/client-go/tools/cache.(*processorListener).run.func1()
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/client-go/tools/cache/shared_informer.go:820 +0xef
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0x11?)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:157 +0x3e
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0xc000db6f38?, {0x240d1e0, 0xc0006424e0}, 0x1, 0xc0009b0000)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:158 +0xb6
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0x0?, 0x3b9aca00, 0x0, 0x0?, 0x0?)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:135 +0x89
k8s.io/apimachinery/pkg/util/wait.Until(...)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:92
k8s.io/client-go/tools/cache.(*processorListener).run(0xc000166e80?)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/client-go/tools/cache/shared_informer.go:812 +0x6b
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:75 +0x5a
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:73 +0x85
panic: interface conversion: interface {} is cache.DeletedFinalStateUnknown, not *v1alpha1.Consumer [recovered]
	panic: interface conversion: interface {} is cache.DeletedFinalStateUnknown, not *v1alpha1.Consumer

goroutine 650 [running]:
k8s.io/apimachinery/pkg/util/runtime.HandleCrash({0x0, 0x0, 0xc007f15f18?})
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/runtime/runtime.go:56 +0xd8
panic({0x1d96260, 0xc00a7161e0})
	/usr/lib/golang/src/runtime/panic.go:838 +0x207
knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup.enqueueConsumerGroupFromConsumer.func1({0x1e458a0?, 0xc00a46aa60?})
	/remote-source/knative-kafka-broker/app/control-plane/pkg/reconciler/consumergroup/controller.go:168 +0x85
k8s.io/client-go/tools/cache.ResourceEventHandlerFuncs.OnDelete(...)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/client-go/tools/cache/controller.go:246
k8s.io/client-go/tools/cache.(*processorListener).run.func1()
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/client-go/tools/cache/shared_informer.go:820 +0xef
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0x11?)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:157 +0x3e
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0xc000db6f38?, {0x240d1e0, 0xc0006424e0}, 0x1, 0xc0009b0000)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:158 +0xb6
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0x0?, 0x3b9aca00, 0x0, 0x0?, 0x0?)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:135 +0x89
k8s.io/apimachinery/pkg/util/wait.Until(...)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:92
k8s.io/client-go/tools/cache.(*processorListener).run(0xc000166e80?)
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/client-go/tools/cache/shared_informer.go:812 +0x6b
k8s.io/apimachinery/pkg/util/wait.(*Group).Start.func1()
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:75 +0x5a
created by k8s.io/apimachinery/pkg/util/wait.(*Group).Start
	/remote-source/knative-kafka-broker/app/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go:73 +0x85
```

The container restart count was 1, and the panic was at the end of the logs of the "--previous" instance, so assuming it  caused a crash.



Pull request:
Perform type assertion on conversion
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #2926

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- as per title

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index c7ae4f70c..935dd13cc 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -167,10 +167,12 @@ func ResyncOnStatefulSetChange(ctx context.Context, handle func(interface{})) {
 
 func enqueueConsumerGroupFromConsumer(enqueue func(name types.NamespacedName)) func(obj interface{}) {
 	return func(obj interface{}) {
-		c := obj.(*kafkainternals.Consumer)
-		or := c.GetConsumerGroup()
-		if or != nil {
-			enqueue(types.NamespacedName{Namespace: c.GetNamespace(), Name: or.Name})
+		c, ok := obj.(*kafkainternals.Consumer)
+		if ok {
+			or := c.GetConsumerGroup()
+			if or != nil {
+				enqueue(types.NamespacedName{Namespace: c.GetNamespace(), Name: or.Name})
+			}
 		}
 	}
 }
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 }}