Repository: knative-sandbox/eventing-kafka-broker
Pull request #2801 (2022-11-22 15:59:09)

Add maxDuplicatesPercentage for Broker sacura test
This adds a limit to duplicates for the sacura test of the Broker

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2801](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0b46753) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/770d515ce488fa6ac2761460240394715e34a0a1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (770d515) will **decrease** coverage by `6.24%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #2801      +/-   ##
============================================
- Coverage     61.69%   55.45%   -6.25%     
============================================
  Files           152       75      -77     
  Lines         10712     8093    -2619     
  Branches        231        0     -231     
============================================
- Hits           6609     4488    -2121     
+ Misses         3638     3281     -357     
+ Partials        465      324     -141     
```

| 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/2801?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ker/dispatcher/impl/consumer/ConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Db25zdW1lclZlcnRpY2xlLmphdmE=) | | |
| [...ve/eventing/kafka/broker/dispatcher/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9NYWluLmphdmE=) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...ative/eventing/kafka/broker/dispatcher/Filter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvRmlsdGVyLmphdmE=) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...tive/eventing/kafka/broker/receiver/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL01haW4uamF2YQ==) | | |
| [...eceiver/impl/IngressProducerReconcilableStore.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL0luZ3Jlc3NQcm9kdWNlclJlY29uY2lsYWJsZVN0b3JlLmphdmE=) | | |
| [...ceiver/impl/handler/IngressRequestHandlerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvSW5ncmVzc1JlcXVlc3RIYW5kbGVySW1wbC5qYXZh) | | |
| [...ispatcher/main/ConsumerVerticleLoggingContext.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlTG9nZ2luZ0NvbnRleHQuamF2YQ==) | | |
| ... and [67 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2801/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

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


Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml
new file mode 100644
index 000000000..d5aebc4a6
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml
@@ -0,0 +1,29 @@
+# Copyright 2022 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-autoscaler
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+    class: "keda.autoscaling.knative.dev"
+    min-scale: "0"
+    max-scale: "50"
+    polling-interval: "10"
+    cooldown-period: "300"
+    lag-threshold: "100"
+    activation-lag-threshold: "1"
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
index 748e38fcf..368face6a 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
@@ -147,6 +147,9 @@ spec:
             - name: DESCHEDULER_CONFIG
               value: 'config-kafka-descheduler'
 
+            - name: AUTOSCALER_CONFIG
+              value: 'config-kafka-autoscaler'
+
             - name: CONFIG_LEADERELECTION_NAME
               value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
diff --git a/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-autoscaler.yaml b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-autoscaler.yaml
new file mode 120000
index 000000000..b23ec2379
--- /dev/null
+++ b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-autoscaler.yaml
@@ -0,0 +1,1 @@
+../../eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml
\ No newline at end of file
diff --git a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
index 31c78f744..b343fbd65 100644
--- a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
@@ -82,6 +82,9 @@ spec:
             - name: DESCHEDULER_CONFIG
               value: 'config-kafka-descheduler'
 
+            - name: AUTOSCALER_CONFIG
+              value: 'config-kafka-autoscaler'
+
             - name: CONFIG_LEADERELECTION_NAME
               value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
diff --git a/control-plane/pkg/apis/core/pod_defaulting.go b/control-plane/pkg/apis/core/pod_defaulting.go
index 2ffaa586b..399d36359 100644
--- a/control-plane/pkg/apis/core/pod_defaulting.go
+++ b/control-plane/pkg/apis/core/pod_defaulting.go
@@ -28,7 +28,7 @@ import (
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
 
-	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
 // DispatcherPodsDefaulting returns the defaulting Callback for dispatcher pods
@@ -57,7 +57,7 @@ func podDefaultingCallback(generator generatorFunc) defaulting.CallbackFunc {
 		}
 
 		volume := &corev1.Volume{
-			Name: kafkainternals.DispatcherVolumeName,
+			Name: eventing.DispatcherVolumeName,
 			VolumeSource: corev1.VolumeSource{
 				ConfigMap: &corev1.ConfigMapVolumeSource{
 					LocalObjectReference: corev1.LocalObjectReference{Name: cmName},
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
index bbc185a6e..a2e822ceb 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
@@ -16,7 +16,29 @@
 
 package eventing
 
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
 const (
 	// ConfigMapVolumeName is the volume name of the data plane ConfigMap
 	ConfigMapVolumeName = "kafka-resources"
+
+	DispatcherVolumeName = "contract-resources"
 )
+
+func ConfigMapNameFromPod(p *corev1.Pod) (string, error) {
+	var vDp *corev1.Volume
+	for i, v := range p.Spec.Volumes {
+		if v.Name == DispatcherVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
+			vDp = &p.Spec.Volumes[i]
+			break
+		}
+	}
+	if vDp == nil {
+		return "", fmt.Errorf("failed to get data plane volume %s in pod %s/%s", ConfigMapVolumeName, p.GetNamespace(), p.GetName())
+	}
+	return vDp.ConfigMap.Name, nil
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index bc303253d..bab2fa5da 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -27,6 +27,7 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
@@ -124,7 +125,9 @@ type DestinationReply struct {
 }
 
 type Auth struct {
-	NetSpec    *bindings.KafkaNetSpec
+	NetSpec *bindings.KafkaNetSpec
+	// Deprecated, use secret spec
+	AuthSpec   *eventingv1alpha1.Auth `json:"AuthSpec,omitempty"`
 	SecretSpec *SecretSpec
 }
 
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index 01104965c..ffaa2a9c9 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -23,6 +23,7 @@ package v1alpha1
 
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
+	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -38,6 +39,11 @@ func (in *Auth) DeepCopyInto(out *Auth) {
 		*out = new(v1beta1.KafkaNetSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.AuthSpec != nil {
+		in, out := &in.AuthSpec, &out.AuthSpec
+		*out = new(eventingv1alpha1.Auth)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.SecretSpec != nil {
 		in, out := &in.SecretSpec, &out.SecretSpec
 		*out = new(SecretSpec)
diff --git a/control-plane/pkg/autoscaler/config.go b/control-plane/pkg/autoscaler/config.go
new file mode 100644
index 000000000..9b84e0878
--- /dev/null
+++ b/control-plane/pkg/autoscaler/config.go
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package autoscaler
+
+import (
+	"fmt"
+	"strconv"
+)
+
+// AutoscalerConfig contains the defaults defined in the autoscaler ConfigMap.
+type AutoscalerConfig struct {
+	AutoscalerClass    map[string]string
+	AutoscalerDefaults map[string]int32
+}
+
+func defaultConfig() *AutoscalerConfig {
+	return &AutoscalerConfig{
+		AutoscalerClass: map[string]string{
+			AutoscalingClassAnnotation: AutoscalingClassDisabledAnnotationValue,
+		},
+		AutoscalerDefaults: map[string]int32{
+			AutoscalingMinScaleAnnotation:        DefaultMinReplicaCount,
+			AutoscalingMaxScaleAnnotation:        DefaultMaxReplicaCount,
+			AutoscalingPollingIntervalAnnotation: DefaultPollingInterval,
+			AutoscalingCooldownPeriodAnnotation:  DefaultCooldownPeriod,
+			AutoscalingLagThreshold:              DefaultLagThreshold,
+			AutoscalingActivationLagThreshold:    DefaultActivationLagThreshold,
+		},
+	}
+}
+
+// NewConfigFromMap creates a AutoscalerConfig from the supplied map
+func NewConfigFromMap(data map[string]string) (*AutoscalerConfig, error) {
+	ac := defaultConfig()
+
+	if class, ok := data["class"]; ok && class == AutoscalingClassDisabledAnnotationValue { //autoscaler disabled - ignore new defaults
+		return ac, nil
+	}
+
+	for key, val := range data {
+		key = convertConfigKeyToAutoscalingAnnotation(key)
+
+		if key == AutoscalingClassAnnotation {
+			ac.AutoscalerClass[AutoscalingClassAnnotation] = val
+		} else {
+			i, err := strconv.ParseInt(val, 10, 32)
+			if err != nil {
+				return nil, fmt.Errorf("Expected value for autoscaling annotation: "+key+" should be integer but got "+val, err)
+			}
+			ac.AutoscalerDefaults[key] = int32(i)
+		}
+	}
+
+	return ac, nil
+}
+
+func convertConfigKeyToAutoscalingAnnotation(configkey string) (autoscalingannotation string) {
+	return Autoscaling + "/" + configkey
+}
diff --git a/control-plane/pkg/autoscaler/keda/keda.go b/control-plane/pkg/autoscaler/keda/keda.go
index 6760fde59..73813a7e1 100644
--- a/control-plane/pkg/autoscaler/keda/keda.go
+++ b/control-plane/pkg/autoscaler/keda/keda.go
@@ -22,6 +22,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	"knative.dev/pkg/kmeta"
 
@@ -46,17 +47,17 @@ func GenerateScaleTarget(cg *kafkainternals.ConsumerGroup) *kedav1alpha1.ScaleTa
 	}
 }
 
-func GenerateScaleTriggers(cg *kafkainternals.ConsumerGroup, triggerAuthentication *kedav1alpha1.TriggerAuthentication) ([]kedav1alpha1.ScaleTriggers, error) {
+func GenerateScaleTriggers(cg *kafkainternals.ConsumerGroup, triggerAuthentication *kedav1alpha1.TriggerAuthentication, aconfig autoscaler.AutoscalerConfig) ([]kedav1alpha1.ScaleTriggers, error) {
 	triggers := make([]kedav1alpha1.ScaleTriggers, 0, len(cg.Spec.Template.Spec.Topics))
 	bootstrapServers := cg.Spec.Template.Spec.Configs.Configs[kafka.BootstrapServersConfigMapKey]
 	consumerGroup := cg.Spec.Template.Spec.Configs.Configs[kafka.GroupIDConfigMapKey]
 
-	lagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingLagThreshold, autoscaler.DefaultLagThreshold)
+	lagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingLagThreshold, aconfig.AutoscalerDefaults[autoscaler.AutoscalingLagThreshold])
 	if err != nil {
 		return nil, err
 	}
 
-	activationLagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingActivationLagThreshold, autoscaler.DefaultActivationLagThreshold)
+	activationLagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingActivationLagThreshold, aconfig.AutoscalerDefaults[autoscaler.AutoscalingActivationLagThreshold])
 	if err != nil {
 		return nil, err
 	}
@@ -206,21 +207,19 @@ func GenerateTriggerAuthentication(cg *kafkainternals.ConsumerGroup, secretData
 }
 
 func retrieveSaslTypeIfPresent(cg *kafkainternals.ConsumerGroup, secret corev1.Secret) *string {
-	var saslType string
-
 	if cg.Spec.Template.Spec.Auth.NetSpec != nil && cg.Spec.Template.Spec.Auth.NetSpec.SASL.Enable && cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef != nil {
 		secretKeyRefKey := cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef.Key
 		if saslTypeValue, ok := secret.Data[secretKeyRefKey]; ok {
-			saslType = string(saslTypeValue)
+			return pointer.String(string(saslTypeValue))
 		}
 	}
 
 	if cg.Spec.Template.Spec.Auth.SecretSpec != nil && cg.Spec.Template.Spec.Auth.SecretSpec.Ref != nil {
 		if saslTypeValue, ok := secret.Data[security.SaslType]; ok {
-			saslType = string(saslTypeValue)
+			return pointer.String(string(saslTypeValue))
 		}
 	}
-	return &saslType
+	return nil
 }
 
 func addAuthSecretTargetRef(parameter string, secretKeyRef v1beta1.SecretValueFromSource, secretTargetRefs []kedav1alpha1.AuthSecretTargetRef) []kedav1alpha1.AuthSecretTargetRef {
diff --git a/control-plane/pkg/autoscaler/keda/resources.go b/control-plane/pkg/autoscaler/keda/resources.go
index de2c24433..b7e9730d4 100644
--- a/control-plane/pkg/autoscaler/keda/resources.go
+++ b/control-plane/pkg/autoscaler/keda/resources.go
@@ -31,21 +31,21 @@ var (
 	KedaSchemeGroupVersion = schema.GroupVersion{Group: "keda.sh", Version: "v1alpha1"}
 )
 
-func GenerateScaledObject(obj metav1.Object, gvk schema.GroupVersionKind, scaleTarget *kedav1alpha1.ScaleTarget, triggers []kedav1alpha1.ScaleTriggers) (*kedav1alpha1.ScaledObject, error) {
+func GenerateScaledObject(obj metav1.Object, gvk schema.GroupVersionKind, scaleTarget *kedav1alpha1.ScaleTarget, triggers []kedav1alpha1.ScaleTriggers, aconfig autoscaler.AutoscalerConfig) (*kedav1alpha1.ScaledObject, error) {
 
-	cooldownPeriod, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingCooldownPeriodAnnotation, autoscaler.DefaultCooldownPeriod)
+	cooldownPeriod, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingCooldownPeriodAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingCooldownPeriodAnnotation])
 	if err != nil {
 		return nil, err
 	}
-	pollingInterval, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingPollingIntervalAnnotation, autoscaler.DefaultPollingInterval)
+	pollingInterval, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingPollingIntervalAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingPollingIntervalAnnotation])
 	if err != nil {
 		return nil, err
 	}
-	minReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMinScaleAnnotation, autoscaler.DefaultMinReplicaCount)
+	minReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMinScaleAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingMinScaleAnnotation])
 	if err != nil {
 		return nil, err
 	}
-	maxReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMaxScaleAnnotation, autoscaler.DefaultMaxReplicaCount)
+	maxReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMaxScaleAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingMaxScaleAnnotation])
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index e447f8c55..68e3386a7 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -25,12 +25,14 @@ import (
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/utils/pointer"
 	"knative.dev/pkg/tracker"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -155,6 +157,25 @@ func (r *Reconciler) createDataPlaneConfigMap(ctx context.Context) (*corev1.Conf
 	return r.KubeClient.CoreV1().ConfigMaps(r.DataPlaneConfigMapNamespace).Create(ctx, cm, metav1.CreateOptions{})
 }
 
+func PodOwnerReference(p *corev1.Pod) ConfigMapOption {
+	expected := metav1.OwnerReference{
+		APIVersion:         corev1.SchemeGroupVersion.String(),
+		Kind:               "Pod",
+		Name:               p.Name,
+		UID:                p.UID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(true),
+	}
+	return func(cm *corev1.ConfigMap) {
+		for _, or := range cm.OwnerReferences {
+			if equality.Semantic.DeepDerivative(expected, or) {
+				return
+			}
+		}
+		cm.OwnerReferences = append(cm.OwnerReferences, expected)
+	}
+}
+
 // GetDataPlaneConfigMapData extracts contract from the given config map.
 func (r *Reconciler) GetDataPlaneConfigMapData(logger *zap.Logger, dataPlaneConfigMap *corev1.ConfigMap) (*contract.Contract, error) {
 	return GetDataPlaneConfigMapData(logger, dataPlaneConfigMap, r.ContractConfigMapFormat)
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 5c88214b8..596862cf7 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -22,12 +22,9 @@ import (
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/utils/pointer"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
@@ -332,7 +329,7 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 	}
 
 	// Get contract associated with the pod.
-	cmName, err := cmNameFromPod(p, c)
+	cmName, err := eventing.ConfigMapNameFromPod(p)
 	if err != nil {
 		return false, err
 	}
@@ -383,41 +380,7 @@ func (r Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconcil
 		DataPlaneNamespace:            p.GetNamespace(),
 		DispatcherLabel:               "",
 		ReceiverLabel:                 "",
-		DataPlaneConfigMapTransformer: podOwnerReference(p),
-	}
-}
-
-func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
-	var vDp *corev1.Volume
-	for i, v := range p.Spec.Volumes {
-		if v.Name == kafkainternals.DispatcherVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
-			vDp = &p.Spec.Volumes[i]
-			break
-		}
-	}
-	if vDp == nil {
-		return "", fmt.Errorf("failed to get data plane volume %s in pod %s/%s", eventing.ConfigMapVolumeName, c.Spec.PodBind.PodNamespace, c.Spec.PodBind.PodName)
-	}
-	return vDp.ConfigMap.Name, nil
-}
-
-func podOwnerReference(p *corev1.Pod) base.ConfigMapOption {
-
-	expected := metav1.OwnerReference{
-		APIVersion:         corev1.SchemeGroupVersion.String(),
-		Kind:               "Pod",
-		Name:               p.Name,
-		UID:                p.UID,
-		Controller:         pointer.Bool(true),
-		BlockOwnerDeletion: pointer.Bool(true),
-	}
-	return func(cm *corev1.ConfigMap) {
-		for _, or := range cm.OwnerReferences {
-			if equality.Semantic.DeepDerivative(expected, or) {
-				return
-			}
-		}
-		cm.OwnerReferences = append(cm.OwnerReferences, expected)
+		DataPlaneConfigMapTransformer: base.PodOwnerReference(p),
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/consumergroup/autoscaler_config.go b/control-plane/pkg/reconciler/consumergroup/autoscaler_config.go
new file mode 100644
index 000000000..1338ef4ab
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/autoscaler_config.go
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package consumergroup
+
+import (
+	"context"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler"
+)
+
+func (r Reconciler) autoscalerDefaultsFromConfigMap(ctx context.Context, configMapName string) (*autoscaler.AutoscalerConfig, error) {
+	cm, err := r.KubeClient.CoreV1().ConfigMaps(r.SystemNamespace).Get(ctx, configMapName, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("error while retrieving the %s config map in namespace %s: %+v", configMapName, r.SystemNamespace, err)
+	}
+
+	config, err := autoscaler.NewConfigFromMap(cm.Data)
+	if err != nil {
+		return nil, fmt.Errorf("error while parsing the %s config map in namespace %s: %+v", configMapName, r.SystemNamespace, err)
+	}
+
+	return config, nil
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 82a43af0d..7f90bce64 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -43,13 +43,14 @@ import (
 	"knative.dev/eventing/pkg/scheduler"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
 	kedaclientset "knative.dev/eventing-kafka-broker/third_party/pkg/client/clientset/versioned"
 )
@@ -59,13 +60,20 @@ var (
 	ErrNoDeadLetterSinkURI = errors.New("no dead letter sink URI resolved")
 )
 
-type schedulerFunc func(s string) scheduler.Scheduler
+type Scheduler struct {
+	scheduler.Scheduler
+	SchedulerConfig
+}
+
+type schedulerFunc func(s string) Scheduler
 
 type Reconciler struct {
 	SchedulerFunc   schedulerFunc
 	ConsumerLister  kafkainternalslisters.ConsumerLister
 	InternalsClient internalv1alpha1.InternalV1alpha1Interface
 	SecretLister    corelisters.SecretLister
+	ConfigMapLister corelisters.ConfigMapLister
+	PodLister       corelisters.PodLister
 	KubeClient      kubernetes.Interface
 
 	NameGenerator names.NameGenerator
@@ -86,6 +94,7 @@ type Reconciler struct {
 
 	KafkaFeatureFlags *config.KafkaFeatureFlags
 	KedaClient        kedaclientset.Interface
+	AutoscalerConfig  string
 }
 
 func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
@@ -93,7 +102,7 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 		return cg.MarkInitializeOffsetFailed("InitializeOffset", err)
 	}
 
-	if err := r.schedule(cg); err != nil {
+	if err := r.schedule(ctx, cg); err != nil {
 		return err
 	}
 	cg.MarkScheduleSucceeded()
@@ -138,7 +147,7 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 func (r Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
 
 	cg.Spec.Replicas = pointer.Int32(0)
-	err := r.schedule(cg) //de-schedule placements
+	err := r.schedule(ctx, cg) //de-schedule placements
 
 	if err != nil {
 		cg.Status.Placements = nil
@@ -295,8 +304,17 @@ func (r Reconciler) finalizeConsumer(ctx context.Context, consumer *kafkainterna
 	return nil
 }
 
-func (r Reconciler) schedule(cg *kafkainternals.ConsumerGroup) error {
-	placements, err := r.SchedulerFunc(cg.GetUserFacingResourceRef().Kind).Schedule(cg)
+func (r Reconciler) schedule(ctx context.Context, cg *kafkainternals.ConsumerGroup) error {
+	statefulSetScheduler := r.SchedulerFunc(cg.GetUserFacingResourceRef().Kind)
+
+	// Ensure Contract configmaps are created before scheduling to avoid having pending pods due to missing
+	// volumes.
+	// See https://github.com/knative-sandbox/eventing-kafka-broker/issues/2750#issuecomment-1304244017
+	if err := r.ensureContractConfigmapsExist(ctx, statefulSetScheduler); err != nil {
+		return cg.MarkScheduleConsumerFailed("Schedule", err)
+	}
+
+	placements, err := statefulSetScheduler.Schedule(cg)
 	if err != nil {
 		return cg.MarkScheduleConsumerFailed("Schedule", err)
 	}
@@ -448,6 +466,11 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 	var triggerAuthentication *kedav1alpha1.TriggerAuthentication
 	var secret *corev1.Secret
 
+	autoscalerDefaults, err := r.autoscalerDefaultsFromConfigMap(ctx, r.AutoscalerConfig)
+	if err != nil {
+		return err
+	}
+
 	if hasSecretSpecConfig(cg.Spec.Template.Spec.Auth) || hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
 		secretData, err := r.retrieveSecretData(ctx, cg)
 		if err != nil {
@@ -460,12 +483,12 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 		}
 	}
 
-	triggers, err := keda.GenerateScaleTriggers(cg, triggerAuthentication)
+	triggers, err := keda.GenerateScaleTriggers(cg, triggerAuthentication, *autoscalerDefaults)
 	if err != nil {
 		return err
 	}
 
-	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), keda.GenerateScaleTarget(cg), triggers)
+	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), keda.GenerateScaleTarget(cg), triggers, *autoscalerDefaults)
 	if err != nil {
 		return err
 	}
@@ -524,7 +547,7 @@ func (r *Reconciler) reconcileScaledObject(ctx context.Context, expectedScaledOb
 	if !metav1.IsControlledBy(scaledObject, obj) {
 		return fmt.Errorf("scaledobject %s/%s is not owned by %s/%s", expectedScaledObject.Namespace, expectedScaledObject.Name, obj.GetNamespace(), obj.GetName())
 	}
-	if !equality.Semantic.DeepDerivative(expectedScaledObject.Spec, scaledObject.Spec) {
+	if !equality.Semantic.DeepDerivative(scaledObject.Spec, expectedScaledObject.Spec) {
 		scaledObject.Spec = expectedScaledObject.Spec
 		if _, err = r.KedaClient.KedaV1alpha1().ScaledObjects(expectedScaledObject.Namespace).Update(ctx, scaledObject, metav1.UpdateOptions{}); err != nil {
 			return fmt.Errorf("failed to update scaled object %s/%s: %w", expectedScaledObject.Namespace, expectedScaledObject.Name, err)
@@ -548,7 +571,7 @@ func (r *Reconciler) reconcileTriggerAuthentication(ctx context.Context, expecte
 	if !metav1.IsControlledBy(triggerAuth, obj) {
 		return fmt.Errorf("triggerauthentication object %s/%s is not owned by %s/%s", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, obj.GetNamespace(), obj.GetName())
 	}
-	if !equality.Semantic.DeepDerivative(expectedTriggerAuth.Spec, triggerAuth.Spec) {
+	if !equality.Semantic.DeepDerivative(triggerAuth.Spec, expectedTriggerAuth.Spec) {
 		triggerAuth.Spec = expectedTriggerAuth.Spec
 		if _, err = r.KedaClient.KedaV1alpha1().TriggerAuthentications(expectedTriggerAuth.Namespace).Update(ctx, triggerAuth, metav1.UpdateOptions{}); err != nil {
 			return fmt.Errorf("failed to update triggerauthentication object %s/%s: %w", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, err)
@@ -596,6 +619,54 @@ func (r *Reconciler) isKEDAEnabled(ctx context.Context, cg *kafkainternals.Consu
 	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.
+		Pods(r.SystemNamespace).
+		List(selector)
+	if err != nil {
+		return fmt.Errorf("failed to list statefulset pods with selector %v: %w", selector.String(), err)
+	}
+
+	for _, p := range pods {
+		cmName, err := eventing.ConfigMapNameFromPod(p)
+		if err != nil {
+			return err
+		}
+		if err := r.ensureContractConfigMapExists(ctx, p, cmName); err != nil {
+			return fmt.Errorf("failed to get ConfigMap %s/%s: %w", r.SystemNamespace, cmName, err)
+		}
+	}
+
+	return nil
+}
+
+func (r Reconciler) ensureContractConfigMapExists(ctx context.Context, p *corev1.Pod, name string) error {
+	// Check if ConfigMap exists in lister cache
+	_, err := r.ConfigMapLister.ConfigMaps(r.SystemNamespace).Get(name)
+	// ConfigMap already exists, return
+	if err == nil {
+		return nil
+	}
+
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get ConfigMap %s/%s: %w", r.SystemNamespace, name, err)
+	}
+
+	b := base.Reconciler{
+		KubeClient:                    r.KubeClient,
+		DataPlaneConfigMapNamespace:   r.SystemNamespace,
+		ContractConfigMapName:         name,
+		DataPlaneNamespace:            r.SystemNamespace,
+		DataPlaneConfigMapTransformer: base.PodOwnerReference(p),
+	}
+
+	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
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 	_ consumergroup.Finalizer = &Reconciler{}
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index 938b206d8..bf9f05f90 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -37,7 +37,9 @@ import (
 	"knative.dev/eventing/pkg/scheduler"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
+	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	nodeinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -73,6 +75,7 @@ type envConfig struct {
 	PodCapacity                int32  `envconfig:"POD_CAPACITY" required:"true"`
 	SchedulerPolicyConfigMap   string `envconfig:"SCHEDULER_CONFIG" required:"true"`
 	DeSchedulerPolicyConfigMap string `envconfig:"DESCHEDULER_CONFIG" required:"true"`
+	AutoscalerConfigMap        string `envconfig:"AUTOSCALER_CONFIG" required:"true"`
 }
 
 type SchedulerConfig struct {
@@ -98,17 +101,19 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		DeSchedulerPolicy: schedulerPolicyFromConfigMapOrFail(ctx, env.DeSchedulerPolicyConfigMap),
 	}
 
-	schedulers := map[string]scheduler.Scheduler{
+	schedulers := map[string]Scheduler{
 		KafkaSourceScheduler: createKafkaScheduler(ctx, c, kafkainternals.SourceStatefulSetName),
 		//KafkaTriggerScheduler: createKafkaScheduler(ctx, c, kafkainternals.BrokerStatefulSetName), //To be added with trigger/v2 reconciler version only
 		//KafkaChannelScheduler: createKafkaScheduler(ctx, c, kafkainternals.ChannelStatefulSetName), //To be added with channel/v2 reconciler version only
 	}
 
 	r := &Reconciler{
-		SchedulerFunc:              func(s string) scheduler.Scheduler { return schedulers[strings.ToLower(s)] },
+		SchedulerFunc:              func(s string) Scheduler { return schedulers[strings.ToLower(s)] },
 		ConsumerLister:             consumer.Get(ctx).Lister(),
 		InternalsClient:            internalsclient.Get(ctx).InternalV1alpha1(),
 		SecretLister:               secretinformer.Get(ctx).Lister(),
+		ConfigMapLister:            configmapinformer.Get(ctx).Lister(),
+		PodLister:                  podinformer.Get(ctx).Lister(),
 		KubeClient:                 kubeclient.Get(ctx),
 		NameGenerator:              names.SimpleNameGenerator,
 		NewKafkaClient:             sarama.NewClient,
@@ -117,6 +122,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		KafkaFeatureFlags:          config.DefaultFeaturesConfig(),
 		KedaClient:                 kedaclient.Get(ctx),
+		AutoscalerConfig:           env.AutoscalerConfigMap,
 	}
 
 	consumerInformer := consumer.Get(ctx)
@@ -167,7 +173,7 @@ func enqueueConsumerGroupFromConsumer(enqueue func(name types.NamespacedName)) f
 	}
 }
 
-func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string) scheduler.Scheduler {
+func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string) Scheduler {
 	lister := consumergroup.Get(ctx).Lister()
 	return createStatefulSetScheduler(
 		ctx,
@@ -212,8 +218,8 @@ func getSelectorLabel(ssName string) map[string]string {
 	return selectorLabel
 }
 
-func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister scheduler.VPodLister) scheduler.Scheduler {
-	return statefulsetscheduler.NewScheduler(
+func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister scheduler.VPodLister) Scheduler {
+	ss := statefulsetscheduler.NewScheduler(
 		ctx,
 		system.Namespace(),
 		c.StatefulSetName,
@@ -226,6 +232,11 @@ func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister s
 		c.SchedulerPolicy,
 		c.DeSchedulerPolicy,
 	)
+
+	return Scheduler{
+		Scheduler:       ss,
+		SchedulerConfig: c,
+	}
 }
 
 // schedulerPolicyFromConfigMapOrFail reads predicates and priorities data from configMap


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 69b538908..3b0dbe541 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -52,8 +52,9 @@ import (
 	configapis "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 
-	kedaclient "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client/fake"
 	cm "knative.dev/pkg/configmap/testing"
+
+	kedaclient "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client/fake"
 )
 
 type SchedulerFunc func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error)
@@ -160,6 +161,93 @@ func TestReconcileKind(t *testing.T) {
 				finalizerUpdatedEvent,
 			},
 		},
+		{
+			Name: "Consumers in multiple pods, with pods pending and unknown phase",
+			Objects: []runtime.Object{
+				NewService(),
+				NewDispatcherPod("p1", PodLabel(kafkainternals.SourceStatefulSetName), PodPending()),
+				NewDispatcherPod("p2", PodLabel(kafkainternals.SourceStatefulSetName)),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+					ConsumerGroupReplicas(2),
+					ConsumerForTrigger(),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+						{PodName: "p2", VReplicas: 1},
+					}, nil
+				}),
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(systemNamespace, "p1", nil),
+				NewConfigMapWithBinaryData(systemNamespace, "p2", nil),
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+					)),
+				),
+				NewConsumer(2,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p2", PodNamespace: systemNamespace}),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+							)),
+							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(0),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 1},
+						}
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
+						return cg
+					}(),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			SkipNamespaceValidation: true,
+		},
 		{
 			Name: "Consumers in multiple pods, with auth spec, one exists - secret not found",
 			Objects: []runtime.Object{
@@ -1555,12 +1643,20 @@ func TestReconcileKind(t *testing.T) {
 		store.OnConfigChanged(exampleConfig)
 
 		r := Reconciler{
-			SchedulerFunc: func(s string) scheduler.Scheduler {
-				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			SchedulerFunc: func(s string) Scheduler {
+				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+				return Scheduler{
+					Scheduler: ss,
+					SchedulerConfig: SchedulerConfig{
+						StatefulSetName: kafkainternals.SourceStatefulSetName,
+					},
+				}
 			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
+			PodLister:       listers.GetPodLister(),
 			KubeClient:      kubeclient.Get(ctx),
 			KedaClient:      kedaclient.Get(ctx),
 			NameGenerator:   &CounterGenerator{},
@@ -1575,7 +1671,8 @@ func TestReconcileKind(t *testing.T) {
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
-			SystemNamespace: systemNamespace,
+			SystemNamespace:  systemNamespace,
+			AutoscalerConfig: "",
 		}
 
 		r.KafkaFeatureFlags = configapis.FromContext(store.ToContext(ctx))
@@ -1688,12 +1785,17 @@ func TestReconcileKindNoAutoscaler(t *testing.T) {
 		ctx, _ = kedaclient.With(ctx)
 
 		r := Reconciler{
-			SchedulerFunc: func(s string) scheduler.Scheduler {
-				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			SchedulerFunc: func(s string) Scheduler {
+				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+				return Scheduler{
+					Scheduler: ss,
+				}
 			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
+			PodLister:       listers.GetPodLister(),
 			KubeClient:      kubeclient.Get(ctx),
 			KedaClient:      kedaclient.Get(ctx),
 			NameGenerator:   &CounterGenerator{},
@@ -1921,12 +2023,17 @@ func TestFinalizeKind(t *testing.T) {
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		r := &Reconciler{
-			SchedulerFunc: func(s string) scheduler.Scheduler {
-				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			SchedulerFunc: func(s string) Scheduler {
+				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+				return Scheduler{
+					Scheduler: ss,
+				}
 			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
+			PodLister:       listers.GetPodLister(),
 			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
 				return &kafkatesting.MockKafkaClient{}, nil
 			},
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index 80a1148de..33e1d51ac 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -26,6 +26,7 @@ import (
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
@@ -47,6 +48,8 @@ const (
 	ConfigKafkaSchedulerName = "config-kafka-scheduler"
 	// ConfigKafkaDeSchedulerName is the name of the ConfigMap to configure the descheduler.
 	ConfigKafkaDeSchedulerName = "config-kafka-descheduler"
+	// ConfigKafkaAutoscalerName is the name of the ConfigMap to configure the autoscaler.
+	ConfigKafkaAutoscalerName = "config-kafka-autoscaler"
 )
 
 func TestNewController(t *testing.T) {
@@ -96,6 +99,7 @@ func TestNewController(t *testing.T) {
 	t.Setenv("POD_CAPACITY", PodCapacity)
 	t.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
 	t.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
+	t.Setenv("AUTOSCALER_CONFIG", ConfigKafkaAutoscalerName)
 	controller := NewController(ctx, configmap.NewStaticWatcher(&corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "config-kafka-features",
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index abcb3f8a2..b279d4f23 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -40,7 +40,7 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
-	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	kafkaeventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
@@ -497,7 +497,7 @@ func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
 		},
 		Spec: corev1.PodSpec{
 			Volumes: []corev1.Volume{{
-				Name: kafkainternals.DispatcherVolumeName,
+				Name: kafkaeventing.DispatcherVolumeName,
 				VolumeSource: corev1.VolumeSource{
 					ConfigMap: &corev1.ConfigMapVolumeSource{
 						LocalObjectReference: corev1.LocalObjectReference{
@@ -516,6 +516,21 @@ func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
 	return p
 }
 
+func PodLabel(value string) PodOption {
+	return func(pod *corev1.Pod) {
+		if pod.Labels == nil {
+			pod.Labels = make(map[string]string, 2)
+		}
+		pod.Labels["app"] = value
+	}
+}
+
+func PodPending() PodOption {
+	return func(pod *corev1.Pod) {
+		pod.Status.Phase = corev1.PodPending
+	}
+}
+
 func DispatcherPodAsOwnerReference(name string) reconcilertesting.ConfigMapOption {
 	d := NewDispatcherPod(name)
 	return func(configMap *corev1.ConfigMap) {
diff --git a/test/config/sacura/resources/200-config.yaml b/test/config/sacura/resources/200-config.yaml
index f5d260761..f4ecc6f53 100644
--- a/test/config/sacura/resources/200-config.yaml
+++ b/test/config/sacura/resources/200-config.yaml
@@ -13,5 +13,6 @@ data:
     receiver:
       port: 8080
       timeout: 5m
+      maxDuplicatesPercentage: 0.1
     duration: 10m
     timeout: 5m
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 83a24b42c..265ec77e7 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20221019-07af4e8e1"
+    kafka.eventing.knative.dev/release: "v20221124-b819e4d6a"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index 35bd1b080..d4e08c3b3 100644
--- a/third_party/eventing-kafka-latest/source-crd.yaml
+++ b/third_party/eventing-kafka-latest/source-crd.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20221019-07af4e8e1"
+    kafka.eventing.knative.dev/release: "v20221124-b819e4d6a"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 7e000feba..ba7528866 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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -436,10 +436,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -454,9 +454,6 @@ data:
   # ALPHA feature: The kreference-mapping allows you to map kreference onto templated URI
   # For more details: https://github.com/knative/eventing/issues/5593
   kreference-mapping: "disabled"
-  # BETA feature: The subscriber-strict flag force subscriptions to define a subscriber
-  # For more details: https://github.com/knative/eventing/issues/5756
-  strict-subscriber: "enabled"
   # ALPHA feature: The new-trigger-filters flag allows you to use the new `filters` field
   # in Trigger objects with its rich filtering capabilities.
   # For more details: https://github.com/knative/eventing/issues/5204
@@ -483,7 +480,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -532,8 +529,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -594,10 +591,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -648,10 +645,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -725,8 +722,8 @@ metadata:
   name: config-sugar
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "62dfac6f"
@@ -783,10 +780,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -841,10 +838,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -854,9 +851,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20221113-82bf83cd7"
+        eventing.knative.dev/release: "v20221129-fb519df96"
         app.kubernetes.io/component: eventing-controller
-        app.kubernetes.io/version: "20221113-82bf83cd7"
+        app.kubernetes.io/version: "20221129-fb519df96"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -874,7 +871,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:46b7a46edfb79fbf94d653fac3dcc6b8e50b45b1442a23290261b74de85794e6
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:d499e1523f52f1e488fe297edc8ff9854f7f0ce4d98b150d8e1f9af2e27e63f5
           resources:
             requests:
               cpu: 100m
@@ -892,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:b0a505eb3ddf358d3df8a6a74483786e9db4dfdedc44907481cef8970e25983d
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:991876752b008128d26b67a6de72bc889ce0915fcfb67bf0590d87a5920eba9c
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -961,9 +958,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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.
@@ -976,9 +973,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20221113-82bf83cd7"
+        eventing.knative.dev/release: "v20221129-fb519df96"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20221113-82bf83cd7"
+        app.kubernetes.io/version: "20221129-fb519df96"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -992,7 +989,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:ae69594c2d0c21ddb717ea6e6ba95fcbd048358313fa42ba47d5490d576e7868
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:87c2ef5f1a2c546928a3968ea53ef638ca28f76468772b6dbd47b64fc57cbda7
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -1065,9 +1062,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1091,9 +1088,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1122,9 +1119,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1135,9 +1132,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20221113-82bf83cd7"
+        eventing.knative.dev/release: "v20221129-fb519df96"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20221113-82bf83cd7"
+        app.kubernetes.io/version: "20221129-fb519df96"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1157,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:444759c0d2ee3461ae602f08c9b94ca3a3f0918424946622b0b45f770d53cadb
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:e7ebd99e0fb7ea8339e8265ec99df0c220d8169575e98d56c6072cf933cd9551
           resources:
             requests:
               # taken from serving.
@@ -1228,10 +1225,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1263,11 +1260,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1479,10 +1476,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1646,11 +1643,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1921,11 +1918,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -2075,9 +2072,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2204,10 +2201,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2459,11 +2456,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2650,10 +2647,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2970,12 +2967,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3162,9 +3159,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3374,9 +3371,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3560,8 +3557,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3574,9 +3571,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3594,9 +3591,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3617,9 +3614,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3644,9 +3641,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3665,9 +3662,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3703,8 +3700,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3730,8 +3727,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3748,8 +3745,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3782,8 +3779,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3796,9 +3793,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3836,9 +3833,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3850,9 +3847,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3864,9 +3861,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3878,9 +3875,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3892,9 +3889,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3907,8 +3904,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3921,8 +3918,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3949,8 +3946,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4085,8 +4082,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4153,8 +4150,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4167,9 +4164,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4215,8 +4212,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4229,9 +4226,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4267,8 +4264,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4354,8 +4351,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4454,8 +4451,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4491,8 +4488,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4529,8 +4526,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4563,8 +4560,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4598,8 +4595,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4623,8 +4620,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221113-82bf83cd7"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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 d27ab9958..ec85a00e4 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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     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: "v20221113-82bf83cd7"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221113-82bf83cd7"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev