Repository: knative-sandbox/eventing-kafka-broker
Pull request #1588 (2021-12-13 11:05:39)

[Automated] Update eventing-kafka-broker-eventing nightly
/assign @knative/delivery-wg-leads
/cc @knative/delivery-wg-leads

Produced via:
```shell
for x in eventing-crds.yaml eventing-core.yaml; do
  curl https://storage.googleapis.com/knative-nightly/eventing/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-latest/$x
done
```

Details:
```

```

knative-prow-robot: @knative-automation: GitHub didn't allow me to request PR reviews from the following users: knative/delivery-wg-leads.

Note that only [knative-sandbox members](https://github.com/orgs/knative-sandbox/people) and repo collaborators can review this PR, and authors cannot review their own PRs.

<details>

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

>/assign @knative/delivery-wg-leads
>/cc @knative/delivery-wg-leads
>
>Produced via:
>```shell
>for x in eventing-crds.yaml eventing-core.yaml; do
>  curl https://storage.googleapis.com/knative-nightly/eventing/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-latest/$x
>done
>```
>
>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.
</details>
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1588#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **evankanderson** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @evankanderson` in a comment when ready.

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

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

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

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1588      +/-   ##
============================================
- Coverage     74.87%   69.01%   -5.86%     
============================================
  Files           116       50      -66     
  Lines          4211     2282    -1929     
  Branches        171        0     -171     
============================================
- Hits           3153     1575    -1578     
+ Misses          802      541     -261     
+ Partials        256      166      -90     
```

| 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/1588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [...ka/broker/dispatcher/impl/NoopResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9Ob29wUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...ka/broker/dispatcher/impl/filter/PrefixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvUHJlZml4RmlsdGVyLmphdmE=) | | |
| [...re/reconciler/impl/ResourcesReconcilerBuilder.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJCdWlsZGVyLmphdmE=) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [...a/broker/dispatcher/impl/RecordDispatcherImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVySW1wbC5qYXZh) | | |
| [...cher/impl/consumer/CloudEventOverridesMutator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50T3ZlcnJpZGVzTXV0YXRvci5qYXZh) | | |
| [...oker/dispatcher/impl/consumer/KeyDeserializer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LZXlEZXNlcmlhbGl6ZXIuamF2YQ==) | | |
| ... and [56 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [fcf4140...859d54e](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1588?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).


Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index 085c86660..8c9f8b626 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -19,8 +19,11 @@ package v1alpha1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 )
 
 // +genclient
@@ -82,6 +85,14 @@ type ConsumerGroupStatus struct {
 
 	// inherits PlaceableStatus Status
 	eventingduckv1alpha1.PlaceableStatus `json:",inline"`
+
+	// SubscriberURI is the resolved URI of the receiver for this Trigger.
+	// +optional
+	SubscriberURI *apis.URL `json:"subscriberUri,omitempty"`
+
+	// DeliveryStatus contains a resolved URL to the dead letter sink address, and any other
+	// resolved delivery options.
+	eventingduckv1.DeliveryStatus `json:",inline"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
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 9c7936fbf..93284aba9 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
@@ -20,6 +20,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
@@ -75,7 +76,7 @@ type ConsumerSpec struct {
 type DeliverySpec struct {
 	// DeliverySpec is the Knative core delivery spec.
 	// DeliverySpec contains the delivery options for event senders.
-	eventingduck.DeliverySpec `json:",inline,omitempty"`
+	*eventingduck.DeliverySpec `json:",inline,omitempty"`
 
 	// Ordering is the ordering of the event delivery.
 	Ordering eventing.DeliveryOrdering `json:"ordering"`
@@ -111,6 +112,14 @@ type ConsumerStatus struct {
 	// * ObservedGeneration - the 'Generation' of the Consumer that was last processed by the controller.
 	// * Conditions - the latest available observations of a resource's current state.
 	duckv1.Status
+
+	// SubscriberURI is the resolved URI of the receiver for this Trigger.
+	// +optional
+	SubscriberURI *apis.URL `json:"subscriberUri,omitempty"`
+
+	// DeliveryStatus contains a resolved URL to the dead letter sink address, and any other
+	// resolved delivery options.
+	eventingduck.DeliveryStatus `json:",inline"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/filter_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/filter_types.go
index 92f9d5cef..3ddcc10c3 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/filter_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/filter_types.go
@@ -27,6 +27,12 @@ import (
 // If not specified, will default to allowing all events.
 type Filters struct {
 
+	// Filter is the filter to apply against all events from the Broker. Only events that pass this
+	// filter will be sent to the Subscriber. If not specified, will default to allowing all events.
+	//
+	// +optional
+	Filter *eventing.TriggerFilter `json:"filter,omitempty"`
+
 	// Filters is an experimental field that conforms to the CNCF CloudEvents Subscriptions
 	// API. It's an array of filter expressions that evaluate to true or false.
 	// If any filter expression in the array evaluates to false, the event MUST
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 479b4e94b..d2d8c9a78 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,7 +23,9 @@ package v1alpha1
 
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
-	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	apis "knative.dev/pkg/apis"
 )
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
@@ -172,6 +174,12 @@ func (in *ConsumerGroupStatus) DeepCopyInto(out *ConsumerGroupStatus) {
 	*out = *in
 	in.Status.DeepCopyInto(&out.Status)
 	in.PlaceableStatus.DeepCopyInto(&out.PlaceableStatus)
+	if in.SubscriberURI != nil {
+		in, out := &in.SubscriberURI, &out.SubscriberURI
+		*out = new(apis.URL)
+		(*in).DeepCopyInto(*out)
+	}
+	in.DeliveryStatus.DeepCopyInto(&out.DeliveryStatus)
 	return
 }
 
@@ -255,6 +263,12 @@ func (in *ConsumerSpec) DeepCopy() *ConsumerSpec {
 func (in *ConsumerStatus) DeepCopyInto(out *ConsumerStatus) {
 	*out = *in
 	in.Status.DeepCopyInto(&out.Status)
+	if in.SubscriberURI != nil {
+		in, out := &in.SubscriberURI, &out.SubscriberURI
+		*out = new(apis.URL)
+		(*in).DeepCopyInto(*out)
+	}
+	in.DeliveryStatus.DeepCopyInto(&out.DeliveryStatus)
 	return
 }
 
@@ -289,7 +303,11 @@ func (in *ConsumerTemplateSpec) DeepCopy() *ConsumerTemplateSpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 	*out = *in
-	in.DeliverySpec.DeepCopyInto(&out.DeliverySpec)
+	if in.DeliverySpec != nil {
+		in, out := &in.DeliverySpec, &out.DeliverySpec
+		*out = new(v1.DeliverySpec)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -306,9 +324,14 @@ func (in *DeliverySpec) DeepCopy() *DeliverySpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Filters) DeepCopyInto(out *Filters) {
 	*out = *in
+	if in.Filter != nil {
+		in, out := &in.Filter, &out.Filter
+		*out = new(eventingv1.TriggerFilter)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Filters != nil {
 		in, out := &in.Filters, &out.Filters
-		*out = make([]v1.SubscriptionsAPIFilter, len(*in))
+		*out = make([]eventingv1.SubscriptionsAPIFilter, len(*in))
 		for i := range *in {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index 27f53f742..e1614b381 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -35,6 +35,7 @@ import (
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
@@ -43,8 +44,6 @@ import (
 
 const (
 	deliveryOrderAnnotation = "kafka.eventing.knative.dev/delivery.order"
-	deliveryOrderOrdered    = "ordered"
-	deliveryOrderUnordered  = "unordered"
 )
 
 type Reconciler struct {
@@ -101,7 +100,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 
 		logger.Debug("broker deleted", zap.String("finalizeDuringReconcile", "deleted"))
 
-		// The associated broker doesn't exist anymore, so clean up Trigger resources.
+		// The associated broker doesn't exist anymore, so clean up Trigger resources and owning consumer group resource.
 		return r.FinalizeKind(ctx, trigger)
 	}
 
@@ -326,11 +325,11 @@ func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
 
 func deliveryOrderFromString(val string) (contract.DeliveryOrder, error) {
 	switch strings.ToLower(val) {
-	case deliveryOrderOrdered:
+	case string(internals.Ordered):
 		return contract.DeliveryOrder_ORDERED, nil
-	case deliveryOrderUnordered:
+	case string(internals.Unordered):
 		return contract.DeliveryOrder_UNORDERED, nil
 	default:
-		return contract.DeliveryOrder_UNORDERED, fmt.Errorf("invalid annotation %s value: %s. Allowed values [ %q | %q ]", deliveryOrderAnnotation, val, deliveryOrderOrdered, deliveryOrderUnordered)
+		return contract.DeliveryOrder_UNORDERED, fmt.Errorf("invalid annotation %s value: %s. Allowed values [ %q | %q ]", deliveryOrderAnnotation, val, internals.Ordered, internals.Unordered)
 	}
 }
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
new file mode 100644
index 000000000..5d87484fb
--- /dev/null
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -0,0 +1,155 @@
+/*
+ * 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 v2
+
+import (
+	"context"
+	"strings"
+
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker"
+	triggerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger"
+	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+
+	apiseventing "knative.dev/eventing/pkg/apis/eventing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
+)
+
+const (
+	ControllerAgentName = "kafka-trigger-controller"
+
+	FinalizerName = "kafka.triggers.eventing.knative.dev"
+)
+
+func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl {
+
+	logger := logging.FromContext(ctx).Desugar()
+
+	brokerInformer := brokerinformer.Get(ctx)
+	triggerInformer := triggerinformer.Get(ctx)
+	triggerLister := triggerInformer.Lister()
+	consumerGroupInformer := consumergroupinformer.Get(ctx)
+
+	reconciler := &ReconcilerV2{
+		BrokerLister:        brokerInformer.Lister(),
+		EventingClient:      eventingclient.Get(ctx),
+		Env:                 configs,
+		ConsumerGroupLister: consumerGroupInformer.Lister(),
+		InternalsClient:     consumergroupclient.Get(ctx),
+	}
+
+	impl := triggerreconciler.NewImpl(ctx, reconciler, func(impl *controller.Impl) controller.Options {
+		return controller.Options{
+			FinalizerName:     FinalizerName,
+			AgentName:         ControllerAgentName,
+			SkipStatusUpdates: false,
+		}
+	})
+
+	triggerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: filterTriggers(reconciler.BrokerLister),
+		Handler:    controller.HandleAll(impl.Enqueue),
+	})
+
+	// Filter Brokers and enqueue associated Triggers
+	brokerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: kafka.BrokerClassFilter(),
+		Handler:    enqueueTriggers(logger, triggerLister, impl.Enqueue),
+	})
+
+	// ConsumerGroup changes and enqueue associated Trigger
+	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		Handler: controller.HandleAll(func(obj interface{}) {
+			if cg, ok := obj.(*internalscg.ConsumerGroup); ok {
+				for _, or := range cg.OwnerReferences {
+					if strings.ToLower(or.Kind) == "trigger" {
+						impl.EnqueueKey(types.NamespacedName{Namespace: cg.GetNamespace(), Name: or.Name}) //enqueue trigger with name from owner ref
+						return
+					}
+				}
+			}
+		})})
+	return impl
+}
+
+func filterTriggers(lister eventinglisters.BrokerLister) func(interface{}) bool {
+	return func(obj interface{}) bool {
+		trigger, ok := obj.(*eventing.Trigger)
+		if !ok {
+			return false
+		}
+
+		if hasKafkaBrokerTriggerFinalizer(trigger.Finalizers, FinalizerName) {
+			return true
+		}
+
+		broker, err := lister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
+		if err != nil {
+			return false
+		}
+
+		value, ok := broker.GetAnnotations()[apiseventing.BrokerClassKey]
+		return ok && value == kafka.BrokerClass
+	}
+}
+
+func hasKafkaBrokerTriggerFinalizer(finalizers []string, finalizerName string) bool {
+	for _, f := range finalizers {
+		if f == finalizerName {
+			return true
+		}
+	}
+	return false
+}
+
+func enqueueTriggers(
+	logger *zap.Logger,
+	lister eventinglisters.TriggerLister,
+	enqueue func(obj interface{})) cache.ResourceEventHandler {
+
+	return controller.HandleAll(func(obj interface{}) {
+
+		if broker, ok := obj.(*eventing.Broker); ok {
+
+			selector := labels.SelectorFromSet(map[string]string{apiseventing.BrokerLabelKey: broker.Name})
+			triggers, err := lister.Triggers(broker.Namespace).List(selector)
+			if err != nil {
+				logger.Warn("Failed to list triggers", zap.Any("broker", broker), zap.Error(err))
+				return
+			}
+
+			for _, trigger := range triggers {
+				enqueue(trigger)
+			}
+		}
+	})
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
new file mode 100644
index 000000000..c1ef22033
--- /dev/null
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -0,0 +1,190 @@
+/*
+ * 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 v2
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
+	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
+	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/reconciler"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned"
+	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+)
+
+const (
+	deliveryOrderAnnotation = "kafka.eventing.knative.dev/delivery.order"
+)
+
+type ReconcilerV2 struct {
+	BrokerLister        eventinglisters.BrokerLister
+	EventingClient      eventingclientset.Interface
+	Env                 *config.Env
+	ConsumerGroupLister internalslst.ConsumerGroupLister
+	InternalsClient     internalsclient.Interface
+}
+
+func (r *ReconcilerV2) ReconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, trigger)
+
+	broker, err := r.BrokerLister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
+	if err != nil && !apierrors.IsNotFound(err) {
+		trigger.Status.MarkBrokerFailed("Failed to get broker", "%v", err)
+		return fmt.Errorf("failed to get broker: %w", err)
+	}
+
+	if apierrors.IsNotFound(err) {
+		// Actually check if the broker doesn't exist.
+		// Note: do not introduce another `broker` variable with `:`
+		broker, err = r.EventingClient.EventingV1().Brokers(trigger.Namespace).Get(ctx, trigger.Spec.Broker, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return fmt.Errorf("failed to get broker: %w", err)
+		}
+	}
+
+	// Ignore Triggers that are associated with a Broker we don't own.
+	if isKnativeKafkaBroker, brokerClass := isKnativeKafkaBroker(broker); !isKnativeKafkaBroker {
+		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
+		return nil
+	}
+
+	trigger.Status.PropagateBrokerCondition(broker.Status.GetTopLevelCondition())
+
+	if !broker.IsReady() {
+		// Trigger will get re-queued once this broker is ready.
+		return nil
+	}
+
+	consgroup, err := r.reconcileConsumerGroup(ctx, trigger)
+	if err != nil {
+		trigger.Status.MarkDependencyFailed("failed to reconcile consumer group", err.Error())
+		return err
+	}
+	trigger.Status.MarkDependencySucceeded()
+
+	trigger.Status.SubscriberURI = consgroup.Status.SubscriberURI
+	trigger.Status.MarkSubscriberResolvedSucceeded()
+
+	trigger.Status.DeadLetterSinkURI = consgroup.Status.DeadLetterSinkURI
+	trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+
+	return nil
+}
+
+func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
+
+	var deliveryOrdering = internals.Unordered
+	var err error
+	deliveryOrderingAnnotationValue, ok := trigger.Annotations[deliveryOrderAnnotation]
+	if ok {
+		deliveryOrdering, err = deliveryOrderingFromString(deliveryOrderingAnnotationValue)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	//todo additionally handle trigger.Spec.Filter
+	newcg := &internalscg.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      string(trigger.UID),
+			Namespace: trigger.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(trigger),
+			},
+		},
+		Spec: internalscg.ConsumerGroupSpec{
+			Template: internalscg.ConsumerTemplateSpec{
+				Spec: internalscg.ConsumerSpec{
+					Topics:  []string{},                                                                               //todo get topics from broker resource
+					Configs: internalscg.ConsumerConfigs{Configs: map[string]string{"group.id": string(trigger.UID)}}, //todo get bootstrap.servers from broker resource
+					Delivery: &internalscg.DeliverySpec{
+						DeliverySpec: trigger.Spec.Delivery,
+						Ordering:     deliveryOrdering},
+					Filters: &internalscg.Filters{
+						Filter:  trigger.Spec.Filter,
+						Filters: trigger.Spec.Filters,
+					},
+					Subscriber: trigger.Spec.Subscriber,
+				},
+			},
+		},
+	}
+
+	//todo validate newcg
+
+	cg, err := r.ConsumerGroupLister.ConsumerGroups(trigger.GetNamespace()).Get(string(trigger.UID)) //Get by consumer group name
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, err
+	}
+
+	if apierrors.IsNotFound(err) {
+		cg, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(newcg.GetNamespace()).Create(ctx, newcg, metav1.CreateOptions{})
+		if err != nil && !apierrors.IsAlreadyExists(err) {
+			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", newcg.GetNamespace(), newcg.GetName(), err)
+		}
+		if apierrors.IsAlreadyExists(err) {
+			return newcg, nil
+		}
+		return cg, nil
+	}
+
+	if equality.Semantic.DeepDerivative(newcg.Spec, cg.Spec) {
+		return cg, nil
+	}
+
+	newCg := &internalscg.ConsumerGroup{
+		TypeMeta:   cg.TypeMeta,
+		ObjectMeta: cg.ObjectMeta,
+		Spec:       newcg.Spec,
+		Status:     cg.Status,
+	}
+	if _, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
+		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
+	}
+
+	return cg, nil
+}
+
+func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
+	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
+	return brokerClass == kafka.BrokerClass, brokerClass
+}
+
+func deliveryOrderingFromString(val string) (internals.DeliveryOrdering, error) {
+	switch strings.ToLower(val) {
+	case string(internals.Ordered):
+		return internals.Ordered, nil
+	case string(internals.Unordered):
+		return internals.Unordered, nil
+	default:
+		return internals.Unordered, fmt.Errorf("invalid annotation %s value: %s. Allowed values [ %q | %q ]", deliveryOrderAnnotation, val, internals.Ordered, internals.Unordered)
+	}
+}


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index 406566673..d42e2b32f 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -99,7 +99,7 @@ func TestConsumerGroup_Validate(t *testing.T) {
 								},
 							},
 							Delivery: &DeliverySpec{
-								DeliverySpec: eventingduck.DeliverySpec{},
+								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
 							Subscriber: duckv1.Destination{
 								URI: &apis.URL{
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index 43930bbc9..2e3b3508c 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -39,6 +39,8 @@ import (
 
 	fakeeventingkafkabrokerclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
+	fakeconsumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 )
 
 const (
@@ -60,6 +62,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 		ctx, sourcesKafkaClient := fakeeventingkafkakafkaclient.With(ctx, listers.GetEventingKafkaObjects()...)
 		ctx, eventingKafkaClient := fakeeventingkafkabrokerclient.With(ctx, listers.GetEventingKafkaBrokerObjects()...)
 		ctx, kubeClient := fakekubeclient.With(ctx, listers.GetKubeObjects()...)
+		ctx, consumerGroupClient := fakeconsumergroupclient.With(ctx, listers.GetConsumerGroupObjects()...)
 
 		ctx, dynamicClient := fakedynamicclient.With(ctx,
 			newScheme(),
@@ -95,6 +98,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 			eventingClient.PrependReactor("*", "*", reactor)
 			eventingKafkaClient.PrependReactor("*", "*", reactor)
 			sourcesKafkaClient.PrependReactor("*", "*", reactor)
+			consumerGroupClient.PrependReactor("*", "*", reactor)
 		}
 
 		actionRecorderList := ActionRecorderList{
@@ -103,6 +107,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 			eventingClient,
 			eventingKafkaClient,
 			sourcesKafkaClient,
+			consumerGroupClient,
 		}
 
 		eventList := EventList{
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 3d3a0ee67..dee137ef4 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -28,6 +28,7 @@ import (
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	"knative.dev/pkg/reconciler/testing"
 
+	eventingkafkabrokerconsumer "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	eventingkafkachannels "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	eventingkafkasources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	fakeeventingkafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
@@ -36,6 +37,8 @@ import (
 
 	eventingkafkabroker "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	fakeeventingkafkabrokerclientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/fake"
+	fakeeventingkafkaconsumerclientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/fake"
+	consumerlisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	eventingkafkabrokerlisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/eventing/v1alpha1"
 )
 
@@ -45,6 +48,7 @@ var clientSetSchemes = []func(*runtime.Scheme) error{
 	fakeapiextensionsclientset.AddToScheme,
 	fakeeventingkafkabrokerclientset.AddToScheme,
 	fakeeventingkafkaclientset.AddToScheme,
+	fakeeventingkafkaconsumerclientset.AddToScheme,
 }
 
 type Listers struct {
@@ -95,6 +99,14 @@ func (l *Listers) GetEventingKafkaObjects() []runtime.Object {
 	return l.sorter.ObjectsForSchemeFunc(fakeeventingkafkaclientset.AddToScheme)
 }
 
+func (l *Listers) GetConsumerGroupObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakeeventingkafkaconsumerclientset.AddToScheme)
+}
+
+func (l *Listers) GetConsumerObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakeeventingkafkaconsumerclientset.AddToScheme)
+}
+
 func (l *Listers) GetBrokerLister() eventinglisters.BrokerLister {
 	return eventinglisters.NewBrokerLister(l.indexerFor(&eventing.Broker{}))
 }
@@ -127,6 +139,14 @@ func (l *Listers) GetKafkaChannelLister() eventingkafkachannelslisters.KafkaChan
 	return eventingkafkachannelslisters.NewKafkaChannelLister(l.indexerFor(&eventingkafkachannels.KafkaChannel{}))
 }
 
+func (l *Listers) GetConsumerGroupLister() consumerlisters.ConsumerGroupLister {
+	return consumerlisters.NewConsumerGroupLister(l.indexerFor(&eventingkafkabrokerconsumer.ConsumerGroup{}))
+}
+
+func (l *Listers) GetConsumerLister() consumerlisters.ConsumerLister {
+	return consumerlisters.NewConsumerLister(l.indexerFor(&eventingkafkabrokerconsumer.Consumer{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 54960b4c1..9ed7e3793 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/utils/pointer"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
@@ -37,6 +38,8 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
 const (
@@ -297,3 +300,21 @@ func allocateStatusAnnotations(obj duckv1.KRShaped) {
 		obj.GetStatus().Annotations = make(map[string]string, 1)
 	}
 }
+
+func NewConsumerGroup() *internalscg.ConsumerGroup {
+	return &internalscg.ConsumerGroup{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: internalscg.ConsumerGroupGroupVersionKind.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      TriggerUUID,
+			Namespace: ServiceNamespace,
+		},
+		Spec: internalscg.ConsumerGroupSpec{
+			Template: internalscg.ConsumerTemplateSpec{
+				Spec: internalscg.ConsumerSpec{},
+			},
+			Replicas: pointer.Int32Ptr(1),
+		},
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 45ad6dbdf..437bdb85a 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -24,6 +24,7 @@ import (
 
 	"google.golang.org/protobuf/testing/protocmp"
 	"k8s.io/utils/pointer"
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -38,7 +39,6 @@ import (
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	eventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
-	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/apis"
@@ -305,7 +305,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					BrokerReady,
 				),
-				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderOrdered)),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered))),
 				NewService(),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
@@ -357,7 +357,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 						reconcilertesting.WithTriggerDependencyReady(),
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(contract.DeliveryOrder_ORDERED),
-						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderOrdered),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered)),
 						reconcilertesting.WithTriggerDeadLetterSinkNotConfigured(),
 					),
 				},
@@ -369,7 +369,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					BrokerReady,
 				),
-				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderUnordered)),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered))),
 				NewService(),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
@@ -421,7 +421,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 						reconcilertesting.WithTriggerDependencyReady(),
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(contract.DeliveryOrder_UNORDERED),
-						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderUnordered),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered)),
 						reconcilertesting.WithTriggerDeadLetterSinkNotConfigured(),
 					),
 				},
@@ -2301,7 +2301,7 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 		return triggerreconciler.NewReconciler(
 			ctx,
 			logger,
-			fakeeventingclient.Get(ctx),
+			eventingclient.Get(ctx),
 			listers.GetTriggerLister(),
 			controller.GetEventRecorder(ctx),
 			reconciler,
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
new file mode 100644
index 000000000..af1533027
--- /dev/null
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
@@ -0,0 +1,41 @@
+/*
+ * 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 v2
+
+import (
+	"testing"
+
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+)
+
+func TestNewControllerV2(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	controller := NewControllerV2(ctx, &config.Env{})
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
diff --git a/test/pkg/broker/broker.go b/test/pkg/broker/broker.go
index 9ac8e60e9..c0b6daa92 100644
--- a/test/pkg/broker/broker.go
+++ b/test/pkg/broker/broker.go
@@ -17,13 +17,19 @@
 package broker
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	eventingtestlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 func Creator(client *eventingtestlib.Client, version string) string {
@@ -33,9 +39,33 @@ func Creator(client *eventingtestlib.Client, version string) string {
 
 	switch version {
 	case "v1":
+		namespace := client.Namespace
+		cmName := "kafka-broker-upgrade-config"
+		// Create Broker's own ConfigMap to prevent using defaults.
+		cm := &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      cmName,
+				Namespace: namespace,
+			},
+			Data: map[string]string{
+				kafka.BootstrapServersConfigMapKey:              testingpkg.BootstrapServersPlaintext,
+				kafka.DefaultTopicNumPartitionConfigMapKey:      fmt.Sprintf("%d", testingpkg.NumPartitions),
+				kafka.DefaultTopicReplicationFactorConfigMapKey: fmt.Sprintf("%d", testingpkg.ReplicationFactor),
+			},
+		}
+		cm, err := client.Kube.CoreV1().ConfigMaps(namespace).Create(context.Background(), cm, metav1.CreateOptions{})
+		if err != nil && !apierrors.IsAlreadyExists(err) {
+			client.T.Fatalf("Failed to create ConfigMap %s/%s: %v", namespace, cm.GetName(), err)
+		}
 		client.CreateBrokerOrFail(
 			name,
 			resources.WithBrokerClassForBroker(kafka.BrokerClass),
+			resources.WithConfigForBroker(&duckv1.KReference{
+				Kind:       "ConfigMap",
+				Namespace:  namespace,
+				Name:       cmName,
+				APIVersion: "v1",
+			}),
 		)
 	default:
 		panic(fmt.Sprintf("Unsupported version of Broker: %q", version))
diff --git a/test/pkg/testing/run.go b/test/pkg/testing/run.go
index f938e3b81..9335e3d7b 100644
--- a/test/pkg/testing/run.go
+++ b/test/pkg/testing/run.go
@@ -31,6 +31,9 @@ const (
 	BootstrapServersSaslPlaintext = "my-cluster-kafka-bootstrap.kafka:9095"
 	BootstrapServersSslSaslScram  = "my-cluster-kafka-bootstrap.kafka:9094"
 
+	NumPartitions     = 10
+	ReplicationFactor = 3
+
 	KafkaClusterNamespace = "kafka"
 	TlsUserSecretName     = "my-tls-user"
 	SaslUserSecretName    = "my-sasl-user"
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 17117e6a2..b66adc2f8 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: "v20211209-94078c662"
+    kafka.eventing.knative.dev/release: "v20211212-58afd241e"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -88,6 +88,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable experimental features in the delivery
                 subscribers:
                   description: This is the list of subscriptions for this subscribable.
                   type: array
@@ -131,6 +132,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable experimental features in the delivery
                       generation:
                         description: Generation of the origin of the subscriber with uid:UID.
                         type: integer
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index f022b0632..9d9e3a338 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: "v20211209-94078c662"
+    kafka.eventing.knative.dev/release: "v20211212-58afd241e"
     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 3893bf629..caaede975 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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 data:
   channelTemplateSpec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -350,8 +350,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -385,10 +385,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -432,10 +432,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -479,7 +479,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -528,8 +528,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "96896b00"
@@ -590,10 +590,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -644,10 +644,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -721,10 +721,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "0492ceb0"
@@ -779,10 +779,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -792,9 +792,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211209-13d751c67"
+        eventing.knative.dev/release: "v20211212-5601111eb"
         app.kubernetes.io/component: eventing-autoscaler
-        app.kubernetes.io/version: "20211209-13d751c67"
+        app.kubernetes.io/version: "20211212-5601111eb"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -812,7 +812,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:6d2c4dad53d911aa078ed9fa058f0e02a70a455ea202c2c2e482a6a5a8e54f0d
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:cdb38b98ad5592148d2f9cdb576857b7575f3a5ab2f26da05fedb9521ae6252d
           resources:
             requests:
               cpu: 100m
@@ -830,7 +830,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:14a72075d4d8aa072d35996e9a2eeff0dfd6561d4237a97f36ba83a0ec0ed1f9
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:1e85cf9729eb08ce13db71df958ec217e1bdebf9b4f7c170e502ad455d2adc8c
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -879,9 +879,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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.
@@ -894,9 +894,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211209-13d751c67"
+        eventing.knative.dev/release: "v20211212-5601111eb"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211209-13d751c67"
+        app.kubernetes.io/version: "20211212-5601111eb"
         app.kubernetes.io/name: knative-eventing
     spec:
       affinity:
@@ -910,7 +910,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:c44931131f6f1b2bba64006f6c9db413b3e3dfb27e0cd55e8b1f80b6b472339e
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:5fae94e4e0ec67b9deb00f67efb9f621395f923d14fcbcbe9f5a7bfc7d3677bf
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -975,9 +975,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1001,9 +1001,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   minAvailable: 80%
@@ -1032,9 +1032,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1045,9 +1045,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211209-13d751c67"
+        eventing.knative.dev/release: "v20211212-5601111eb"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20211209-13d751c67"
+        app.kubernetes.io/version: "20211212-5601111eb"
         app.kubernetes.io/name: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1067,7 +1067,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:f41ba86df53ded3159105172a81dc19a42223f4c03939a555b0d2b2baaa1cc71
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:1b3653a9af921cf472bab3790fcf76e8c9dbab0871cd00fe0f13c4ab82b709c1
           resources:
             requests:
               # taken from serving.
@@ -1136,10 +1136,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1171,11 +1171,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1387,10 +1387,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1554,11 +1554,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1826,11 +1826,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1980,9 +1980,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2109,10 +2109,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2364,11 +2364,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2555,10 +2555,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2875,12 +2875,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3067,9 +3067,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3279,9 +3279,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3465,8 +3465,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3479,9 +3479,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3499,9 +3499,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3522,9 +3522,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3549,9 +3549,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3570,9 +3570,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3608,8 +3608,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3635,8 +3635,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3653,8 +3653,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3687,8 +3687,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3701,9 +3701,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3741,9 +3741,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3755,9 +3755,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3769,9 +3769,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3783,9 +3783,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3797,9 +3797,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3812,8 +3812,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3826,8 +3826,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3854,8 +3854,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3990,8 +3990,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4058,8 +4058,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4072,9 +4072,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4120,8 +4120,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4134,9 +4134,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4172,8 +4172,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4259,8 +4259,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4359,8 +4359,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4396,8 +4396,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4434,8 +4434,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4468,8 +4468,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4503,8 +4503,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 # The data is populated at install time.
 
@@ -4528,8 +4528,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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 58d958b85..041c3372b 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: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     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: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -672,11 +672,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -826,9 +826,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -955,10 +955,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1210,11 +1210,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1401,10 +1401,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1721,12 +1721,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1913,9 +1913,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2125,9 +2125,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211209-13d751c67"
+    eventing.knative.dev/release: "v20211212-5601111eb"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211209-13d751c67"
+    app.kubernetes.io/version: "20211212-5601111eb"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev