Repository: knative-sandbox/eventing-kafka-broker
Pull request #2922 (2023-01-26 06:30:28)

Going w/ Apache Kafka 3.3.2
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- after strimzi bump, going with Kafkas 3.3.2


<!--
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.
-->


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/2922#" title="Author self-approved">matzew</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)~~ [matzew]

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":[]} -->
matzew: /assign @dilipgb 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2922](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a6f8e30) 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.29%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #2922      +/-   ##
============================================
- Coverage     63.85%   58.56%   -5.29%     
============================================
  Files           155       78      -77     
  Lines         10931     8312    -2619     
  Branches        231        0     -231     
============================================
- Hits           6980     4868    -2112     
+ Misses         3449     3078     -371     
+ 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/2922?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/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `53.73% <0.00%> (ø)` | |
| [...ka/broker/dispatcher/impl/NoopResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9Ob29wUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | | |
| [...r/dispatcher/main/ConsumerVerticleFactoryImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlRmFjdG9yeUltcGwuamF2YQ==) | | |
| [.../kafka/broker/dispatcher/main/ProducerFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Qcm9kdWNlckZhY3RvcnkuamF2YQ==) | | |
| [...ng/kafka/broker/dispatcher/main/DispatcherEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9EaXNwYXRjaGVyRW52LmphdmE=) | | |
| [...ve/eventing/kafka/broker/dispatcher/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9NYWluLmphdmE=) | | |
| [...ker/receiver/impl/StrictRequestToRecordMapper.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1N0cmljdFJlcXVlc3RUb1JlY29yZE1hcHBlci5qYXZh) | | |
| [...dispatcher/impl/ResponseToHttpEndpointHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZVRvSHR0cEVuZHBvaW50SGFuZGxlci5qYXZh) | | |
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| ... and [68 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2922?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]: @matzew: 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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_main | a6f8e30eff86e346399953e1c28e036b24958916 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2922/upgrade-tests_eventing-kafka-broker_main/1618496582246731776) | true | `/test upgrade-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 -->

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/config/eventingkafkaconfig.go b/control-plane/pkg/config/eventingkafkaconfig.go
index c21c52b56..e2021ade6 100644
--- a/control-plane/pkg/config/eventingkafkaconfig.go
+++ b/control-plane/pkg/config/eventingkafkaconfig.go
@@ -19,8 +19,6 @@ package config
 import (
 	"github.com/Shopify/sarama"
 	"k8s.io/apimachinery/pkg/api/resource"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
 // EKKubernetesConfig and these EK sub-structs contain our custom configuration settings,
@@ -86,10 +84,9 @@ type EKSaramaConfig struct {
 
 // EventingKafkaConfig is the main struct that holds the Receiver, Dispatcher, and Kafka sub-items
 type EventingKafkaConfig struct {
-	Channel     EKChannelConfig           `json:"channel,omitempty"`
-	CloudEvents EKCloudEventConfig        `json:"cloudevents,omitempty"`
-	Kafka       EKKafkaConfig             `json:"kafka,omitempty"`
-	Sarama      EKSaramaConfig            `json:"sarama,omitempty"`
-	Source      EKSourceConfig            `json:"source,omitempty"`
-	Auth        *security.KafkaAuthConfig `json:"-"` // Not directly part of the configmap; loaded from the secret
+	Channel     EKChannelConfig    `json:"channel,omitempty"`
+	CloudEvents EKCloudEventConfig `json:"cloudevents,omitempty"`
+	Kafka       EKKafkaConfig      `json:"kafka,omitempty"`
+	Sarama      EKSaramaConfig     `json:"sarama,omitempty"`
+	Source      EKSourceConfig     `json:"source,omitempty"`
 }
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
-}
diff --git a/control-plane/pkg/security/config.go b/control-plane/pkg/security/config.go
index 9d44ac1c2..a35902e6a 100644
--- a/control-plane/pkg/security/config.go
+++ b/control-plane/pkg/security/config.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -143,27 +142,3 @@ func (a *AnnotationsSecretLocator) SecretNamespace() (string, bool) {
 	name, ok := a.SecretName()
 	return a.Namespace, len(a.Namespace) > 0 && ok && len(name) > 0
 }
-
-type KafkaAuthConfig struct {
-	TLS  *KafkaTlsConfig
-	SASL *KafkaSaslConfig
-}
-
-type KafkaTlsConfig struct {
-	Cacert   string
-	Usercert string
-	Userkey  string
-}
-
-type KafkaSaslConfig struct {
-	User     string
-	Password string
-	SaslType string
-}
-
-// HasSameSettings returns true if all of the SASL settings in the provided config are the same as in this struct
-func (c *KafkaSaslConfig) HasSameSettings(saramaConfig *sarama.Config) bool {
-	return saramaConfig.Net.SASL.User == c.User &&
-		saramaConfig.Net.SASL.Password == c.Password &&
-		string(saramaConfig.Net.SASL.Mechanism) == c.SaslType
-}
diff --git a/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go b/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go
index b149d9d83..5913628e4 100644
--- a/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go
+++ b/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go
@@ -19,7 +19,6 @@ package security
 import (
 	"strconv"
 
-	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -30,8 +29,7 @@ func ResolveAuthContextFromLegacySecret(s *corev1.Secret) (*NetSpecAuthContext,
 		return &NetSpecAuthContext{}, nil
 	}
 
-	legacyAuth := getAuthConfigFromSecret(s)
-	protocolStr, protocolContract := extractProtocol(legacyAuth)
+	protocolStr, protocolContract := getProtocolFromLegacyChannelSecret(s)
 
 	virtualSecret := s.DeepCopy()
 	virtualSecret.Data[ProtocolKey] = []byte(protocolStr)
@@ -104,83 +102,30 @@ func maybeAddKeyFieldRef(s *corev1.Secret, key string, field contract.SecretFiel
 	return false
 }
 
-func extractProtocol(auth *KafkaAuthConfig) (string, contract.Protocol) {
-	if hasTLSEnabled(auth) && hasSASLEnabled(auth) {
+func getProtocolFromLegacyChannelSecret(secret *corev1.Secret) (string, contract.Protocol) {
+	if hasTLSEnabled(secret) && hasSASLEnabled(secret) {
 		return ProtocolSASLSSL, contract.Protocol_SASL_SSL
 	}
-	if hasTLSEnabled(auth) {
+	if hasTLSEnabled(secret) {
 		return ProtocolSSL, contract.Protocol_SSL
 	}
-	if hasSASLEnabled(auth) {
+	if hasSASLEnabled(secret) {
 		return ProtocolSASLPlaintext, contract.Protocol_SASL_PLAINTEXT
 	}
 	return ProtocolPlaintext, contract.Protocol_PLAINTEXT
 }
 
-func hasTLSEnabled(auth *KafkaAuthConfig) bool {
-	return auth.TLS != nil && (auth.TLS.Userkey != "" || auth.TLS.Cacert != "" || auth.TLS.Usercert != "")
-}
+func hasTLSEnabled(secret *corev1.Secret) bool {
+	tlsCaCert := secret.Data[CaCertificateKey]
+	tlsEnabled, _ := strconv.ParseBool(string(secret.Data[SSLLegacyEnabled]))
 
-func hasSASLEnabled(auth *KafkaAuthConfig) bool {
-	return auth.SASL != nil && auth.SASL.User != "" || auth.SASL.Password != ""
+	return tlsEnabled || string(tlsCaCert) != ""
 }
 
-// getAuthConfigFromSecret Looks Up And Returns Kafka Auth Config And Brokers From Provided Secret
-func getAuthConfigFromSecret(secret *corev1.Secret) *KafkaAuthConfig {
-	if secret == nil || secret.Data == nil {
-		return nil
-	}
-
-	username := string(secret.Data[SaslUsernameKey])
-	saslType := string(secret.Data[SaslType])
-	var authConfig KafkaAuthConfig
-	// Backwards-compatibility - Support old consolidated secret fields if present
-	// (TLS data is now in the configmap, e.g. sarama.Config.Net.TLS.Config.RootPEMs)
-	_, hasTlsCaCert := secret.Data[CaCertificateKey]
-	_, hasTlsEnabled := secret.Data[SSLLegacyEnabled]
-	if hasTlsEnabled || hasTlsCaCert {
-		parseTls(secret, &authConfig)
-		username = string(secret.Data[SaslUserKey])
-		saslType = string(secret.Data[SaslTypeLegacy]) // old "saslType" is different than new "sasltype"
-	}
-
-	// If we don't convert the empty string to the "PLAIN" default, the client.HasSameSettings()
-	// function will assume that they should be treated as differences and needlessly reconfigure
-	if saslType == "" {
-		saslType = sarama.SASLTypePlaintext
-	}
-
-	authConfig.SASL = &KafkaSaslConfig{
-		User:     username,
-		Password: string(secret.Data[SaslPasswordKey]),
-		SaslType: saslType,
-	}
+func hasSASLEnabled(secret *corev1.Secret) bool {
+	_, hasSaslPassword := secret.Data[SaslPasswordKey]
+	_, hasSaslUsername := secret.Data[SaslUsernameKey] // legacy secret
+	_, hasSaslUser := secret.Data[SaslUserKey]
 
-	return &authConfig
-}
-
-// parseTls allows backward-compatibility with older consolidated channel secrets
-func parseTls(secret *corev1.Secret, kafkaAuthConfig *KafkaAuthConfig) {
-
-	// self-signed CERTs we need CA CERT, USER CERT and KEY
-	if string(secret.Data[CaCertificateKey]) != "" {
-		// We have a self-signed TLS cert
-		tls := &KafkaTlsConfig{
-			Cacert:   string(secret.Data[CaCertificateKey]),
-			Usercert: string(secret.Data[UserCertificate]),
-			Userkey:  string(secret.Data[UserKey]),
-		}
-		kafkaAuthConfig.TLS = tls
-	} else {
-		// Public CERTS from a proper CA do not need this,
-		// we can just say `tls.enabled: true`
-		tlsEnabled, err := strconv.ParseBool(string(secret.Data[SSLLegacyEnabled]))
-		if err != nil {
-			tlsEnabled = false
-		}
-		if tlsEnabled {
-			// Looks like TLS is desired/enabled:
-			kafkaAuthConfig.TLS = &KafkaTlsConfig{}
-		}
-	}
+	return hasSaslPassword && (hasSaslUsername || hasSaslUser)
 }


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/control-plane/pkg/security/secrets_provider_legacy_channel_secret_test.go b/control-plane/pkg/security/secrets_provider_legacy_channel_secret_test.go
index ebbd897fe..6ac8d0989 100644
--- a/control-plane/pkg/security/secrets_provider_legacy_channel_secret_test.go
+++ b/control-plane/pkg/security/secrets_provider_legacy_channel_secret_test.go
@@ -19,70 +19,108 @@ package security
 import (
 	"testing"
 
-	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/pkg/system"
 )
 
-func TestGetAuthConfigFromSecret(t *testing.T) {
+func Test_getProtocolFromLegacyChannelSecret(t *testing.T) {
 	// Define The TestCase Struct
 	type TestCase struct {
-		name      string
-		data      map[string][]byte
-		expectNil bool
-	}
-
-	// Asserts that, if there is a key in the data, that its mapped value is the same as the given value
-	assertKey := func(t *testing.T, data map[string][]byte, key string, value string) {
-		mapValue, ok := data[key]
-		if !ok {
-			return
-		}
-		assert.Equal(t, string(mapValue), value)
+		name                     string
+		data                     map[string][]byte
+		expectedProtocolStr      string
+		expectedProtocolContract contract.Protocol
 	}
 
 	// Create The TestCases
 	testCases := []TestCase{
 		{
-			name: "Valid secret",
+			name:                     "Plaintext",
+			data:                     map[string][]byte{},
+			expectedProtocolStr:      ProtocolPlaintext,
+			expectedProtocolContract: contract.Protocol_PLAINTEXT,
+		},
+		{
+			name: "SSL/TLS via CA cert",
+			data: map[string][]byte{
+				CaCertificateKey: []byte("test-cacert"),
+			},
+			expectedProtocolStr:      ProtocolSSL,
+			expectedProtocolContract: contract.Protocol_SSL,
+		},
+		{
+			name: "SSL/TLS via tls.enabled",
+			data: map[string][]byte{
+				SSLLegacyEnabled: []byte("true"),
+			},
+			expectedProtocolStr:      ProtocolSSL,
+			expectedProtocolContract: contract.Protocol_SSL,
+		},
+		{
+			name: "SASL",
+			data: map[string][]byte{
+				SaslUsernameKey: []byte("test-sasluser"),
+				SaslPasswordKey: []byte("test-password"),
+			},
+			expectedProtocolStr:      ProtocolSASLPlaintext,
+			expectedProtocolContract: contract.Protocol_SASL_PLAINTEXT,
+		},
+		{
+			name: "SASL via legacy channel secret",
 			data: map[string][]byte{
-				SaslUsernameKey: []byte("OldAuthUsername"),
-				SaslPasswordKey: []byte("OldAuthPassword"),
-				SaslType:        []byte("OldAuthSaslType"),
+				SaslUserKey:     []byte("test-sasluser"),
+				SaslPasswordKey: []byte("test-password"),
 			},
+			expectedProtocolStr:      ProtocolSASLPlaintext,
+			expectedProtocolContract: contract.Protocol_SASL_PLAINTEXT,
 		},
+
 		{
-			name: "Valid secret, backwards-compatibility, TLS and SASL",
+			name: "SASL and TLS/SSL via ca cert",
 			data: map[string][]byte{
 				CaCertificateKey: []byte("test-cacert"),
-				UserCertificate:  []byte("test-usercert"),
-				UserKey:          []byte("test-userkey"),
+
+				SaslUsernameKey: []byte("test-sasluser"),
+				SaslPasswordKey: []byte("test-password"),
+			},
+			expectedProtocolStr:      ProtocolSASLSSL,
+			expectedProtocolContract: contract.Protocol_SASL_SSL,
+		},
+		{
+			name: "SASL and TLS/SSL via tls.enabled",
+			data: map[string][]byte{
 				SSLLegacyEnabled: []byte("true"),
-				SaslUserKey:      []byte("test-sasluser"),
-				SaslPasswordKey:  []byte("test-password"),
+
+				SaslUsernameKey: []byte("test-sasluser"),
+				SaslPasswordKey: []byte("test-password"),
 			},
+			expectedProtocolStr:      ProtocolSASLSSL,
+			expectedProtocolContract: contract.Protocol_SASL_SSL,
 		},
 		{
-			name: "Valid secret, backwards-compatibility, TLS/SASL without CA cert",
+			name: "SASL and TLS/SSL",
 			data: map[string][]byte{
+				CaCertificateKey: []byte("test-cacert"),
+				UserCertificate:  []byte("test-usercert"),
+				UserKey:          []byte("test-userkey"),
 				SSLLegacyEnabled: []byte("true"),
 				SaslUserKey:      []byte("test-sasluser"),
 				SaslPasswordKey:  []byte("test-password"),
 			},
+			expectedProtocolStr:      ProtocolSASLSSL,
+			expectedProtocolContract: contract.Protocol_SASL_SSL,
 		},
 		{
-			name: "Valid secret, backwards-compatibility, SASL, Invalid TLS",
+			name: "SASL and TLS/SSL (with invalid TLS)",
 			data: map[string][]byte{
 				SSLLegacyEnabled: []byte("invalid-bool"),
 				SaslUserKey:      []byte("test-sasluser"),
 				SaslPasswordKey:  []byte("test-password"),
 			},
-		},
-		{
-			name:      "Valid secret, backwards-compatibility, nil data",
-			data:      nil,
-			expectNil: true,
+			expectedProtocolStr:      ProtocolSASLPlaintext,
+			expectedProtocolContract: contract.Protocol_SASL_PLAINTEXT,
 		},
 	}
 
@@ -100,21 +138,13 @@ func TestGetAuthConfigFromSecret(t *testing.T) {
 				},
 				Data: testCase.data,
 			}
-			kafkaAuth := getAuthConfigFromSecret(secret)
-			if testCase.expectNil {
-				assert.Nil(t, kafkaAuth)
-			} else {
-				assert.NotNil(t, kafkaAuth)
-				assert.NotNil(t, kafkaAuth.SASL)
-				assertKey(t, testCase.data, SaslUsernameKey, kafkaAuth.SASL.User)
-				assertKey(t, testCase.data, SaslType, kafkaAuth.SASL.SaslType)
-				assertKey(t, testCase.data, SaslUserKey, kafkaAuth.SASL.User)
-				assertKey(t, testCase.data, SaslPasswordKey, kafkaAuth.SASL.Password)
-				if kafkaAuth.TLS != nil {
-					assertKey(t, testCase.data, CaCertificateKey, kafkaAuth.TLS.Cacert)
-					assertKey(t, testCase.data, UserCertificate, kafkaAuth.TLS.Usercert)
-					assertKey(t, testCase.data, UserKey, kafkaAuth.TLS.Userkey)
-				}
+
+			gotProtocolStr, gotRrotocolContract := getProtocolFromLegacyChannelSecret(secret)
+			if gotProtocolStr != testCase.expectedProtocolStr {
+				t.Errorf("getProtocolFromLegacyChannelSecret() gotProtocolStr = %v, want %v", gotProtocolStr, testCase.expectedProtocolStr)
+			}
+			if gotRrotocolContract != testCase.expectedProtocolContract {
+				t.Errorf("getProtocolFromLegacyChannelSecret() gotRrotocolContract = %v, want %v", gotRrotocolContract, testCase.expectedProtocolContract)
 			}
 		})
 	}
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/kafka/kafka-ephemeral.yaml b/test/kafka/kafka-ephemeral.yaml
index 83613d195..d304aa921 100644
--- a/test/kafka/kafka-ephemeral.yaml
+++ b/test/kafka/kafka-ephemeral.yaml
@@ -18,7 +18,7 @@ metadata:
   name: my-cluster
 spec:
   kafka:
-    version: 3.3.1
+    version: 3.3.2
     replicas: 3
     listeners:
       # PLAINTEXT
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 }}
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index f8e2d417a..ee402f407 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,8 +16,8 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -50,8 +50,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -67,8 +67,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -84,8 +84,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -101,8 +101,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -118,8 +118,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -151,8 +151,8 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -160,8 +160,8 @@ kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -193,8 +193,8 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -202,8 +202,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -220,8 +220,8 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -237,8 +237,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -254,8 +254,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -287,8 +287,8 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 data:
   channel-template-spec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -354,8 +354,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -389,10 +389,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -436,10 +436,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -480,7 +480,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -529,8 +529,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -591,10 +591,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -645,10 +645,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -722,8 +722,8 @@ metadata:
   name: config-sugar
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "62dfac6f"
@@ -780,10 +780,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -838,10 +838,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -851,9 +851,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20230116-694f82436"
+        eventing.knative.dev/release: "v20230125-466d12318"
         app.kubernetes.io/component: eventing-controller
-        app.kubernetes.io/version: "20230116-694f82436"
+        app.kubernetes.io/version: "20230125-466d12318"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -871,7 +871,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:c2b6976cb2bf239f165deeca6d864392f6f8ab0f7c69607d8cc2d66ebe2daf2c
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:96a09bed3f78f5ab019e6d3b1991055e8297eb4b41d303a2e4044b3f453b4ac8
           resources:
             requests:
               cpu: 100m
@@ -889,7 +889,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:8b0c38aa3c090525d342d50afaf47ab96bf64651a40d4479977a2a50d63ba990
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:25609f8694b7fabf54272ed442c1724480ede94f82cd4c920e147f7bf2f1cb63
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -958,9 +958,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
@@ -973,9 +973,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20230116-694f82436"
+        eventing.knative.dev/release: "v20230125-466d12318"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20230116-694f82436"
+        app.kubernetes.io/version: "20230125-466d12318"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -989,7 +989,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:29e0cf00f11d090497df3688c3cfac34e2883ff7d0acc906ee948c154ffad1bc
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:47143c2f29db5096a88a3a359fbfaa17b1398e9ff53bea7f0fde0c0358ad2837
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -1062,9 +1062,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1088,9 +1088,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1119,9 +1119,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1132,9 +1132,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20230116-694f82436"
+        eventing.knative.dev/release: "v20230125-466d12318"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20230116-694f82436"
+        app.kubernetes.io/version: "20230125-466d12318"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1154,7 +1154,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:b48d261b9de0d82f9854549caf86cb250313cb79f4633c2ff2a48b5026f7150a
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:466105a36d0fb070dd4fb51f6bf42fef108a1d00a1d319cce858d5c5e8cdfed7
           resources:
             requests:
               # taken from serving.
@@ -1225,10 +1225,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1260,11 +1260,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1476,10 +1476,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1643,11 +1643,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1918,11 +1918,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -2072,9 +2072,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2201,10 +2201,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2456,11 +2456,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2647,10 +2647,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2967,12 +2967,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3159,9 +3159,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3371,9 +3371,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3557,8 +3557,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3571,9 +3571,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3591,9 +3591,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3614,9 +3614,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3641,9 +3641,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3662,9 +3662,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3700,8 +3700,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3727,8 +3727,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3745,8 +3745,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3779,8 +3779,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3793,9 +3793,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3833,9 +3833,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3847,9 +3847,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3861,9 +3861,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3875,9 +3875,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3889,9 +3889,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3904,8 +3904,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3918,8 +3918,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3946,8 +3946,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4082,8 +4082,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4150,8 +4150,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4164,9 +4164,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4212,8 +4212,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4226,9 +4226,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4264,8 +4264,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4351,8 +4351,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4462,8 +4462,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4499,8 +4499,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4537,8 +4537,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4571,8 +4571,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4606,8 +4606,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4631,8 +4631,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
-    app.kubernetes.io/version: "20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 5c9f7e686..a19e42298 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,11 +17,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -233,10 +233,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -400,11 +400,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -675,11 +675,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -829,9 +829,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -958,10 +958,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1213,11 +1213,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1404,10 +1404,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1724,12 +1724,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1916,9 +1916,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2128,9 +2128,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20230116-694f82436"
+    eventing.knative.dev/release: "v20230125-466d12318"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20230116-694f82436"
+    app.kubernetes.io/version: "20230125-466d12318"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev