Repository: knative-sandbox/eventing-kafka-broker
Pull request #2654 (2022-09-15 22:20:28)

Create a feature flag for KEDA autoscaling of consumers
Signed-off-by: aavarghese <avarghese@us.ibm.com>

## Proposed Changes

- Setup a feature flag for autoscaling consumers with KEDA


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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["dprotaso"]} -->
aavarghese: /cc @pierDipi 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2654](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (bb277ce) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/17b742d8a648bdf941814f4b495fc7f59ce81ec6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (17b742d) will **decrease** coverage by `6.14%`.
> The diff coverage is `40.00%`.

> :exclamation: Current head bb277ce differs from pull request most recent head 7690677. Consider uploading reports for the commit 7690677 to get more accurate results

```diff
@@             Coverage Diff              @@
##               main    #2654      +/-   ##
============================================
- Coverage     61.99%   55.85%   -6.15%     
============================================
  Files           148       75      -73     
  Lines         10391     7862    -2529     
  Branches        229        0     -229     
============================================
- Hits           6442     4391    -2051     
+ Misses         3500     3154     -346     
+ Partials        449      317     -132     
```

| 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/2654?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `70.84% <ø> (ø)` | |
| [...l-plane/pkg/reconciler/consumergroup/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnRyb2xsZXIuZ28=) | `56.57% <33.33%> (-1.25%)` | :arrow_down: |
| [control-plane/pkg/apis/config/features.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb25maWcvZmVhdHVyZXMuZ28=) | `87.30% <50.00%> (-0.84%)` | :arrow_down: |
| [...ting/kafka/broker/core/utils/CollectionsUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29sbGVjdGlvbnNVdGlscy5qYXZh) | | |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | | |
| [...ive/eventing/kafka/broker/core/utils/Shutdown.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvU2h1dGRvd24uamF2YQ==) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...ka/broker/core/reconciler/ResourcesReconciler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9SZXNvdXJjZXNSZWNvbmNpbGVyLmphdmE=) | | |
| [...enting/kafka/broker/core/OrderedAsyncExecutor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvT3JkZXJlZEFzeW5jRXhlY3V0b3IuamF2YQ==) | | |
| [...tive/eventing/kafka/broker/receiver/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL01haW4uamF2YQ==) | | |
| ... and [67 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2654/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

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

aslom: /lgtm
knative-prow[bot]: New changes are detected. LGTM label has been removed.

Patch:
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 7a012145c..60238956e 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -120,7 +120,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "consumergroup-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return consumergroup.NewController(ctx)
+				return consumergroup.NewController(ctx, watcher)
 			},
 		},
 
diff --git a/control-plane/cmd/kafka-source-controller/main.go b/control-plane/cmd/kafka-source-controller/main.go
index 52073cba9..0df4b6225 100644
--- a/control-plane/cmd/kafka-source-controller/main.go
+++ b/control-plane/cmd/kafka-source-controller/main.go
@@ -51,7 +51,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "consumergroup-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return consumergroup.NewController(ctx)
+				return consumergroup.NewController(ctx, watcher)
 			},
 		},
 
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
index 8541d67fa..3e7a5b9c0 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
@@ -4,7 +4,7 @@ metadata:
   name: config-kafka-features
   namespace: knative-eventing
   annotations:
-    knative.dev/example-checksum: "a4fb44fb"
+    knative.dev/example-checksum: "76de22c2"
 data:
   _example: |-
     ################################
@@ -29,5 +29,10 @@ data:
     # 1. Enabled: The metrics are recorded.
     # 2. Disabled: The metrics are not recorded.
     dispatcher.ordered-executor-metrics: "disabled"
+    # Controls whether the controller should autoscale consumer resources with KEDA
+    # 1. Enabled: KEDA autoscaling of consumers will be setup.
+    # 2. Disabled: KEDA autoscaling of consumers will not be setup.
+    controller.autoscale: "disabled"
   dispatcher.rate-limiter: "disabled"
   dispatcher.ordered-executor-metrics: "disabled"
+  controller.autoscaler: "disabled"
diff --git a/control-plane/pkg/apis/config/features.go b/control-plane/pkg/apis/config/features.go
index 3292855a2..26e04a191 100644
--- a/control-plane/pkg/apis/config/features.go
+++ b/control-plane/pkg/apis/config/features.go
@@ -34,6 +34,7 @@ const (
 type features struct {
 	DispatcherRateLimiter            feature.Flag
 	DispatcherOrderedExecutorMetrics feature.Flag
+	ControllerAutoscaler             feature.Flag
 }
 
 type KafkaFeatureFlags struct {
@@ -46,6 +47,7 @@ func DefaultFeaturesConfig() *KafkaFeatureFlags {
 		features: features{
 			DispatcherRateLimiter:            feature.Disabled,
 			DispatcherOrderedExecutorMetrics: feature.Disabled,
+			ControllerAutoscaler:             feature.Disabled,
 		},
 	}
 }
@@ -56,6 +58,7 @@ func newFeaturesConfigFromMap(cm *corev1.ConfigMap) (*KafkaFeatureFlags, error)
 	err := configmap.Parse(cm.Data,
 		asFlag("dispatcher.rate-limiter", &nc.features.DispatcherRateLimiter),
 		asFlag("dispatcher.ordered-executor-metrics", &nc.features.DispatcherOrderedExecutorMetrics),
+		asFlag("controller.autoscaler", &nc.features.ControllerAutoscaler),
 	)
 	return nc, err
 }
@@ -74,6 +77,10 @@ func (f *KafkaFeatureFlags) IsDispatcherOrderedExecutorMetricsEnabled() bool {
 	return f.features.DispatcherOrderedExecutorMetrics == feature.Enabled
 }
 
+func (f *KafkaFeatureFlags) IsControllerAutoscalerEnabled() bool {
+	return f.features.ControllerAutoscaler == feature.Enabled
+}
+
 // Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
 // +k8s:deepcopy-gen=false
 type Store struct {
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index c6ba0902d..e3f39e36f 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -41,6 +41,7 @@ import (
 
 	"knative.dev/eventing/pkg/scheduler"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
@@ -77,6 +78,8 @@ type Reconciler struct {
 	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
+
+	KafkaFeatureFlags *config.KafkaFeatureFlags
 }
 
 func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index 87d87ab65..f56cdac4d 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -39,12 +39,14 @@ import (
 	"knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
 	nodeinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
 
 	statefulsetscheduler "knative.dev/eventing/pkg/scheduler/statefulset"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer"
@@ -79,7 +81,7 @@ type SchedulerConfig struct {
 	DeSchedulerPolicy *scheduler.SchedulerPolicy
 }
 
-func NewController(ctx context.Context) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
 	logger := logging.FromContext(ctx)
 
 	env := &envConfig{}
@@ -111,13 +113,21 @@ func NewController(ctx context.Context) *controller.Impl {
 		InitOffsetsFunc:            offset.InitOffsets,
 		SystemNamespace:            system.Namespace(),
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		KafkaFeatureFlags:          config.DefaultFeaturesConfig(),
 	}
 
-	impl := cgreconciler.NewImpl(ctx, r)
 	consumerInformer := consumer.Get(ctx)
 
 	consumerGroupInformer := consumergroup.Get(ctx)
 
+	impl := cgreconciler.NewImpl(ctx, r)
+
+	configStore := config.NewStore(ctx, func(name string, value *config.KafkaFeatureFlags) {
+		r.KafkaFeatureFlags.Reset(value)
+		impl.GlobalResync(consumerGroupInformer.Informer())
+	})
+	configStore.WatchConfigs(watcher)
+
 	consumerGroupInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 	consumerInformer.Informer().AddEventHandler(controller.HandleAll(enqueueConsumerGroupFromConsumer(impl.EnqueueKey)))
 
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index 764f3e2bf..06bfae172 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -125,15 +125,16 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       final var oldEgress = cachedEgress.getKey();
 
       if (resourceEquals(newResource, oldResource) && egressEquals(newEgress, oldEgress)) {
-        logger.info("Nothing changed for egress {} {} {} {}",
-          keyValue("id", newEgress.getUid()),
-          keyValue("consumerGroup", newEgress.getConsumerGroup()),
-          keyValue("destination", newEgress.getDestination()),
-          keyValue("contractGeneration", generation)
-        );
         return;
       }
 
+      logger.debug("Config changed for egress {} {} {} {}",
+        keyValue("id", newEgress.getUid()),
+        keyValue("consumerGroup", newEgress.getConsumerGroup()),
+        keyValue("destination", newEgress.getDestination()),
+        keyValue("contractGeneration", generation)
+      );
+
       futures.add(
         this.egressReconcilerListener.onUpdateEgress(newResource, newEgress)
           // If we fail to update the egress we can't put it in the cache.


Test patch:
diff --git a/control-plane/pkg/apis/config/features_test.go b/control-plane/pkg/apis/config/features_test.go
index 526f32bac..ab725d786 100644
--- a/control-plane/pkg/apis/config/features_test.go
+++ b/control-plane/pkg/apis/config/features_test.go
@@ -30,6 +30,7 @@ func TestFlags_IsDispatcherRateLimiterEnabled(t *testing.T) {
 	nc := DefaultFeaturesConfig()
 	require.False(t, nc.features.DispatcherRateLimiter == feature.Enabled)
 	require.False(t, nc.features.DispatcherOrderedExecutorMetrics == feature.Enabled)
+	require.False(t, nc.features.ControllerAutoscaler == feature.Enabled)
 }
 
 func TestFlags_IsEnabled_ContainingFlag(t *testing.T) {
@@ -38,10 +39,12 @@ func TestFlags_IsEnabled_ContainingFlag(t *testing.T) {
 		features: features{
 			DispatcherRateLimiter:            feature.Enabled,
 			DispatcherOrderedExecutorMetrics: feature.Enabled,
+			ControllerAutoscaler:             feature.Enabled,
 		},
 	})
 	require.True(t, nc.features.DispatcherRateLimiter == feature.Enabled)
 	require.True(t, nc.features.DispatcherOrderedExecutorMetrics == feature.Enabled)
+	require.True(t, nc.features.ControllerAutoscaler == feature.Enabled)
 }
 
 func TestGetFlags(t *testing.T) {
@@ -51,6 +54,7 @@ func TestGetFlags(t *testing.T) {
 
 	require.True(t, flags.IsDispatcherRateLimiterEnabled())
 	require.True(t, flags.IsDispatcherOrderedExecutorMetricsEnabled())
+	require.True(t, flags.IsControllerAutoscalerEnabled())
 }
 
 func TestStoreLoadWithConfigMap(t *testing.T) {
@@ -64,6 +68,7 @@ func TestStoreLoadWithConfigMap(t *testing.T) {
 
 	require.Equal(t, expected.IsDispatcherRateLimiterEnabled(), have.IsDispatcherRateLimiterEnabled())
 	require.Equal(t, expected.IsDispatcherOrderedExecutorMetricsEnabled(), have.IsDispatcherOrderedExecutorMetricsEnabled())
+	require.Equal(t, expected.IsControllerAutoscalerEnabled(), have.IsControllerAutoscalerEnabled())
 }
 
 func TestStoreLoadWithContext(t *testing.T) {
@@ -71,4 +76,5 @@ func TestStoreLoadWithContext(t *testing.T) {
 
 	require.False(t, have.IsDispatcherRateLimiterEnabled())
 	require.False(t, have.IsDispatcherOrderedExecutorMetricsEnabled())
+	require.False(t, have.IsControllerAutoscalerEnabled())
 }
diff --git a/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml b/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
index f2044d361..406ab46c0 100644
--- a/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
+++ b/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
@@ -7,3 +7,4 @@ data:
   _example: |
     dispatcher.rate-limiter: "enabled"
     dispatcher.ordered-executor-metrics: "enabled"
+    controller.autoscaler: "enabled"
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index d6accac56..b425884dd 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -50,6 +50,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 
+	configapis "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
@@ -1530,7 +1531,8 @@ func TestReconcileKind(t *testing.T) {
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
-			SystemNamespace: systemNamespace,
+			SystemNamespace:   systemNamespace,
+			KafkaFeatureFlags: configapis.DefaultFeaturesConfig(),
 		}
 
 		return consumergroup.NewReconciler(
@@ -1752,6 +1754,7 @@ func TestFinalizeKind(t *testing.T) {
 					T: t,
 				}, nil
 			},
+			KafkaFeatureFlags: configapis.DefaultFeaturesConfig(),
 		}
 
 		return consumergroup.NewReconciler(
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index aa2002d2e..786b7fcb3 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -29,6 +29,7 @@ import (
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/node/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"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
@@ -91,7 +92,11 @@ func TestNewController(t *testing.T) {
 	t.Setenv("POD_CAPACITY", PodCapacity)
 	t.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
 	t.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
-	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/test/config/100-config-kafka-features.yaml b/test/config/100-config-kafka-features.yaml
index 2ba1b59a7..a7f7e6c95 100644
--- a/test/config/100-config-kafka-features.yaml
+++ b/test/config/100-config-kafka-features.yaml
@@ -6,3 +6,4 @@ metadata:
 data:
   dispatcher.rate-limiter: "enabled"
   dispatcher.ordered-executor-metrics: "enabled"
+  controller.autoscaler: "enabled"