Repository: knative-sandbox/eventing-kafka-broker
Pull request #2816 (2022-11-30 11:40:05)

Handle ConsumerGroup finalization for deleted topics
When a topic is deleted before a ConsumerGroup, we should
not fail to finalize the ConsumerGroup when deleting the
Consumer group in Kafka.

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/2816#" 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":[]} -->
pierDipi: /cc @aavarghese @matzew 
knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 0a9ee32c27b4bfe58c5abfa107ed847ed253b78e | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2816/unit-tests_eventing-kafka-broker_main/1597918389655834624) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2816](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0a9ee32) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/2ae4aa33329b13b763a8de5d7559da95d725b7ad?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2ae4aa3) will **increase** coverage by `19.39%`.
> The diff coverage is `n/a`.

```diff
@@              Coverage Diff              @@
##               main    #2816       +/-   ##
=============================================
+ Coverage     61.58%   80.98%   +19.39%     
+ Complexity      749      747        -2     
=============================================
  Files           153       77       -76     
  Lines         10767     2619     -8148     
  Branches        231      231               
=============================================
- Hits           6631     2121     -4510     
+ Misses         3668      357     -3311     
+ Partials        468      141      -327     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `80.98% <ø> (-0.39%)` | :arrow_down: |

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/2816?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | `74.68% <0.00%> (-6.33%)` | :arrow_down: |
| [...roker/dispatcher/main/ConsumerVerticleBuilder.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlQnVpbGRlci5qYXZh) | `65.95% <0.00%> (-4.26%)` | :arrow_down: |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | `95.69% <0.00%> (-1.08%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | | |
| [...ontrol-plane/pkg/reconciler/consumer/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb250cm9sbGVyLmdv) | | |
| [...kafka/eventing/v1alpha1/consumer\_group\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfZGVmYXVsdHMuZ28=) | | |
| [control-plane/pkg/contract/log\_marshaller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvbG9nX21hcnNoYWxsZXIuZ28=) | | |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | | |
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | | |
| [...als/kafka/eventing/v1alpha1/consumer\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdmFsaWRhdGlvbi5nbw==) | | |
| ... and [69 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2816/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)

pierDipi: /test unit-tests

Flake? 
`{Failed  === RUN   TestReconcileKind/Consumers_in_multiple_pods,_with_pods_pending_and_unknown_phase`
https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2816/unit-tests_eventing-kafka-broker_main/1597918389655834624
knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | dbcee0cef0e60b9b8f58ee33837983aaa7345877 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2816/unit-tests_eventing-kafka-broker_main/1597922570059911168) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: /test unit-tests
pierDipi: /cherry-pick release-1.8
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.8 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2816#issuecomment-1332140070):

>/cherry-pick release-1.8


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
pierDipi: /cherry-pick release-1.7
pierDipi: /cherry-pick release-1.6

Patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 7f90bce64..417c5a5f6 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -187,12 +187,11 @@ func (r Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.Consume
 	defer kafkaClusterAdminClient.Close()
 
 	groupId := cg.Spec.Template.Spec.Configs.Configs["group.id"]
-	if err := kafkaClusterAdminClient.DeleteConsumerGroup(groupId); err != nil && !errors.Is(sarama.ErrGroupIDNotFound, err) {
-		logging.FromContext(ctx).Errorw("unable to delete the consumer group", zap.String("id", groupId), zap.Error(err))
-		return err
+	if err = kafkaClusterAdminClient.DeleteConsumerGroup(groupId); err != nil && !errorIsOneOf(err, sarama.ErrUnknownTopicOrPartition, sarama.ErrGroupIDNotFound) {
+		return fmt.Errorf("unable to delete the consumer group %s: %w", groupId, err)
 	}
 
-	logging.FromContext(ctx).Infow("consumer group deleted", zap.String("id", groupId))
+	logging.FromContext(ctx).Debug("consumer group deleted", zap.String("id", groupId))
 	return nil
 }
 
@@ -667,6 +666,16 @@ func (r Reconciler) ensureContractConfigMapExists(ctx context.Context, p *corev1
 	return nil
 }
 
+func errorIsOneOf(err error, errs ...error) bool {
+	for _, e := range errs {
+		if errors.Is(err, e) {
+			return true
+		}
+	}
+
+	return false
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 	_ consumergroup.Finalizer = &Reconciler{}
diff --git a/vendor/github.com/manifestival/manifestival/fake/client.go b/vendor/github.com/manifestival/manifestival/fake/client.go
new file mode 100644
index 000000000..60e9c31be
--- /dev/null
+++ b/vendor/github.com/manifestival/manifestival/fake/client.go
@@ -0,0 +1,102 @@
+package fake
+
+import (
+	"fmt"
+
+	mf "github.com/manifestival/manifestival"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+var _ mf.Client = &Client{}
+
+// A convenient way to stub out a Client for test fixtures. Default
+// behavior does nothing and returns a nil error.
+type Client struct {
+	Stubs
+}
+
+// Override any of the Client functions
+type Stubs struct {
+	Create mutator
+	Update mutator
+	Delete mutator
+	Get    accessor
+}
+
+type mutator func(obj *unstructured.Unstructured) error
+type accessor func(obj *unstructured.Unstructured) (*unstructured.Unstructured, error)
+
+// New returns a fully-functioning Client, "persisting" resources in a
+// map, optionally initialized with some API objects
+func New(objs ...runtime.Object) Client {
+	store := map[string]*unstructured.Unstructured{}
+	key := func(u *unstructured.Unstructured) string {
+		return fmt.Sprintf("%s, %s/%s", u.GroupVersionKind(), u.GetNamespace(), u.GetName())
+	}
+	for _, obj := range objs {
+		u := &unstructured.Unstructured{}
+		if err := scheme.Scheme.Convert(obj, u, nil); err != nil {
+			panic(err)
+		}
+		store[key(u)] = u
+	}
+	apply := func(u *unstructured.Unstructured) error {
+		store[key(u)] = u
+		return nil
+	}
+	return Client{
+		Stubs{
+			Create: apply,
+			Update: apply,
+			Delete: func(u *unstructured.Unstructured) error {
+				delete(store, key(u))
+				return nil
+			},
+			Get: func(u *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+				v, found := store[key(u)]
+				if !found {
+					gvk := u.GroupVersionKind()
+					gr := schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}
+					return nil, errors.NewNotFound(gr, u.GetName())
+				}
+				return v, nil
+			},
+		},
+	}
+}
+
+// Manifestival.Client.Create
+func (c Client) Create(obj *unstructured.Unstructured, options ...mf.ApplyOption) error {
+	if c.Stubs.Create != nil {
+		return c.Stubs.Create(obj)
+	}
+	return nil
+}
+
+// Manifestival.Client.Update
+func (c Client) Update(obj *unstructured.Unstructured, options ...mf.ApplyOption) error {
+	if c.Stubs.Update != nil {
+		return c.Stubs.Update(obj)
+	}
+	return nil
+}
+
+// Manifestival.Client.Delete
+func (c Client) Delete(obj *unstructured.Unstructured, options ...mf.DeleteOption) error {
+	if c.Stubs.Delete != nil {
+		return c.Stubs.Delete(obj)
+	}
+	return nil
+}
+
+// Manifestival.Client.Get
+func (c Client) Get(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	if c.Stubs.Get != nil {
+		return c.Stubs.Get(obj)
+	}
+	return nil, nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index a78c282d0..07d626cba 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -329,6 +329,7 @@ github.com/manifestival/client-go-client/pkg/dynamic
 # github.com/manifestival/manifestival v0.7.1
 ## explicit; go 1.15
 github.com/manifestival/manifestival
+github.com/manifestival/manifestival/fake
 github.com/manifestival/manifestival/internal/overlay
 github.com/manifestival/manifestival/internal/patch
 github.com/manifestival/manifestival/internal/sources


Test patch:
diff --git a/control-plane/pkg/kafka/testing/admin_mock.go b/control-plane/pkg/kafka/testing/admin_mock.go
index edb857529..5b7b2c096 100644
--- a/control-plane/pkg/kafka/testing/admin_mock.go
+++ b/control-plane/pkg/kafka/testing/admin_mock.go
@@ -26,6 +26,10 @@ import (
 
 var _ sarama.ClusterAdmin = &MockKafkaClusterAdmin{}
 
+const (
+	ErrorOnDeleteConsumerGroupTestKey = "error-on-delete-consumer-group"
+)
+
 type MockKafkaClusterAdmin struct {
 	// (Create|Delete)Topic
 	ExpectedTopicName string
@@ -45,6 +49,8 @@ type MockKafkaClusterAdmin struct {
 	ExpectedErrorOnDescribeTopics          error
 	ExpectedTopicsMetadataOnDescribeTopics []*sarama.TopicMetadata
 
+	ErrorOnDeleteConsumerGroup error
+
 	T *testing.T
 }
 
@@ -166,7 +172,7 @@ func (m *MockKafkaClusterAdmin) ListConsumerGroupOffsets(group string, topicPart
 }
 
 func (m *MockKafkaClusterAdmin) DeleteConsumerGroup(group string) error {
-	return nil
+	return m.ErrorOnDeleteConsumerGroup
 }
 
 func (m *MockKafkaClusterAdmin) DescribeCluster() (brokers []*sarama.Broker, controllerID int32, err error) {
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index b2a6d4344..16864013a 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -90,13 +90,14 @@ var (
 )
 
 var DefaultEnv = &config.Env{
-	DataPlaneConfigMapNamespace: "knative-eventing",
-	ContractConfigMapName:       "kafka-broker-brokers-triggers",
-	GeneralConfigMapName:        "kafka-broker-config",
-	IngressName:                 "kafka-broker-ingress",
-	SystemNamespace:             "knative-eventing",
-	ContractConfigMapFormat:     base.Json,
-	DefaultBackoffDelayMs:       1000,
+	DataPlaneConfigMapNamespace:  "knative-eventing",
+	DataPlaneConfigConfigMapName: "config-kafka-broker-data-plane",
+	ContractConfigMapName:        "kafka-broker-brokers-triggers",
+	GeneralConfigMapName:         "kafka-broker-config",
+	IngressName:                  "kafka-broker-ingress",
+	SystemNamespace:              "knative-eventing",
+	ContractConfigMapFormat:      base.Json,
+	DefaultBackoffDelayMs:        1000,
 }
 
 func TestBrokerReconciler(t *testing.T) {
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
new file mode 100644
index 000000000..7d8695076
--- /dev/null
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -0,0 +1,320 @@
+/*
+ * Copyright 2020 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 broker_test // different package name due to import cycles. (broker -> testing -> broker)
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	"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/kafka"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
+
+	mffake "github.com/manifestival/manifestival/fake"
+
+	"github.com/Shopify/sarama"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
+	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+)
+
+func TestNamespacedBrokerReconciler(t *testing.T) {
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
+
+	t.Parallel()
+
+	for _, f := range Formats {
+		namespacedBrokerReconciliation(t, f, *DefaultEnv)
+	}
+}
+
+func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env) {
+
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
+
+	env.ContractConfigMapFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewNamespacedBroker(
+					WithBrokerConfig(
+						KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
+					),
+				),
+				BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace)),
+				DataPlaneConfigMap(SystemNamespace, env.DataPlaneConfigConfigMapName, ConsumerConfigKey,
+					DataPlaneConfigInitialOffset(ConsumerConfigKey, sources.OffsetLatest),
+				),
+				reconcilertesting.NewConfigMap("config-tracing", SystemNamespace),
+				reconcilertesting.NewConfigMap("kafka-config-logging", SystemNamespace),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				NewService(),
+				BrokerReceiverPod(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPod(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				reconcilertesting.NewDeployment("kafka-broker-receiver", SystemNamespace),
+				reconcilertesting.NewDeployment("kafka-broker-dispatcher", SystemNamespace),
+				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
+				NewClusterRoleBinding("knative-kafka-broker-data-plane",
+					WithSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane", "knative-kafka-broker-data-plane"),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(BrokerNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat,
+					&contract.Contract{
+						Resources: []*contract.Resource{
+							{
+								Uid:              BrokerUUID,
+								Topics:           []string{BrokerTopic()},
+								Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+								BootstrapServers: bootstrapServers,
+								Reference:        BrokerReference(),
+							},
+						},
+						Generation: 1,
+					},
+					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
+					WithConfigmapOwnerRef(&metav1.OwnerReference{
+						APIVersion:         eventing.SchemeGroupVersion.String(),
+						Kind:               "Broker",
+						Name:               BrokerName,
+						UID:                BrokerUUID,
+						Controller:         pointer.Bool(false),
+						BlockOwnerDeletion: pointer.Bool(true),
+					}),
+				),
+				BrokerReceiverPodUpdate(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewNamespacedBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						WithBrokerConfig(
+							KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
+						),
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusBrokerTopicReady,
+						NamespacedBrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerConfigMapAnnotations(),
+						WithTopicStatusAnnotation(BrokerTopic()),
+					),
+				},
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTableNamespaced(t, table, &env)
+}
+
+func TestNamespacedBrokerFinalizer(t *testing.T) {
+	t.Parallel()
+
+	for _, f := range Formats {
+		namespacedBrokerFinalization(t, f, *DefaultEnv)
+	}
+}
+
+func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
+
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
+
+	env.ContractConfigMapFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewDeletedBroker(reconcilertesting.WithBrokerClass(kafka.NamespacedBrokerClass)),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+						},
+					},
+					Generation: 1,
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTableNamespaced(t, table, &env)
+}
+
+func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
+
+	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		defaultTopicDetail := sarama.TopicDetail{
+			NumPartitions:     DefaultNumPartitions,
+			ReplicationFactor: DefaultReplicationFactor,
+		}
+
+		var onCreateTopicError error
+		if want, ok := row.OtherTestData[wantErrorOnCreateTopic]; ok {
+			onCreateTopicError = want.(error)
+		}
+
+		var onDeleteTopicError error
+		if want, ok := row.OtherTestData[wantErrorOnDeleteTopic]; ok {
+			onDeleteTopicError = want.(error)
+		}
+
+		expectedTopicDetail := defaultTopicDetail
+		if td, ok := row.OtherTestData[ExpectedTopicDetail]; ok {
+			expectedTopicDetail = td.(sarama.TopicDetail)
+		}
+
+		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+		if t, ok := row.OtherTestData[externalTopic]; ok {
+			expectedTopicName = t.(string)
+		}
+
+		var metadata []*sarama.TopicMetadata
+		metadata = append(metadata, &sarama.TopicMetadata{
+			Name:       ExternalTopicName,
+			IsInternal: false,
+			Partitions: []*sarama.PartitionMetadata{{}},
+		})
+
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
+		mfcMockClient := mffake.New()
+
+		reconciler := &NamespacedReconciler{
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				SecretLister:                listers.GetSecretLister(),
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				ContractConfigMapName:       env.ContractConfigMapName,
+				ContractConfigMapFormat:     env.ContractConfigMapFormat,
+				DataPlaneNamespace:          env.SystemNamespace,
+				DispatcherLabel:             base.BrokerDispatcherLabel,
+				ReceiverLabel:               base.BrokerReceiverLabel,
+			},
+			ConfigMapLister:          listers.GetConfigMapLister(),
+			DeploymentLister:         listers.GetDeploymentLister(),
+			ServiceAccountLister:     listers.GetServiceAccountLister(),
+			ServiceLister:            listers.GetServiceLister(),
+			ClusterRoleBindingLister: listers.GetClusterRoleBindingLister(),
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopicName:                      expectedTopicName,
+					ExpectedTopicDetail:                    expectedTopicDetail,
+					ErrorOnCreateTopic:                     onCreateTopicError,
+					ErrorOnDeleteTopic:                     onDeleteTopicError,
+					ExpectedTopics:                         []string{expectedTopicName},
+					ExpectedTopicsMetadataOnDescribeTopics: metadata,
+					T:                                      t,
+				}, nil
+			},
+			Env:                env,
+			Prober:             proberMock,
+			ManifestivalClient: mfcMockClient,
+		}
+
+		reconciler.ConfigMapTracker = &FakeTracker{}
+		reconciler.SecretTracker = &FakeTracker{}
+
+		r := brokerreconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingclient.Get(ctx),
+			listers.GetBrokerLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+			kafka.NamespacedBrokerClass,
+		)
+
+		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
+		reconciler.IPsLister = prober.NewIPListerWithMapping()
+
+		return r
+	}))
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 3b0dbe541..79ff0fd70 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -446,7 +446,7 @@ func TestReconcileKind(t *testing.T) {
 							{PodName: "p1", VReplicas: 1},
 							{PodName: "p2", VReplicas: 1},
 						}
-						cg.MarkReconcileConsumersFailedCondition(&apis.Condition{
+						_ = cg.MarkReconcileConsumersFailedCondition(&apis.Condition{
 							Type:    kafkainternals.ConsumerConditionBind,
 							Status:  corev1.ConditionFalse,
 							Reason:  "ConsumerBinding",
@@ -2017,11 +2017,161 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
 		},
-		//need to add support for DeleteConsumerGroup in admin_mock.go before adding more testcases
+		{
+			Name: "Finalize normal - failed consumer group deletion with " + sarama.ErrUnknownTopicOrPartition.Error(),
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrUnknownTopicOrPartition,
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
+		{
+			Name: "Finalize normal - failed consumer group deletion with " + sarama.ErrGroupIDNotFound.Error(),
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrGroupIDNotFound,
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
+		{
+			Name: "Finalize error - failed consumer group deletion with " + sarama.ErrClusterAuthorizationFailed.Error(),
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			WantErr: true,
+			Key:     testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrClusterAuthorizationFailed,
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"unable to delete the consumer group my.group.id: "+sarama.ErrClusterAuthorizationFailed.Error(),
+				),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
 	}
 
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
+		errorOnDeleteKafkaCG := row.OtherTestData[kafkatesting.ErrorOnDeleteConsumerGroupTestKey]
+
 		r := &Reconciler{
 			SchedulerFunc: func(s string) Scheduler {
 				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
@@ -2039,7 +2189,8 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
-					T: t,
+					T:                          t,
+					ErrorOnDeleteConsumerGroup: ErrorAssertOrNil(errorOnDeleteKafkaCG),
 				}, nil
 			},
 			KafkaFeatureFlags: configapis.DefaultFeaturesConfig(),
diff --git a/control-plane/pkg/reconciler/testing/errors.go b/control-plane/pkg/reconciler/testing/errors.go
new file mode 100644
index 000000000..2ed85491b
--- /dev/null
+++ b/control-plane/pkg/reconciler/testing/errors.go
@@ -0,0 +1,25 @@
+/*
+ * 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 testing
+
+// ErrorAssertOrNil returns an error given an interface{} variable
+func ErrorAssertOrNil(e interface{}) error {
+	if e == nil {
+		return nil
+	}
+	return e.(error)
+}
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 31a261c78..7169afed7 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -17,11 +17,15 @@
 package testing
 
 import (
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	fakeapiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	appslisters "k8s.io/client-go/listers/apps/v1"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	rbaclisters "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
@@ -114,6 +118,10 @@ func (l *Listers) GetPodLister() corelisters.PodLister {
 	return corelisters.NewPodLister(l.indexerFor(&corev1.Pod{}))
 }
 
+func (l *Listers) GetDeploymentLister() appslisters.DeploymentLister {
+	return appslisters.NewDeploymentLister(l.indexerFor(&appsv1.Deployment{}))
+}
+
 func (l *Listers) GetSecretLister() corelisters.SecretLister {
 	return corelisters.NewSecretLister(l.indexerFor(&corev1.Secret{}))
 }
@@ -134,6 +142,10 @@ func (l *Listers) GetServiceLister() corelisters.ServiceLister {
 	return corelisters.NewServiceLister(l.indexerFor(&corev1.Service{}))
 }
 
+func (l *Listers) GetServiceAccountLister() corelisters.ServiceAccountLister {
+	return corelisters.NewServiceAccountLister(l.indexerFor(&corev1.ServiceAccount{}))
+}
+
 func (l *Listers) GetKafkaSinkLister() eventingkafkabrokerlisters.KafkaSinkLister {
 	return eventingkafkabrokerlisters.NewKafkaSinkLister(l.indexerFor(&eventingkafkabroker.KafkaSink{}))
 }
@@ -154,6 +166,10 @@ func (l *Listers) GetConsumerLister() consumerlisters.ConsumerLister {
 	return consumerlisters.NewConsumerLister(l.indexerFor(&eventingkafkabrokerconsumer.Consumer{}))
 }
 
+func (l *Listers) GetClusterRoleBindingLister() rbaclisters.ClusterRoleBindingLister {
+	return rbaclisters.NewClusterRoleBindingLister(l.indexerFor(&rbacv1.ClusterRoleBinding{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 8194717e4..2af9d7705 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -219,6 +219,12 @@ func WithAutoscalingAnnotationsTrigger() reconcilertesting.TriggerOption {
 
 type CMOption func(cm *corev1.ConfigMap)
 
+func WithConfigMapNamespace(namespace string) CMOption {
+	return func(cm *corev1.ConfigMap) {
+		cm.ObjectMeta.Namespace = namespace
+	}
+}
+
 func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int, options ...CMOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -299,17 +305,25 @@ func StatusBrokerConfigNotParsed(reason string) func(broker *eventing.Broker) {
 }
 
 func BrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
-
 	return func(broker *eventing.Broker) {
+		brokerAddressable(broker, env.IngressName, env.SystemNamespace)
+	}
+}
 
-		broker.Status.Address.URL = &apis.URL{
-			Scheme: "http",
-			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
-		}
+func NamespacedBrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		brokerAddressable(broker, env.IngressName, broker.Namespace)
+	}
+}
 
-		broker.GetConditionSet().Manage(&broker.Status).MarkTrue(base.ConditionAddressable)
+func brokerAddressable(broker *eventing.Broker, serviceName, serviceNamespace string) {
+	broker.Status.Address.URL = &apis.URL{
+		Scheme: "http",
+		Host:   network.GetServiceHostname(serviceName, serviceNamespace),
+		Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
 	}
+
+	broker.GetConditionSet().Manage(&broker.Status).MarkTrue(base.ConditionAddressable)
 }
 
 func BrokerReference() *contract.Reference {
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index b279d4f23..5667432a1 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -158,6 +158,39 @@ func NewRoleBinding(namespace string, name string, mutations ...func(account *rb
 	return s
 }
 
+func WithSubjectServiceAccount(saNamespace, saName, roleName string) func(account *rbacv1.ClusterRoleBinding) {
+	return func(cb *rbacv1.ClusterRoleBinding) {
+		cb.Subjects = []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: saNamespace,
+				Name:      saName,
+			},
+		}
+		cb.RoleRef = rbacv1.RoleRef{
+			Kind:     "ClusterRole",
+			Name:     roleName,
+			APIGroup: "rbac.authorization.k8s.io",
+		}
+	}
+}
+
+func NewClusterRoleBinding(name string, mutations ...func(account *rbacv1.ClusterRoleBinding)) *rbacv1.ClusterRoleBinding {
+	s := &rbacv1.ClusterRoleBinding{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ClusterRoleBinding",
+			APIVersion: rbacv1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+	for _, mut := range mutations {
+		mut(s)
+	}
+	return s
+}
+
 func NewConfigMapWithBinaryData(namespace string, name string, data []byte, options ...reconcilertesting.ConfigMapOption) runtime.Object {
 	return reconcilertesting.NewConfigMap(
 		name,
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 8483db71b..e354b2de5 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -19,6 +19,8 @@ package testing
 import (
 	"time"
 
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -229,3 +231,9 @@ func NewDeletedConsumeGroup(opts ...ConsumerGroupOption) runtime.Object {
 func WithDeletedTimeStampConsumeGroup(cg *kafkainternals.ConsumerGroup) {
 	cg.GetObjectMeta().SetDeletionTimestamp(&metav1.Time{Time: time.Now()})
 }
+
+func WithConfigmapOwnerRef(ownerref *metav1.OwnerReference) reconcilertesting.ConfigMapOption {
+	return func(cg *corev1.ConfigMap) {
+		cg.ObjectMeta.OwnerReferences = []metav1.OwnerReference{*ownerref}
+	}
+}
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-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 803951582..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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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.
@@ -480,7 +480,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -529,8 +529,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -591,10 +591,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -645,10 +645,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -722,8 +722,8 @@ metadata:
   name: config-sugar
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "62dfac6f"
@@ -780,10 +780,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -838,10 +838,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -851,9 +851,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20221123-41653ce64"
+        eventing.knative.dev/release: "v20221129-fb519df96"
         app.kubernetes.io/component: eventing-controller
-        app.kubernetes.io/version: "20221123-41653ce64"
+        app.kubernetes.io/version: "20221129-fb519df96"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -871,7 +871,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:74de26003602f29ab533830f6dc9792675aad5c9b201011a1956c1cb0bb9705b
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:d499e1523f52f1e488fe297edc8ff9854f7f0ce4d98b150d8e1f9af2e27e63f5
           resources:
             requests:
               cpu: 100m
@@ -889,7 +889,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:ab3fdfc1cd2f49a716aa2430607d731fdf8ae05c5420620a3e7502ca3e58ed36
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:991876752b008128d26b67a6de72bc889ce0915fcfb67bf0590d87a5920eba9c
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -958,9 +958,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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.
@@ -973,9 +973,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20221123-41653ce64"
+        eventing.knative.dev/release: "v20221129-fb519df96"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20221123-41653ce64"
+        app.kubernetes.io/version: "20221129-fb519df96"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -989,7 +989,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:c06ee9df08be3ec309a27428799fcb23347055694bbb7bc5aa848d329d46db8c
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:87c2ef5f1a2c546928a3968ea53ef638ca28f76468772b6dbd47b64fc57cbda7
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -1062,9 +1062,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1088,9 +1088,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1119,9 +1119,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1132,9 +1132,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20221123-41653ce64"
+        eventing.knative.dev/release: "v20221129-fb519df96"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20221123-41653ce64"
+        app.kubernetes.io/version: "20221129-fb519df96"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1154,7 +1154,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:7754cd3f498f2224dbd71a2b2900a74b52360de3e17b7194894d9feae85bae11
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:e7ebd99e0fb7ea8339e8265ec99df0c220d8169575e98d56c6072cf933cd9551
           resources:
             requests:
               # taken from serving.
@@ -1225,10 +1225,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1260,11 +1260,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1476,10 +1476,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1643,11 +1643,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1918,11 +1918,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -2072,9 +2072,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2201,10 +2201,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2456,11 +2456,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2647,10 +2647,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2967,12 +2967,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3159,9 +3159,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3371,9 +3371,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3557,8 +3557,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3571,9 +3571,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -3591,9 +3591,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -3614,9 +3614,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -3641,9 +3641,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -3662,9 +3662,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -3700,8 +3700,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3727,8 +3727,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3745,8 +3745,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3779,8 +3779,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3793,9 +3793,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -3833,9 +3833,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3847,9 +3847,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3861,9 +3861,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3875,9 +3875,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3889,9 +3889,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3904,8 +3904,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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"]
@@ -3918,8 +3918,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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"]
@@ -3946,8 +3946,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4082,8 +4082,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4150,8 +4150,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4164,9 +4164,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -4212,8 +4212,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4226,9 +4226,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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:
@@ -4264,8 +4264,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4351,8 +4351,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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.
@@ -4451,8 +4451,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4488,8 +4488,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4526,8 +4526,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4560,8 +4560,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4595,8 +4595,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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.
 
@@ -4620,8 +4620,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20221123-41653ce64"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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 7e4c11bab..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: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    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: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    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: "v20221123-41653ce64"
+    eventing.knative.dev/release: "v20221129-fb519df96"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20221123-41653ce64"
+    app.kubernetes.io/version: "20221129-fb519df96"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev