Repository: knative-sandbox/eventing-kafka-broker
Pull request #1756 (2022-01-18 20:45:48)

Trigger reconciler tests for ConsumerGroup scaling
Signed-off-by: aavarghese <avarghese@us.ibm.com>

Part of #1537

/cc @pierDipi 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Adding new Trigger reconciler unit tests for consumer scheduling and scaling

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


knative-prow-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/1756#" title="Author self-approved">aavarghese</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **lionelvillard** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @lionelvillard` 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":["lionelvillard"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1756](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (668d634) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/51139ead182d9018860681c25cf54e2dad9e0f5a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (51139ea) will **decrease** coverage by `7.55%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/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/1756?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    #1756      +/-   ##
============================================
- Coverage     66.35%   58.80%   -7.56%     
============================================
  Files           129       60      -69     
  Lines          5282     3260    -2022     
  Branches        177        0     -177     
============================================
- Hits           3505     1917    -1588     
+ Misses         1492     1146     -346     
+ Partials        285      197      -88     
```

| 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/1756?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ol-plane/pkg/reconciler/trigger/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | `58.49% <100.00%> (ø)` | |
| [...ntrol-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL3RyaWdnZXJ2Mi5nbw==) | `86.74% <100.00%> (+86.74%)` | :arrow_up: |
| [...oker/dispatcher/main/ConsumerDeployerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lckRlcGxveWVyVmVydGljbGUuamF2YQ==) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...er/impl/consumer/InvalidCloudEventInterceptor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudEludGVyY2VwdG9yLmphdmE=) | | |
| [...venting/kafka/broker/receiver/IngressProducer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9JbmdyZXNzUHJvZHVjZXIuamF2YQ==) | | |
| [...enting/kafka/broker/core/utils/Configurations.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29uZmlndXJhdGlvbnMuamF2YQ==) | | |
| [...fka/broker/dispatcher/impl/filter/ExactFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvRXhhY3RGaWx0ZXIuamF2YQ==) | | |
| ... and [62 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756/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/1756?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/1756?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [51139ea...668d634](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1756?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).

knative-prow-robot: @aavarghese: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 668d634fb79eb21738d3832663f68640b614737a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1756/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1483541159002247168) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1756). [Your PR dashboard](https://gubernator.knative.dev/pr/aavarghese).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
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/1756#" title="Author self-approved">aavarghese</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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":["pierdipi"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1756/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1483585030549671936/artifacts/line-cov.html#file1) | 0.0% | 88.3% | 88.3


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 2de8551e4..675ef46bf 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -99,7 +99,9 @@ aliases:
   - matzew
   eventing-rabbitmq-approvers:
   - benmoss
+  - gabo1208
   - gvmw
+  - ikvmw
   - salaboy
   - xtreme-sameer-vohra
   eventing-redis-approvers:
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 9fbba76e3..0f769d11b 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
@@ -141,7 +141,7 @@ func (c *ConsumerGroup) GetUntypedSpec() interface{} {
 	return c.Spec
 }
 
-// GetStatus retrieves the status of the ConsumerGroupt. Implements the KRShaped interface.
+// GetStatus retrieves the status of the ConsumerGroup. Implements the KRShaped interface.
 func (c *ConsumerGroup) GetStatus() *duckv1.Status {
 	return &c.Status.Status
 }
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 7fcee74bb..08ea153b0 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -283,12 +283,24 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return err
 	}
 
-	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
+	broker.Status.Address.URL = nil
+
 	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
 	//  because topic metadata aren't available anymore.
 	// 	See (under discussions KIPs, unlikely to be accepted as they are):
 	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
 	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+	address := receiver.Address(r.IngressHost, broker)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: broker.GetNamespace(),
+			Name:      broker.GetName(),
+		},
+	}
+	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+		return nil // Object will get re-queued once probe status changes.
+	}
 
 	topicConfig, brokerConfig, err := r.topicConfig(ctx, logger, broker)
 	if err != nil {
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 87b097039..620b29361 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -322,6 +322,8 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 		logger.Debug("Contract config map updated")
 	}
 
+	channel.Status.Address = nil
+
 	// We update receiver and dispatcher pods annotation regardless of our contract changed or not due to the fact
 	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
 	// to update anyway remaining annotations with the contract generation that was saved in the CM.
@@ -336,12 +338,22 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 		return err
 	}
 
-	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
 	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
 	//  because topic metadata aren't available anymore.
 	// 	See (under discussions KIPs, unlikely to be accepted as they are):
 	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
 	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+	address := receiver.Address(r.IngressHost, channel)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: channel.GetNamespace(),
+			Name:      channel.GetName(),
+		},
+	}
+	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+		return nil // Object will get re-queued once probe status changes.
+	}
 
 	// get the channel configmap
 	channelConfigMap, err := r.channelConfigMap()
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index c4f870ae7..4725d56bc 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -277,6 +277,8 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		return err
 	}
 
+	ks.Status.Address.URL = nil
+
 	// We update receiver pods annotation regardless of our contract changed or not due to the fact  that in a previous
 	// reconciliation we might have failed to update one of our data plane pod annotation, so we want to anyway update
 	// remaining annotations with the contract generation that was saved in the CM.
@@ -287,12 +289,22 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		return err
 	}
 
-	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
 	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
 	//  because topic metadata aren't available anymore.
 	// 	See (under discussions KIPs, unlikely to be accepted as they are):
 	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
 	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+	address := receiver.Address(r.IngressHost, ks)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: ks.GetNamespace(),
+			Name:      ks.GetName(),
+		},
+	}
+	if status := r.Prober.Probe(ctx, proberAddressable); status != prober.StatusNotReady {
+		return nil // Object will get re-queued once probe status changes.
+	}
 
 	if ks.GetStatus().Annotations[base.TopicOwnerAnnotation] == ControllerTopicOwner {
 		secret, err := security.Secret(ctx, &SecretLocator{KafkaSink: ks}, r.SecretProviderFunc())
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 22229f0b9..563eb06da 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -48,7 +48,7 @@ const (
 	FinalizerName = "kafka.triggers.eventing.knative.dev"
 )
 
-func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl {
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	logger := logging.FromContext(ctx).Desugar()
 
@@ -57,7 +57,7 @@ func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl
 	triggerLister := triggerInformer.Lister()
 	consumerGroupInformer := consumergroupinformer.Get(ctx)
 
-	reconciler := &ReconcilerV2{
+	reconciler := &Reconciler{
 		BrokerLister:        brokerInformer.Lister(),
 		EventingClient:      eventingclient.Get(ctx),
 		Env:                 configs,
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index 8a79670b9..eade6ff7b 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -44,7 +44,7 @@ const (
 	deliveryOrderAnnotation = "kafka.eventing.knative.dev/delivery.order"
 )
 
-type ReconcilerV2 struct {
+type Reconciler struct {
 	BrokerLister        eventinglisters.BrokerLister
 	EventingClient      eventingclientset.Interface
 	Env                 *config.Env
@@ -52,7 +52,7 @@ type ReconcilerV2 struct {
 	InternalsClient     internalsclient.Interface
 }
 
-func (r *ReconcilerV2) ReconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
+func (r *Reconciler) 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)
@@ -83,23 +83,31 @@ func (r *ReconcilerV2) ReconcileKind(ctx context.Context, trigger *eventing.Trig
 		return nil
 	}
 
-	consgroup, err := r.reconcileConsumerGroup(ctx, trigger)
+	cg, 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
+	if cg.IsReady() {
+		trigger.Status.MarkDependencySucceeded()
+	} else {
+		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
+		if topLevelCondition == nil {
+			trigger.Status.MarkDependencyUnknown("failed to reconcile consumer group", "consumer group not ready")
+		} else {
+			trigger.Status.MarkDependencyFailed(topLevelCondition.Reason, topLevelCondition.Message)
+		}
+	}
+	trigger.Status.SubscriberURI = cg.Status.SubscriberURI
 	trigger.Status.MarkSubscriberResolvedSucceeded()
 
-	trigger.Status.DeadLetterSinkURI = consgroup.Status.DeadLetterSinkURI
+	trigger.Status.DeadLetterSinkURI = cg.Status.DeadLetterSinkURI
 	trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
 
 	return nil
 }
 
-func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
+func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
 
 	var deliveryOrdering = internals.Unordered
 	var err error
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index a7ee077b2..9d34d66bb 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -20,8 +20,8 @@ Lists of 158 third-party dependencies.
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1.1-android - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.2 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.2 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.3 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.3 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.12 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
@@ -43,48 +43,48 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.3.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
      (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.8.0.Final - https://debezium.io/debezium-parent/debezium-api)
      (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.8.0.Final - https://debezium.io/debezium-parent/debezium-core)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.10.1 - http://fabric8.io/kubernetes-client/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Apps (io.fabric8:kubernetes-model-apps:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apps/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Autoscaling (io.fabric8:kubernetes-model-autoscaling:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-autoscaling/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Batch (io.fabric8:kubernetes-model-batch:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-batch/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Certificates (io.fabric8:kubernetes-model-certificates:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-certificates/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Common (io.fabric8:kubernetes-model-common:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-common/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Coordination (io.fabric8:kubernetes-model-coordination:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-coordination/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Core (io.fabric8:kubernetes-model-core:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-core/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Discovery (io.fabric8:kubernetes-model-discovery:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-discovery/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Events (io.fabric8:kubernetes-model-events:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-events/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Extensions (io.fabric8:kubernetes-model-extensions:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-extensions/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: FlowControl (io.fabric8:kubernetes-model-flowcontrol:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-flowcontrol/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Metrics (io.fabric8:kubernetes-model-metrics:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-metrics/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Networking (io.fabric8:kubernetes-model-networking:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-networking/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Node (io.fabric8:kubernetes-model-node:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-node/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Policy (io.fabric8:kubernetes-model-policy:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-policy/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: RBAC (io.fabric8:kubernetes-model-rbac:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-rbac/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Scheduling (io.fabric8:kubernetes-model-scheduling:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-scheduling/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Storage Class (io.fabric8:kubernetes-model-storageclass:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-storageclass/)
-     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.10.1 - http://fabric8.io/kubernetes-server-mock/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.11.1 - http://fabric8.io/kubernetes-client/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Apps (io.fabric8:kubernetes-model-apps:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apps/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Autoscaling (io.fabric8:kubernetes-model-autoscaling:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-autoscaling/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Batch (io.fabric8:kubernetes-model-batch:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-batch/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Certificates (io.fabric8:kubernetes-model-certificates:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-certificates/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Common (io.fabric8:kubernetes-model-common:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-common/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Coordination (io.fabric8:kubernetes-model-coordination:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-coordination/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Core (io.fabric8:kubernetes-model-core:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-core/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Discovery (io.fabric8:kubernetes-model-discovery:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-discovery/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Events (io.fabric8:kubernetes-model-events:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-events/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Extensions (io.fabric8:kubernetes-model-extensions:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-extensions/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: FlowControl (io.fabric8:kubernetes-model-flowcontrol:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-flowcontrol/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Metrics (io.fabric8:kubernetes-model-metrics:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-metrics/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Networking (io.fabric8:kubernetes-model-networking:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-networking/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Node (io.fabric8:kubernetes-model-node:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-node/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Policy (io.fabric8:kubernetes-model-policy:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-policy/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: RBAC (io.fabric8:kubernetes-model-rbac:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-rbac/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Scheduling (io.fabric8:kubernetes-model-scheduling:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-scheduling/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Storage Class (io.fabric8:kubernetes-model-storageclass:5.11.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-storageclass/)
+     (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.11.1 - http://fabric8.io/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.2.2 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
      (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.2 - https://github.com/micrometer-metrics/micrometer)
      (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.8.2 - https://github.com/micrometer-metrics/micrometer)
-     (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.72.Final - https://netty.io/netty-buffer/)
-     (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.72.Final - https://netty.io/netty-codec/)
-     (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.72.Final - https://netty.io/netty-codec-dns/)
-     (Apache License, Version 2.0) Netty/Codec/HTTP (io.netty:netty-codec-http:4.1.72.Final - https://netty.io/netty-codec-http/)
-     (Apache License, Version 2.0) Netty/Codec/HTTP2 (io.netty:netty-codec-http2:4.1.72.Final - https://netty.io/netty-codec-http2/)
-     (Apache License, Version 2.0) Netty/Codec/Socks (io.netty:netty-codec-socks:4.1.72.Final - https://netty.io/netty-codec-socks/)
-     (Apache License, Version 2.0) Netty/Common (io.netty:netty-common:4.1.72.Final - https://netty.io/netty-common/)
-     (Apache License, Version 2.0) Netty/Handler (io.netty:netty-handler:4.1.72.Final - https://netty.io/netty-handler/)
-     (Apache License, Version 2.0) Netty/Handler/Proxy (io.netty:netty-handler-proxy:4.1.72.Final - https://netty.io/netty-handler-proxy/)
-     (Apache License, Version 2.0) Netty/Resolver (io.netty:netty-resolver:4.1.72.Final - https://netty.io/netty-resolver/)
-     (Apache License, Version 2.0) Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.72.Final - https://netty.io/netty-resolver-dns/)
+     (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.73.Final - https://netty.io/netty-buffer/)
+     (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.73.Final - https://netty.io/netty-codec/)
+     (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.73.Final - https://netty.io/netty-codec-dns/)
+     (Apache License, Version 2.0) Netty/Codec/HTTP (io.netty:netty-codec-http:4.1.73.Final - https://netty.io/netty-codec-http/)
+     (Apache License, Version 2.0) Netty/Codec/HTTP2 (io.netty:netty-codec-http2:4.1.73.Final - https://netty.io/netty-codec-http2/)
+     (Apache License, Version 2.0) Netty/Codec/Socks (io.netty:netty-codec-socks:4.1.73.Final - https://netty.io/netty-codec-socks/)
+     (Apache License, Version 2.0) Netty/Common (io.netty:netty-common:4.1.73.Final - https://netty.io/netty-common/)
+     (Apache License, Version 2.0) Netty/Handler (io.netty:netty-handler:4.1.73.Final - https://netty.io/netty-handler/)
+     (Apache License, Version 2.0) Netty/Handler/Proxy (io.netty:netty-handler-proxy:4.1.73.Final - https://netty.io/netty-handler-proxy/)
+     (Apache License, Version 2.0) Netty/Resolver (io.netty:netty-resolver:4.1.73.Final - https://netty.io/netty-resolver/)
+     (Apache License, Version 2.0) Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.73.Final - https://netty.io/netty-resolver-dns/)
      (Unknown license) Netty/TomcatNative [OpenSSL - Classes] (io.netty:netty-tcnative-classes:2.0.46.Final - https://github.com/netty/netty-tcnative/netty-tcnative-classes/)
-     (Apache License, Version 2.0) Netty/Transport (io.netty:netty-transport:4.1.72.Final - https://netty.io/netty-transport/)
-     (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.72.Final - https://netty.io/netty-transport-classes-epoll/)
-     (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.72.Final - https://netty.io/netty-transport-native-epoll/)
-     (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.72.Final - https://netty.io/netty-transport-native-unix-common/)
+     (Apache License, Version 2.0) Netty/Transport (io.netty:netty-transport:4.1.73.Final - https://netty.io/netty-transport/)
+     (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.73.Final - https://netty.io/netty-transport-classes-epoll/)
+     (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.73.Final - https://netty.io/netty-transport-native-epoll/)
+     (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.73.Final - https://netty.io/netty-transport-native-unix-common/)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
@@ -102,15 +102,15 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - Common (io.prometheus:simpleclient_tracer_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_common)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - OpenTelemetry (io.prometheus:simpleclient_tracer_otel:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_otel)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - OpenTelemetry Agent (io.prometheus:simpleclient_tracer_otel_agent:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_otel_agent)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-auth-common (io.vertx:vertx-auth-common:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-circuit-breaker (io.vertx:vertx-circuit-breaker:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-circuit-breaker)
-     (Eclipse Public License - v 2.0) (The Apache Software License, Version 2.0) Vert.x Core (io.vertx:vertx-core:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-core)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x JUnit 5 support :: Core (io.vertx:vertx-junit5:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-junit5)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x Kafka Client (io.vertx:vertx-kafka-client:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-kafka-client)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x metrics implementation for Micrometer.io (io.vertx:vertx-micrometer-metrics:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-micrometer-metrics)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x OpenTelemetry (io.vertx:vertx-opentelemetry:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-tracing-parent/vertx-opentelemetry)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-web-client (io.vertx:vertx-web-client:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web-client)
-     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-web-common (io.vertx:vertx-web-common:4.2.3 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web-common)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-auth-common (io.vertx:vertx-auth-common:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-circuit-breaker (io.vertx:vertx-circuit-breaker:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-circuit-breaker)
+     (Eclipse Public License - v 2.0) (The Apache Software License, Version 2.0) Vert.x Core (io.vertx:vertx-core:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-core)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x JUnit 5 support :: Core (io.vertx:vertx-junit5:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-junit5)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x Kafka Client (io.vertx:vertx-kafka-client:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-kafka-client)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x metrics implementation for Micrometer.io (io.vertx:vertx-micrometer-metrics:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-micrometer-metrics)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) Vert.x OpenTelemetry (io.vertx:vertx-opentelemetry:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-tracing-parent/vertx-opentelemetry)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-web-client (io.vertx:vertx-web-client:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web-client)
+     (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-web-common (io.vertx:vertx-web-common:4.2.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-web-parent/vertx-web-common)
      (The Apache Software License, Version 2.0) Zipkin Reporter: Core (io.zipkin.reporter2:zipkin-reporter:2.16.3 - https://github.com/openzipkin/zipkin-reporter-java/zipkin-reporter)
      (The Apache Software License, Version 2.0) Zipkin Sender: OkHttp 3 (io.zipkin.reporter2:zipkin-sender-okhttp3:2.16.3 - https://github.com/openzipkin/zipkin-reporter-java/zipkin-sender-okhttp3)
      (The Apache Software License, Version 2.0) Zipkin Core Library (io.zipkin.zipkin2:zipkin:2.23.2 - https://github.com/openzipkin/zipkin/zipkin)
@@ -136,12 +136,12 @@ Lists of 158 third-party dependencies.
      (BSD Licence 3) Hamcrest (org.hamcrest:hamcrest:2.1 - http://hamcrest.org/JavaHamcrest/)
      (New BSD License) Hamcrest Core (org.hamcrest:hamcrest-core:1.3 - https://github.com/hamcrest/JavaHamcrest/hamcrest-core)
      (BSD-2-Clause) (Public Domain, per Creative Commons CC0) HdrHistogram (org.hdrhistogram:HdrHistogram:2.1.12 - http://hdrhistogram.github.io/HdrHistogram/)
-     (Eclipse Public License v2.0) JUnit Jupiter (Aggregator) (org.junit.jupiter:junit-jupiter:5.8.1 - https://junit.org/junit5/)
-     (Eclipse Public License v2.0) JUnit Jupiter API (org.junit.jupiter:junit-jupiter-api:5.8.1 - https://junit.org/junit5/)
-     (Eclipse Public License v2.0) JUnit Jupiter Engine (org.junit.jupiter:junit-jupiter-engine:5.8.1 - https://junit.org/junit5/)
-     (Eclipse Public License v2.0) JUnit Jupiter Params (org.junit.jupiter:junit-jupiter-params:5.8.1 - https://junit.org/junit5/)
-     (Eclipse Public License v2.0) JUnit Platform Commons (org.junit.platform:junit-platform-commons:1.8.1 - https://junit.org/junit5/)
-     (Eclipse Public License v2.0) JUnit Platform Engine API (org.junit.platform:junit-platform-engine:1.8.1 - https://junit.org/junit5/)
+     (Eclipse Public License v2.0) JUnit Jupiter (Aggregator) (org.junit.jupiter:junit-jupiter:5.8.2 - https://junit.org/junit5/)
+     (Eclipse Public License v2.0) JUnit Jupiter API (org.junit.jupiter:junit-jupiter-api:5.8.2 - https://junit.org/junit5/)
+     (Eclipse Public License v2.0) JUnit Jupiter Engine (org.junit.jupiter:junit-jupiter-engine:5.8.2 - https://junit.org/junit5/)
+     (Eclipse Public License v2.0) JUnit Jupiter Params (org.junit.jupiter:junit-jupiter-params:5.8.2 - https://junit.org/junit5/)
+     (Eclipse Public License v2.0) JUnit Platform Commons (org.junit.platform:junit-platform-commons:1.8.2 - https://junit.org/junit5/)
+     (Eclipse Public License v2.0) JUnit Platform Engine API (org.junit.platform:junit-platform-engine:1.8.2 - https://junit.org/junit5/)
      (Public Domain, per Creative Commons CC0) LatencyUtils (org.latencyutils:LatencyUtils:2.0.3 - http://latencyutils.github.io/LatencyUtils/)
      (The Apache Software License, Version 2.0) LZ4 and xxHash (org.lz4:lz4-java:1.7.1 - https://github.com/lz4/lz4-java)
      (The MIT License) mockito-core (org.mockito:mockito-core:4.2.0 - https://github.com/mockito/mockito)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index cc4affb15..1810db1c0 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -41,20 +41,20 @@
     <mycila.license.plugin.version>4.1</mycila.license.plugin.version>
 
     <!-- dependencies version -->
-    <vertx.version>4.2.3</vertx.version>
+    <vertx.version>4.2.4</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.2</micrometer.version>
     <opentelemetry.version>1.10.0</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
-    <protobuf.version>3.19.2</protobuf.version>
+    <protobuf.version>3.19.3</protobuf.version>
     <slf4j.version>1.7.33</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <assertj.version>3.22.0</assertj.version>
     <awaitility.version>4.1.1</awaitility.version>
-    <junit.jupiter.version>5.8.1</junit.jupiter.version>
+    <junit.jupiter.version>5.8.2</junit.jupiter.version>
     <mokito.junit.jupiter.version>4.2.0</mokito.junit.jupiter.version>
-    <fabric8.kubernetes.version>5.10.1</fabric8.kubernetes.version>
+    <fabric8.kubernetes.version>5.11.1</fabric8.kubernetes.version>
     <kafka.version>2.8.1</kafka.version>
     <debezium.version>1.8.0.Final</debezium.version>
   </properties>
diff --git a/hack/release.sh b/hack/release.sh
index d9e302166..a7a7c295f 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -21,7 +21,7 @@ set -o pipefail
 source $(dirname $0)/../vendor/knative.dev/hack/release.sh
 source $(dirname $0)/data-plane.sh
 source $(dirname $0)/control-plane.sh
-source $(dirname $0)/hack/artifacts-env.sh
+source $(dirname $0)/artifacts-env.sh
 
 function fail() {
   echo "$1"


Test patch:
diff --git a/.github/workflows/kind-e2e.yaml b/.github/workflows/kind-e2e.yaml
index a61f40041..ba6bfc419 100644
--- a/.github/workflows/kind-e2e.yaml
+++ b/.github/workflows/kind-e2e.yaml
@@ -18,7 +18,6 @@ jobs:
       fail-fast: false # Keep running if one leg fails.
       matrix:
         k8s-version:
-          - v1.20.7
           - v1.21.1
           - v1.22.2
 
@@ -33,9 +32,6 @@ jobs:
         # This is attempting to make it a bit clearer what's being tested.
         # See: https://github.com/kubernetes-sigs/kind/releases/tag/v0.9.0
         include:
-          - k8s-version: v1.20.7
-            kind-version: v0.11.1
-            kind-image-sha: sha256:cbeaf907fc78ac97ce7b625e4bf0de16e3ea725daf6b04f930bd14c67c671ff9
           - k8s-version: v1.21.1
             kind-version: v0.11.1
             kind-image-sha: sha256:69860bda5563ac81e3c0057d654b5253219618a22ec3a346306239bba8cfa1a6
diff --git a/control-plane/pkg/receiver/address_test.go b/control-plane/pkg/receiver/address_test.go
new file mode 100644
index 000000000..3154bfc1f
--- /dev/null
+++ b/control-plane/pkg/receiver/address_test.go
@@ -0,0 +1,42 @@
+/*
+ * 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 receiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+)
+
+func TestAddress(t *testing.T) {
+	host := "example.com"
+	ks := &eventing.KafkaSink{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "ns",
+			Name:      "ks",
+		},
+	}
+	url := Address(host, ks)
+
+	require.Equal(t, host, url.Host)
+	require.Equal(t, "http", url.Scheme)
+	require.Contains(t, url.Path, ks.GetNamespace())
+	require.Contains(t, url.Path, ks.GetName())
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 757685137..93dad84b3 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1856,6 +1856,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}),
 				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
 		},
 		{
 			Name: "Reconciled normal - with DLS",
@@ -1885,6 +1888,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}),
 				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
 		},
 		{
 			Name: "Failed to delete topic",
@@ -1921,6 +1927,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			},
 			OtherTestData: map[string]interface{}{
 				wantErrorOnDeleteTopic: deleteTopicError,
+				testProber:             probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
 		{
@@ -1942,6 +1949,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
 		},
 		{
 			Name: "Reconciled normal - preserve config map previous state",
@@ -1980,6 +1990,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}),
 				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
 		},
 		{
 			Name: "Reconciled normal - topic doesn't exist",
@@ -2020,6 +2033,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			},
 			OtherTestData: map[string]interface{}{
 				wantErrorOnDeleteTopic: sarama.ErrUnknownTopicOrPartition,
+				testProber:             probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
 		{
@@ -2044,6 +2058,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
 		},
 	}
 
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 524d54a8f..4f0c278a3 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -1158,6 +1158,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 			OtherTestData: map[string]interface{}{
 				wantErrorOnDeleteTopic: deleteTopicError,
 				wantTopicName:          "topic-2",
+				testProber:             probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
 	}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index e6f9d1e9a..e7db0e0eb 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -21,8 +21,11 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
@@ -81,6 +84,24 @@ func NewConsumerSpec(opts ...ConsumerSpecOption) kafkainternals.ConsumerSpec {
 	return *spec
 }
 
+func NewConsumerSpecDelivery(order internals.DeliveryOrdering) *kafkainternals.DeliverySpec {
+	return &kafkainternals.DeliverySpec{
+		Ordering: order,
+	}
+}
+
+func NewConsumerSpecAuth() *kafkainternals.Auth {
+	return &kafkainternals.Auth{
+		NetSpec: &v1beta1.KafkaNetSpec{},
+	}
+}
+
+func NewConsumerSpecFilters() *kafkainternals.Filters {
+	return &kafkainternals.Filters{
+		Filter: &v1.TriggerFilter{},
+	}
+}
+
 func ConsumerSpec(spec kafkainternals.ConsumerSpec) ConsumerOption {
 	return func(cg *kafkainternals.Consumer) {
 		cg.Spec = spec
@@ -130,3 +151,33 @@ func ConsumerVReplicas(vreplicas int32) ConsumerSpecOption {
 		c.VReplicas = &vreplicas
 	}
 }
+
+func ConsumerAuth(auth *kafkainternals.Auth) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Auth = auth
+	}
+}
+
+func ConsumerDelivery(delivery *kafkainternals.DeliverySpec) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Delivery = delivery
+	}
+}
+
+func ConsumerSubscriber(dest duckv1.Destination) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Subscriber = dest
+	}
+}
+
+func ConsumerCloudEventOverrides(ce *duckv1.CloudEventOverrides) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.CloudEventOverrides = ce
+	}
+}
+
+func ConsumerFilters(filters *kafkainternals.Filters) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Filters = filters
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index a62463d2a..72f40c1be 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -17,8 +17,10 @@
 package testing
 
 import (
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
@@ -68,6 +70,45 @@ func NewConsumerGroup(opts ...ConsumerGroupOption) *kafkainternals.ConsumerGroup
 	return cg
 }
 
+func ConsumerGroupReady(cg *kafkainternals.ConsumerGroup) {
+	cg.Status.Conditions = duckv1.Conditions{
+		{
+			Type:   apis.ConditionReady,
+			Status: corev1.ConditionTrue,
+		},
+	}
+}
+
+func WithConsumerGroupFailed(reason string, msg string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.GetConditionSet().Manage(cg.GetStatus()).MarkFalse(kafkainternals.ConditionConsumerGroupConsumers, reason, msg)
+	}
+}
+
+func WithConsumerGroupName(name string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.ObjectMeta.Name = name
+	}
+}
+
+func WithConsumerGroupNamespace(namespace string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.ObjectMeta.Namespace = namespace
+	}
+}
+
+func WithConsumerGroupOwnerRef(ownerref *metav1.OwnerReference) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.ObjectMeta.OwnerReferences = []metav1.OwnerReference{*ownerref}
+	}
+}
+
+func WithConsumerGroupLabels(labels map[string]string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Spec.Template.Labels = labels
+	}
+}
+
 func ConsumerGroupReplicas(replicas int32) ConsumerGroupOption {
 	return func(cg *kafkainternals.ConsumerGroup) {
 		cg.Spec.Replicas = pointer.Int32Ptr(replicas)
@@ -85,3 +126,13 @@ func ConsumerGroupConsumerSpec(spec kafkainternals.ConsumerSpec) ConsumerGroupOp
 		cg.Spec.Template.Spec = spec
 	}
 }
+
+func WithDeadLetterSinkURI(uri string) func(cg *kafkainternals.ConsumerGroup) {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		u, err := apis.ParseURL(uri)
+		if err != nil {
+			panic(err)
+		}
+		cg.Status.DeadLetterSinkURI = u
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 17feb9077..04ce2639f 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -135,8 +135,13 @@ func InitSourceConditions(obj duckv1.KRShaped) {
 func StatusSourceSinkResolved(uri string) KRShapedOption {
 	return func(obj duckv1.KRShaped) {
 		ks := obj.(*sources.KafkaSource)
-		ks.Status.SinkURI, _ = apis.ParseURL(uri)
-		ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(sources.KafkaConditionSinkProvided)
+		res, _ := apis.ParseURL(uri)
+		ks.Status.SinkURI = res
+		if !res.IsEmpty() {
+			ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(sources.KafkaConditionSinkProvided)
+		} else {
+			ks.GetConditionSet().Manage(ks.GetStatus()).MarkUnknown(sources.KafkaConditionSinkProvided, "SinkEmpty", "Sink has resolved to empty.%s", "")
+		}
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
index af1533027..455ac194c 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
@@ -31,10 +31,10 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 )
 
-func TestNewControllerV2(t *testing.T) {
+func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	controller := NewControllerV2(ctx, &config.Env{})
+	controller := NewController(ctx, &config.Env{})
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
new file mode 100644
index 000000000..6e6fe7e24
--- /dev/null
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -0,0 +1,588 @@
+/*
+ * 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"
+	"testing"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgotesting "k8s.io/client-go/testing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingclient "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"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+
+	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+)
+
+const (
+	// name of the trigger under test
+	triggerName = "test-trigger"
+	// namespace of the trigger under test
+	triggerNamespace = "test-namespace"
+)
+
+var DefaultEnv = &config.Env{
+	DataPlaneConfigMapNamespace: "knative-eventing",
+	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+	GeneralConfigMapName:        "kafka-broker-config",
+	SystemNamespace:             "knative-eventing",
+}
+
+var (
+	url = &apis.URL{
+		Scheme: "http",
+		Host:   "localhost",
+		Path:   "/path",
+	}
+)
+
+func TestReconcileKind(t *testing.T) {
+
+	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with ordered delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered))),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered)),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with unordered delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered))),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered)),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with invalid delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, "invalid")),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(
+						"invalid annotation %s value: invalid. Allowed values [ \"ordered\" | \"unordered\" ]",
+						deliveryOrderAnnotation,
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerDependencyFailed("failed to reconcile consumer group", "invalid annotation kafka.eventing.knative.dev/delivery.order value: invalid. Allowed values [ \"ordered\" | \"unordered\" ]"),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, "invalid"),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(TriggerUUID),
+						WithConsumerGroupNamespace(triggerNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(TriggerUUID),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+							ConsumerFilters(NewConsumerSpecFilters()),
+						)),
+						ConsumerGroupReady,
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyReady(),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update but not ready",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(TriggerUUID),
+						WithConsumerGroupNamespace(triggerNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(TriggerUUID),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+							ConsumerFilters(NewConsumerSpecFilters()),
+						)),
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+					ConsumerGroupReady,
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyReady(),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update but not ready",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg but failed",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+					WithConsumerGroupFailed("failed", "failed"),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyFailed("failed", "failed"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal with dead letter sink uri",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+					WithDeadLetterSinkURI(url.String()),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(url.String()),
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker not found",
+			Objects: []runtime.Object{
+				newTrigger(),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(
+						"failed to get broker: brokers.eventing.knative.dev \"%s\" not found",
+						BrokerName,
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker not ready",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewBroker(
+					func(v *eventing.Broker) { v.Status.InitializeConditions() },
+					StatusBrokerConfigNotParsed("wrong"),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker deleted",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewDeletedBroker(),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerNotConfigured(),
+					),
+				},
+			},
+		},
+		{
+			Name: "Don't reconcile trigger associated with a broker with a different broker class",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewBroker(func(b *eventing.Broker) {
+					b.Annotations = map[string]string{
+						eventing.BrokerClassAnnotationKey: "MTChannelBasedBroker",
+					}
+				}),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+	}
+
+	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		logger := logging.FromContext(ctx)
+
+		reconciler := &Reconciler{
+			BrokerLister:        listers.GetBrokerLister(),
+			EventingClient:      eventingclient.Get(ctx),
+			Env:                 env,
+			ConsumerGroupLister: listers.GetConsumerGroupLister(),
+			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
+		}
+
+		return triggerreconciler.NewReconciler(
+			ctx,
+			logger,
+			eventingclient.Get(ctx),
+			listers.GetTriggerLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+	}))
+}
+
+func newTrigger(options ...reconcilertesting.TriggerOption) *eventing.Trigger {
+	return reconcilertesting.NewTrigger(
+		triggerName,
+		triggerNamespace,
+		BrokerName,
+		append(
+			options,
+			//reconcilertesting.WithTriggerSubscriberURI(ServiceURL),
+			func(t *eventing.Trigger) {
+				t.UID = TriggerUUID
+			},
+		)...,
+	)
+}
+
+func withDeadLetterSinkURI(uri string) func(trigger *eventing.Trigger) {
+	return func(trigger *eventing.Trigger) {
+		u, err := apis.ParseURL(uri)
+		if err != nil {
+			panic(err)
+		}
+		trigger.Status.DeadLetterSinkURI = u
+		trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+	}
+}
+
+func withTriggerSubscriberResolvedSucceeded() func(*eventing.Trigger) {
+	return func(t *eventing.Trigger) {
+		t.GetConditionSet().Manage(&t.Status).MarkTrue(
+			eventing.TriggerConditionSubscriberResolved,
+		)
+	}
+}
diff --git a/test/upgrade/continual/verification.go b/test/upgrade/continual/verification.go
index 49f63597e..02a206071 100644
--- a/test/upgrade/continual/verification.go
+++ b/test/upgrade/continual/verification.go
@@ -18,6 +18,7 @@ package continual
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/kelseyhightower/envconfig"
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade/continual"
@@ -77,6 +78,7 @@ func configurator(theSut sut.SystemUnderTest, configTemplate string) prober.Conf
 		// TODO: knative/eventing#5176 - this is cumbersome
 		config.ConfigTemplate = fmt.Sprintf("../../../../../../%s",
 			configTemplate)
+		config.FinishedSleep = 5 * time.Minute
 		// envconfig.Process invocation is repeated from within prober.NewConfig to
 		// make sure every knob is configurable, but using defaults from Eventing
 		// Kafka instead of Core. The prefix is also changed.
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 7808ca1b1..ece583caf 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -204,6 +204,9 @@ spec:
                     namespace:
                       description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                       type: string
+                deadLetterSinkUri:
+                  description: DeadLetterSinkURI is the resolved URI of the dead letter ref if one is specified in the Spec.Delivery.
+                  type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 6dde72e0e..fc912c3f7 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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 data:
   channel-template-spec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
   annotations:
     knative.dev/example-checksum: "9185c153"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 data:
   _example: |
@@ -432,10 +432,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
   annotations:
     knative.dev/example-checksum: "f7948630"
@@ -590,10 +590,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/high-availability: "true"
     app.kubernetes.io/component: eventing-controller
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -792,9 +792,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20220114-61d97d366"
+        eventing.knative.dev/release: "v20220118-f6fec1088"
         app.kubernetes.io/component: eventing-autoscaler
-        app.kubernetes.io/version: "20220114-61d97d366"
+        app.kubernetes.io/version: "20220118-f6fec1088"
         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:2f10ad1a6aed877f80567d61b5eb34481ef341a9314d02a9dfc8d603d2acb98b
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:9328449502724216bd52e31dc13d25818b3d3c93f537717eb161ff85926a7ef4
           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:b0033248d3e7a4854fdb0c7a7805921cf24551ac4978e69d0ea2d22fcd870f17
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:033a012326798e22f0fb6f707c5f15b6b1b601eb76b1b7d7b2712cca9b1da058
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -879,9 +879,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     app.kubernetes.io/component: pingsource-mt-adapter
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+        eventing.knative.dev/release: "v20220118-f6fec1088"
         app.kubernetes.io/component: pingsource-mt-adapter
-        app.kubernetes.io/version: "20220114-61d97d366"
+        app.kubernetes.io/version: "20220118-f6fec1088"
         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:f85f335714b489cd3edcdf69fc15823aac7967fd3475780dfbb33e801199cdb8
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:11f8c49f7531debd5348a945522fbe9634890804528a53bf2b3148d4831f166b
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -975,9 +975,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 spec:
   scaleTargetRef:
@@ -1001,9 +1001,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 spec:
   selector:
@@ -1045,9 +1045,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20220114-61d97d366"
+        eventing.knative.dev/release: "v20220118-f6fec1088"
         app.kubernetes.io/component: eventing-webhook
-        app.kubernetes.io/version: "20220114-61d97d366"
+        app.kubernetes.io/version: "20220118-f6fec1088"
         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:b29beced91a15a48eed7d20fbeaf36b0a0515cdd2fe38b0331993b42bd9156e3
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:0e8afc747d9adfa2c66596713f530dd0faa0eab760c4ac4191f68c51378baf43
           resources:
             requests:
               # taken from serving.
@@ -1136,10 +1136,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     role: eventing-webhook
     app.kubernetes.io/component: eventing-webhook
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3635,8 +3635,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3653,8 +3653,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -3687,8 +3687,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3701,9 +3701,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4058,8 +4058,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 rules:
   - apiGroups:
@@ -4259,8 +4259,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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 5292d93ba..26eebdebe 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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     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: "v20220114-61d97d366"
+    eventing.knative.dev/release: "v20220118-f6fec1088"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20220114-61d97d366"
+    app.kubernetes.io/version: "20220118-f6fec1088"
     app.kubernetes.io/name: knative-eventing
 spec:
   group: eventing.knative.dev