Repository: knative-sandbox/eventing-kafka-broker
Pull request #1757 (2022-01-18 20:58:51)

KafkaSource reconciler tests for ConsumerGroup scaling
Signed-off-by: aavarghese <avarghese@us.ibm.com>

Part of #1537
~Will be rebased on #1754~ **done**

/cc @pierDipi 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Adding new KafkaSource 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/1757#" 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 **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` 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":["aliok"]} -->
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/source/v2/sourcev2.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1757/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1483544442995478528/artifacts/line-cov.html#file1) | 0.0% | 78.6% | 78.6

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1757](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5f33b29) 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 `8.36%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/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/1757?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    #1757      +/-   ##
============================================
- Coverage     66.35%   57.99%   -8.37%     
============================================
  Files           129       60      -69     
  Lines          5282     3259    -2023     
  Branches        177        0     -177     
============================================
- Hits           3505     1890    -1615     
+ Misses         1492     1172     -320     
+ 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/1757?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...rol-plane/pkg/reconciler/source/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvdjIvY29udHJvbGxlcnYyLmdv) | `100.00% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/v2/sourcev2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvdjIvc291cmNldjIuZ28=) | `80.35% <100.00%> (+80.35%)` | :arrow_up: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `76.82% <0.00%> (ø)` | |
| [...fka/broker/receiver/impl/handler/ProbeHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvUHJvYmVIYW5kbGVyLmphdmE=) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...roker/dispatcher/impl/filter/BaseStringFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQmFzZVN0cmluZ0ZpbHRlci5qYXZh) | | |
| [...dispatcher/impl/ResponseToHttpEndpointHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZXNwb25zZVRvSHR0cEVuZHBvaW50SGFuZGxlci5qYXZh) | | |
| [...venting/kafka/broker/receiver/IngressProducer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9JbmdyZXNzUHJvZHVjZXIuamF2YQ==) | | |
| [...ting/kafka/broker/dispatcher/CloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvQ2xvdWRFdmVudFNlbmRlci5qYXZh) | | |
| [...er/dispatcher/impl/consumer/InvalidCloudEvent.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudC5qYXZh) | | |
| ... and [63 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757/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/1757?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/1757?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [51139ea...5f33b29](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1757?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: PR needs rebase.

<details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1757#" 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-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-unit-tests | 962095ec3c0332d2114e721cecea5d62f84215e3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1757/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1483552580373057536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1757). [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-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/source/v2/sourcev2.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1757/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1483576276009619456/artifacts/line-cov.html#file1) | 0.0% | 76.5% | 76.5

aavarghese: /cc @pierDipi 

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/source/v2/controllerv2.go b/control-plane/pkg/reconciler/source/v2/controllerv2.go
index dbd161ca1..b9b072de1 100644
--- a/control-plane/pkg/reconciler/source/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2.go
@@ -33,14 +33,14 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
 )
 
-func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl {
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	kafkaInformer := kafkainformer.Get(ctx)
 	consumerGroupInformer := consumergroupinformer.Get(ctx)
 
 	sources.RegisterAlternateKafkaConditionSet(conditionSet)
 
-	r := &ReconcilerV2{
+	r := &Reconciler{
 		ConsumerGroupLister: consumerGroupInformer.Lister(),
 		InternalsClient:     consumergroupclient.Get(ctx),
 	}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
index fc79cb028..e68eb8950 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -45,15 +45,16 @@ const (
 var (
 	conditionSet = apis.NewLivingConditionSet(
 		KafkaConditionConsumerGroup,
+		sources.KafkaConditionSinkProvided,
 	)
 )
 
-type ReconcilerV2 struct {
+type Reconciler struct {
 	ConsumerGroupLister internalslst.ConsumerGroupLister
 	InternalsClient     internalsclient.Interface
 }
 
-func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
 
 	cg, err := r.reconcileConsumerGroup(ctx, ks)
 	if err != nil {
@@ -64,11 +65,15 @@ func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSourc
 		ks.GetConditionSet().Manage(&ks.Status).MarkTrue(KafkaConditionConsumerGroup)
 	} else {
 		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
-		ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
-			KafkaConditionConsumerGroup,
-			topLevelCondition.Reason,
-			topLevelCondition.Message,
-		)
+		if topLevelCondition == nil {
+			ks.GetConditionSet().Manage(&ks.Status).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group not ready")
+		} else {
+			ks.GetConditionSet().Manage(&ks.Status).MarkFalse(
+				KafkaConditionConsumerGroup,
+				topLevelCondition.Reason,
+				topLevelCondition.Message,
+			)
+		}
 	}
 
 	ks.Status.MarkSink(cg.Status.SubscriberURI)
@@ -80,7 +85,7 @@ func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSourc
 	return nil
 }
 
-func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
+func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
 
 	newcg := &internalscg.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{
@@ -105,15 +110,18 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, ks *sources.Ka
 						Ordering: DefaultDeliveryOrder,
 					},
 					Subscriber: ks.Spec.Sink,
-					CloudEventOverrides: &duckv1.CloudEventOverrides{
-						Extensions: ks.Spec.CloudEventOverrides.Extensions,
-					},
 				},
 			},
 			Replicas: ks.Spec.Consumers,
 		},
 	}
 
+	if ks.Spec.CloudEventOverrides != nil {
+		newcg.Spec.Template.Spec.CloudEventOverrides = &duckv1.CloudEventOverrides{
+			Extensions: ks.Spec.CloudEventOverrides.Extensions,
+		}
+	}
+
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(ks.GetNamespace()).Get(string(ks.UID)) //Get by consumer group id
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, err
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..8ea9aebed 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,74 +43,74 @@ 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/)
-     (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)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (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.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.10.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.10.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.10.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.10.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.12.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_common)
      (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/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
index ce6cd36cc..e49cdec82 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
@@ -160,10 +160,30 @@ public final class OffsetManager implements RecordDispatcherListener {
   }
 
   /**
-   * This offset tracker keeps track of the committed records.
+   * This offset tracker keeps track of the committed records for a
+   * single partition.
    */
   private static class OffsetTracker {
 
+    /*
+     * We use a BitSet as a replacement for an array of booleans.
+     * Each bit represents an offset of a partition.
+     *
+     * A bit in the BitSet is set when an offset can be committed
+     * to Kafka.
+     *
+     * Example case:
+     *
+     *  partition offsets   [0, 1, 2, 3, 4, 5, 6, 7, ... ]  <-- BitSet
+     *  t1                                                  <-- time 1
+     *  t1 success           ^     ^  ^        ^            <-- offsets recorded
+     *  t1 to commit            &                           <-- offset to commit (next unset offset)
+     *
+     *  t2                                                  <-- time 2
+     *  t2 success           ^  ^  ^  ^        ^            <-- offsets recorded
+     *  t2 to commit                     &                  <-- offset to commit (next unset offset)
+     */
+
     // In order to not use a huge amount of memory we cap the BitSet to a _dynamic_ max size governed by this
     // threshold.
     private static final int RESET_TRACKER_THRESHOLD = 1_000_000;
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index 75c1d3eaa..ba9caf8e8 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -51,8 +51,11 @@ public final class WebClientCloudEventSender implements CloudEventSender {
    */
   public WebClientCloudEventSender(final WebClient client, final CircuitBreaker circuitBreaker, final String target) {
     Objects.requireNonNull(client, "provide client");
-    if (target == null || target.equals("") || !URI.create(target).isAbsolute()) {
-      throw new IllegalArgumentException("provide a valid target");
+    if (target == null || target.equals("")) {
+      throw new IllegalArgumentException("provide a target");
+    }
+    if (!URI.create(target).isAbsolute()){
+      throw new IllegalArgumentException("provide a valid target, provided target: " + target);
     }
     Objects.requireNonNull(circuitBreaker, "provide circuitBreaker");
 
@@ -79,7 +82,7 @@ public final class WebClientCloudEventSender implements CloudEventSender {
 
   private void send(final CloudEvent event, final Promise<HttpResponse<Buffer>> breaker) {
     VertxMessageFactory
-      .createWriter(client.postAbs(target))
+      .createWriter(client.postAbs(target).putHeader("Prefer", "reply"))
       .writeBinary(event)
       .onFailure(ex -> {
         logError(event, ex);
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index cc4affb15..9eb0b1664 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>
+    <opentelemetry.version>1.10.1</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"
diff --git a/hack/tools.go b/hack/tools.go
index 74dfd0bf2..4f73a57f2 100644
--- a/hack/tools.go
+++ b/hack/tools.go
@@ -27,6 +27,7 @@ import (
 	// Test images from eventing
 	_ "knative.dev/eventing/test/test_images/event-flaker"
 	_ "knative.dev/eventing/test/test_images/event-library"
+	_ "knative.dev/eventing/test/test_images/event-library/kodata/events"
 	_ "knative.dev/eventing/test/test_images/event-sender"
 	_ "knative.dev/eventing/test/test_images/heartbeats"
 	_ "knative.dev/eventing/test/test_images/performance"
diff --git a/vendor/modules.txt b/vendor/modules.txt
index b173e8ed4..8597fc4c5 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1219,18 +1219,27 @@ knative.dev/eventing/test/performance/infra/event_state
 knative.dev/eventing/test/performance/infra/receiver
 knative.dev/eventing/test/performance/infra/sender
 knative.dev/eventing/test/rekt/features/broker
+knative.dev/eventing/test/rekt/features/channel
 knative.dev/eventing/test/rekt/features/knconf
 knative.dev/eventing/test/rekt/features/trigger
+knative.dev/eventing/test/rekt/resources/account_role
 knative.dev/eventing/test/rekt/resources/addressable
 knative.dev/eventing/test/rekt/resources/broker
+knative.dev/eventing/test/rekt/resources/channel
+knative.dev/eventing/test/rekt/resources/channel_impl
+knative.dev/eventing/test/rekt/resources/containersource
 knative.dev/eventing/test/rekt/resources/delivery
 knative.dev/eventing/test/rekt/resources/eventlibrary
+knative.dev/eventing/test/rekt/resources/eventlibrary/events
 knative.dev/eventing/test/rekt/resources/flaker
+knative.dev/eventing/test/rekt/resources/pingsource
+knative.dev/eventing/test/rekt/resources/source
 knative.dev/eventing/test/rekt/resources/subscription
 knative.dev/eventing/test/rekt/resources/trigger
 knative.dev/eventing/test/test_images
 knative.dev/eventing/test/test_images/event-flaker
 knative.dev/eventing/test/test_images/event-library
+knative.dev/eventing/test/test_images/event-library/kodata/events
 knative.dev/eventing/test/test_images/event-sender
 knative.dev/eventing/test/test_images/heartbeats
 knative.dev/eventing/test/test_images/performance


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/source/v2/controllerv2_test.go b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
index e378c42b1..ddf18dc2c 100644
--- a/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
@@ -61,7 +61,7 @@ func TestNewControllerV2(t *testing.T) {
 
 	dynamicclient.With(ctx, dynamicScheme)
 
-	controller := NewControllerV2(ctx, configs)
+	controller := NewController(ctx, configs)
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
new file mode 100644
index 000000000..f920ae1f9
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -0,0 +1,402 @@
+/*
+ * 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"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/pkg/kmeta"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+
+	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	eventingkafkasourcereconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+)
+
+func TestReconcileKind(t *testing.T) {
+
+	testKey := fmt.Sprintf("%s/%s", SourceNamespace, SourceName)
+
+	sources.RegisterAlternateKafkaConditionSet(conditionSet)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewSource(),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - ce overrides",
+			Objects: []runtime.Object{
+				NewSourceSinkObject(),
+				NewSource(WithCloudEventOverrides(&duckv1.CloudEventOverrides{
+					Extensions: map[string]string{"a": "foo", "b": "foo"},
+				})),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerCloudEventOverrides(&duckv1.CloudEventOverrides{
+							Extensions: map[string]string{"a": "foo", "b": "foo"},
+						}),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithCloudEventOverrides(&duckv1.CloudEventOverrides{
+							Extensions: map[string]string{"a": "foo", "b": "foo"},
+						}),
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(SourceUUID),
+						WithConsumerGroupNamespace(SourceNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(SourceConsumerGroup),
+								ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							),
+							ConsumerAuth(NewConsumerSpecAuth()),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerSubscriber(NewSourceSinkReference()),
+						)),
+						ConsumerGroupReplicas(1),
+						ConsumerGroupReady,
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update but not ready",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(SourceUUID),
+						WithConsumerGroupNamespace(SourceNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(SourceConsumerGroup),
+								ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							),
+							ConsumerAuth(NewConsumerSpecAuth()),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+							ConsumerSubscriber(NewSourceSinkReference()),
+						)),
+						ConsumerGroupReplicas(1),
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update but not ready",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg but failed",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+					WithConsumerGroupFailed("failed", "failed"),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupFailed("failed", "failed"),
+						StatusSourceSinkResolved(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with replicas set in status",
+			Objects: []runtime.Object{
+				NewSource(),
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+					ConsumerGroupReplicas(1),
+					ConsumerGroupReplicasStatus(1),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroup(),
+						StatusSourceSinkResolved(""),
+						StatusSourceConsumerGroupReplicas(1),
+					),
+				},
+			},
+		},
+	}
+
+	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		reconciler := &Reconciler{
+			ConsumerGroupLister: listers.GetConsumerGroupLister(),
+			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
+		}
+
+		r := eventingkafkasourcereconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingkafkaclient.Get(ctx),
+			listers.GetKafkaSourceLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+		return r
+	}))
+}
+
+func StatusSourceConsumerGroup() KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(KafkaConditionConsumerGroup)
+	}
+}
+
+func StatusSourceConsumerGroupFailed(reason string, msg string) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkFalse(KafkaConditionConsumerGroup, reason, msg)
+	}
+}
+
+func StatusSourceConsumerGroupUnknown() KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.GetConditionSet().Manage(ks.GetStatus()).MarkUnknown(KafkaConditionConsumerGroup, "failed to reconcile consumer group", "consumer group not ready")
+	}
+}
+
+func StatusSourceConsumerGroupReplicas(replicas int32) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Status.Consumers = replicas
+	}
+}
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..a569f4f9a 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,12 +70,57 @@ 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)
 	}
 }
 
+func ConsumerGroupReplicasStatus(replicas int32) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Status.Replicas = pointer.Int32Ptr(replicas)
+	}
+}
+
 func ConsumerGroupSelector(selector map[string]string) ConsumerGroupOption {
 	return func(cg *kafkainternals.ConsumerGroup) {
 		cg.Spec.Selector = selector
@@ -85,3 +132,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/e2e-tests.sh b/test/e2e-tests.sh
index 2b9a88e17..d9a36da09 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -21,7 +21,7 @@ header "Running tests"
 
 export_logs_continuously
 
-go_test_e2e -timeout=30m ./test/e2e/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite failed"
+go_test_e2e -timeout=1h ./test/e2e/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E suite failed"
 
 go_test_e2e -tags=deletecm ./test/e2e/... -channels=messaging.knative.dev/v1beta1:KafkaChannel || fail_test "E2E (deletecm) suite failed"
 
diff --git a/test/e2e_new/kafka_channel_upstream_test.go b/test/e2e_new/kafka_channel_upstream_test.go
new file mode 100644
index 000000000..132670950
--- /dev/null
+++ b/test/e2e_new/kafka_channel_upstream_test.go
@@ -0,0 +1,330 @@
+//go:build e2e && cloudevents
+// +build e2e,cloudevents
+
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e_new
+
+import (
+	"testing"
+
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/manifest"
+
+	"github.com/cloudevents/sdk-go/v2/binding"
+	"knative.dev/eventing/test/rekt/features/channel"
+	ch "knative.dev/eventing/test/rekt/resources/channel"
+	chimpl "knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	"knative.dev/pkg/system"
+	_ "knative.dev/pkg/system/testing"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	// needed for vendoring the test resource files
+	_ "knative.dev/eventing/test/rekt/resources/eventlibrary/events"
+)
+
+// TestChannelConformance
+func TestChannelConformance(t *testing.T) {
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	channelName := "mychannelimpl"
+
+	// Install and wait for a Ready Channel.
+	env.Prerequisite(ctx, t, channel.ImplGoesReady(channelName))
+
+	env.TestSet(ctx, t, channel.ControlPlaneConformance(channelName))
+	env.TestSet(ctx, t, channel.DataPlaneConformance(channelName))
+}
+
+// TestSmoke_Channel
+func TestSmoke_Channel(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment()
+	t.Cleanup(env.Finish)
+
+	names := []string{
+		"customname",
+		"name-with-dash",
+		"name1with2numbers3",
+		"name63-01234567890123456789012345678901234567890123456789012345",
+	}
+
+	for _, name := range names {
+		env.Test(ctx, t, channel.GoesReady(name))
+	}
+}
+
+// TestSmoke_ChannelImpl
+func TestSmoke_ChannelImpl(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment()
+	t.Cleanup(env.Finish)
+
+	names := []string{
+		"customname",
+		"name-with-dash",
+		"name1with2numbers3",
+		"name63-01234567890123456789012345678901234567890123456789012345",
+	}
+
+	for _, name := range names {
+		env.Test(ctx, t, channel.ImplGoesReady(name))
+	}
+
+}
+
+// TestSmoke_ChannelWithSubscription
+func TestSmoke_ChannelWithSubscription(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment()
+	t.Cleanup(env.Finish)
+
+	channelName := "mychannel"
+
+	// Install and wait for a Ready Channel.
+	env.Prerequisite(ctx, t, channel.GoesReady(channelName))
+	chRef := ch.AsRef(channelName)
+
+	names := []string{
+		"customname",
+		"name-with-dash",
+		"name1with2numbers3",
+		"name63-01234567890123456789012345678901234567890123456789012345",
+	}
+
+	for _, name := range names {
+		env.Test(ctx, t, channel.SubscriptionGoesReady(name,
+			subscription.WithChannel(chRef),
+			subscription.WithSubscriber(nil, "http://example.com")),
+		)
+	}
+}
+
+// TestSmoke_ChannelImplWithSubscription
+func TestSmoke_ChannelImplWithSubscription(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment()
+	t.Cleanup(env.Finish)
+
+	channelName := "mychannelimpl"
+
+	// Install and wait for a Ready Channel.
+	env.Prerequisite(ctx, t, channel.ImplGoesReady(channelName))
+	chRef := chimpl.AsRef(channelName)
+
+	names := []string{
+		"customname",
+		"name-with-dash",
+		"name1with2numbers3",
+		"name63-01234567890123456789012345678901234567890123456789012345",
+	}
+
+	for _, name := range names {
+		env.Test(ctx, t, channel.SubscriptionGoesReady(name,
+			subscription.WithChannel(chRef),
+			subscription.WithSubscriber(nil, "http://example.com")),
+		)
+	}
+}
+
+/*
+TestChannelChainByUsingReplyAsSubscriber tests the following scenario:
+
+EventSource ---> (Channel ---> Subscription) x 10 ---> Sink
+
+It uses Subscription's spec.reply as spec.subscriber.
+
+This test should fail with https://github.com/knative/eventing/issues/5756 done.
+
+*/
+func TestChannelChainByUsingReplyAsSubscriber(t *testing.T) {
+	// This test fails with our KafkaChannel here since the test assumes it is ok to leave Subscription's `spec.subscriber` empty.
+	// Leaving that empty is not handled in the data channel as there won't be any `egress.Destination` in the contract resource.
+	// Dataplane won't know what to do when the destination is empty.
+	// See https://github.com/knative/eventing/issues/5756#issuecomment-1012996766
+	t.Skip("We don't allow nil `spec.subscriber` in Subscription. See https://github.com/knative/eventing/issues/5756#issuecomment-1012996766")
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	createSubscriberFn := func(ref *duckv1.KReference, uri string) manifest.CfgFn {
+		return subscription.WithReply(ref, uri)
+	}
+	env.Test(ctx, t, channel.ChannelChain(10, createSubscriberFn))
+}
+
+/*
+TestChannelChain tests the following scenario:
+
+EventSource ---> (Channel ---> Subscription) x 10 ---> Sink
+
+*/
+func TestChannelChain(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	createSubscriberFn := func(ref *duckv1.KReference, uri string) manifest.CfgFn {
+		return subscription.WithSubscriber(ref, uri)
+	}
+	env.Test(ctx, t, channel.ChannelChain(10, createSubscriberFn))
+}
+
+/*
+TestChannelDeadLetterSink tests if the events that cannot be delivered end up in
+the dead letter sink.
+
+It uses Subscription's spec.reply as spec.subscriber.
+*/
+func TestChannelDeadLetterSink(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	createSubscriberFn := func(ref *duckv1.KReference, uri string) manifest.CfgFn {
+		return subscription.WithSubscriber(ref, uri)
+	}
+	env.Test(ctx, t, channel.DeadLetterSink(createSubscriberFn))
+}
+
+/*
+TestChannelDeadLetterSinkByUsingReplyAsSubscriber tests if the events that cannot be delivered end up in
+the dead letter sink.
+
+It uses Subscription's spec.reply as spec.subscriber.
+
+This test should fail with https://github.com/knative/eventing/issues/5756 done.
+*/
+func TestChannelDeadLetterSinkByUsingReplyAsSubscriber(t *testing.T) {
+	// This test fails with our KafkaChannel here since the test assumes it is ok to leave Subscription's `spec.subscriber` empty.
+	// Leaving that empty is not handled in the data channel as there won't be any `egress.Destination` in the contract resource.
+	// Dataplane won't know what to do when the destination is empty.
+	// See https://github.com/knative/eventing/issues/5756#issuecomment-1012184663
+	t.Skip("We don't allow nil `spec.subscriber` in Subscription. See https://github.com/knative/eventing/issues/5756#issuecomment-1012184663")
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	createSubscriberFn := func(ref *duckv1.KReference, uri string) manifest.CfgFn {
+		return subscription.WithReply(ref, uri)
+	}
+	env.Test(ctx, t, channel.DeadLetterSink(createSubscriberFn))
+}
+
+/*
+TestEventTransformationForSubscription tests the following scenario:
+
+             1            2                 5            6                  7
+EventSource ---> Channel ---> Subscription ---> Channel ---> Subscription ----> Service(Logger)
+                                   |  ^
+                                 3 |  | 4
+                                   |  |
+                                   |  ---------
+                                   -----------> Service(Transformation)
+*/
+func TestEventTransformationForSubscriptionV1(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, channel.EventTransformation())
+}
+
+/*
+TestBinaryEventForChannel tests the following scenario:
+
+EventSource (binary-encoded messages) ---> Channel ---> Subscription ---> Service(Logger)
+
+*/
+func TestBinaryEventForChannel(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, channel.SingleEventWithEncoding(binding.EncodingBinary))
+}
+
+/*
+TestStructuredEventForChannel tests the following scenario:
+
+EventSource (structured-encoded messages) ---> Channel ---> Subscription ---> Service(Logger)
+
+*/
+func TestStructuredEventForChannel(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, channel.SingleEventWithEncoding(binding.EncodingStructured))
+}
diff --git a/test/e2e_new/main_test.go b/test/e2e_new/main_test.go
index c31e03102..f8c717918 100644
--- a/test/e2e_new/main_test.go
+++ b/test/e2e_new/main_test.go
@@ -24,6 +24,8 @@ import (
 	"os"
 	"testing"
 
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
@@ -45,6 +47,10 @@ func init() {
 	// environment.InitFlags registers state and level filter flags.
 	environment.InitFlags(flag.CommandLine)
 
+	// CHANNEL_GROUP_KIND=KafkaChannel.messaging.knative.dev;CHANNEL_VERSION=v1beta1
+	channel_impl.EnvCfg.ChannelGK = "KafkaChannel.messaging.knative.dev"
+	channel_impl.EnvCfg.ChannelV = "v1beta1"
+
 	// Force the broker class to be configured properly
 	broker.EnvCfg.BrokerClass = kafka.BrokerClass
 }
diff --git a/test/reconciler-tests.sh b/test/reconciler-tests.sh
index 829013be1..72636f544 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -21,7 +21,7 @@ header "Running tests"
 
 export_logs_continuously
 
-go_test_e2e -timeout=30m ./test/e2e_new/... || fail_test "E2E (new) suite failed"
+go_test_e2e -timeout=1h ./test/e2e_new/... || fail_test "E2E (new) suite failed"
 
 go_test_e2e -tags=deletecm ./test/e2e_new/... || fail_test "E2E (new deletecm) suite failed"
 
diff --git a/test/upgrade-tests.sh b/test/upgrade-tests.sh
index 7856de479..954548670 100755
--- a/test/upgrade-tests.sh
+++ b/test/upgrade-tests.sh
@@ -40,6 +40,7 @@ set -Eeuo pipefail
 TIMEOUT=${TIMEOUT:-60m}
 GO_TEST_VERBOSITY="${GO_TEST_VERBOSITY:-standard-verbose}"
 
+EVENTING_KAFKA_BROKER_UPGRADE_TESTS_FINISHEDSLEEP="5m" \
 go_test_e2e -v \
   -tags=upgrade \
   -timeout="${TIMEOUT}" \
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
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go b/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go
new file mode 100644
index 000000000..1c5cf2c3b
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go
@@ -0,0 +1,264 @@
+/*
+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 channel
+
+import (
+	"context"
+
+	"github.com/google/go-cmp/cmp"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/apis/messaging"
+	"knative.dev/eventing/test/rekt/features/knconf"
+	"knative.dev/eventing/test/rekt/resources/account_role"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/apis/duck"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/resources/svc"
+)
+
+func ControlPlaneConformance(channelName string) *feature.FeatureSet {
+	fs := &feature.FeatureSet{
+		Name: "Knative Channel Specification - Control Plane",
+		Features: []feature.Feature{
+			*ControlPlaneChannel(channelName),
+		},
+	}
+
+	return fs
+}
+
+func ControlPlaneChannel(channelName string) *feature.Feature {
+	f := feature.NewFeatureNamed("Conformance")
+
+	f.Setup("Set Channel Name", setChannelableName(channelName))
+
+	sacmName := feature.MakeRandomK8sName("channelable-manipulator")
+	f.Setup("Create Service Account for Channelable Manipulator",
+		account_role.Install(sacmName, account_role.AsChannelableManipulator))
+
+	saarName := feature.MakeRandomK8sName("addressale-resolver")
+	f.Setup("Create Service Account for Addressable Resolver",
+		account_role.Install(saarName, account_role.AsAddressableResolver))
+
+	f.Stable("Aggregated Channelable Manipulator ClusterRole").
+		Must("Every CRD MUST create a corresponding ClusterRole, that will be aggregated into the channelable-manipulator ClusterRole."+
+			"This ClusterRole MUST include permissions to create, get, list, watch, patch, update and delete the CRD's custom objects and their status. "+
+			"Each channel MUST have the duck.knative.dev/channelable: \"true\" label on its channelable-manipulator ClusterRole.",
+			serviceAccountIsChannelableManipulator(sacmName))
+
+	f.Stable("Aggregated Addressable Resolver ClusterRole").
+		Must("Every CRD MUST create a corresponding ClusterRole, that will be aggregated into the addressable-resolver ClusterRole. "+
+			"This ClusterRole MUST include permissions to get, list, and watch the CRD's custom objects and their status. "+
+			"Each channel MUST have the duck.knative.dev/addressable: \"true\" label on its addressable-resolver ClusterRole.",
+			serviceAccountIsAddressableResolver(saarName))
+
+	f.Stable("CustomResourceDefinition per Channel").
+		Must("Each channel is namespaced", crdOfChannelIsNamespaced).
+		Must("label of messaging.knative.dev/subscribable: true",
+			crdOfChannelIsLabeled(messaging.SubscribableDuckVersionAnnotation, "true")).
+		Must("label of duck.knative.dev/addressable: true",
+			crdOfChannelIsLabeled(duck.AddressableDuckVersionLabel, "true")).
+		Must("The category `channel`", crdOfChannelHasCategory("channel"))
+
+	f.Stable("Annotation Requirements").
+		Should("each instance SHOULD have annotation: messaging.knative.dev/subscribable: v1",
+			channelHasAnnotations)
+
+	f.Stable("Spec and Status Requirements for Subscribers").
+		Must("Each channel CRD MUST contain an array of subscribers: spec.subscribers. "+
+			"Each channel CRD MUST have a status subresource which contains [subscribers (as an array)]. "+
+			"The ready field of the subscriber identified by its uid MUST be set to True when the subscription is ready to be processed.",
+			channelAllowsSubscribersAndStatus)
+		// Special note for Channel tests: The array of subscribers MUST NOT be
+		// set directly on the generic Channel custom object, but rather
+		// appended to the backing channel by the subscription itself.
+
+	f.Stable("Status Requirements").
+		Must("observedGeneration MUST be populated if present",
+			knconf.KResourceHasObservedGeneration(channel_impl.GVR(), channelName)).
+		Should("SHOULD have in status observedGeneration. "+
+			"SHOULD have in status conditions (as an array)",
+			knconf.KResourceHasReadyInConditions(channel_impl.GVR(), channelName)).
+		Should("status.conditions SHOULD indicate status transitions and error reasons if present",
+			todo) // how to test this?
+
+	cName := feature.MakeRandomK8sName("channel")
+	sink := feature.MakeRandomK8sName("sink")
+
+	f.Setup("Set Channel Name", setChannelableName(cName))
+
+	f.Setup("install a service", svc.Install(sink, "app", "rekt"))
+	f.Setup("update Channel", channel_impl.Install(cName, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
+	f.Setup("Channel goes ready", channel_impl.IsReady(cName))
+
+	f.Stable("Channel Status").
+		Must("When the channel instance is ready to receive events status.address.url MUST be populated. "+
+			"Each Channel CRD MUST have a status subresource which contains [address]. "+
+			"When the Channel instance is ready to receive events status.address.url status.addressable MUST be set to True",
+			readyChannelIsAddressable).
+		Should("Set the Channel status.deadLetterSinkURI if there is a valid spec.delivery.deadLetterSink defined",
+			readyChannelWithDLSHaveStatusUpdated)
+
+	return f
+}
+
+func serviceAccountIsChannelableManipulator(name string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		gvr := channel_impl.GVR()
+		for _, verb := range []string{"get", "list", "watch", "update", "patch", "delete"} {
+			ServiceAccountSubjectAccessReviewAllowedOrFail(ctx, t, gvr, "", name, verb)
+			ServiceAccountSubjectAccessReviewAllowedOrFail(ctx, t, gvr, "status", name, verb)
+		}
+	}
+}
+
+func serviceAccountIsAddressableResolver(name string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		gvr := channel_impl.GVR()
+		for _, verb := range []string{"get", "list", "watch"} {
+			ServiceAccountSubjectAccessReviewAllowedOrFail(ctx, t, gvr, "", name, verb)
+			ServiceAccountSubjectAccessReviewAllowedOrFail(ctx, t, gvr, "status", name, verb)
+		}
+	}
+}
+
+func channelHasAnnotations(ctx context.Context, t feature.T) {
+	ch := getChannelable(ctx, t)
+	if version, found := ch.Annotations["messaging.knative.dev/subscribable"]; !found {
+		t.Error(`expected annotations["messaging.knative.dev/subscribable"] to exist`)
+	} else if version != "v1" {
+		t.Error(`expected "messaging.knative.dev/subscribable" to be "v1", found`, version)
+	}
+}
+
+func channelAllowsSubscribersAndStatus(ctx context.Context, t feature.T) {
+	ch := getChannelable(ctx, t)
+	original := ch.DeepCopy()
+
+	u, _ := apis.ParseURL("http://example.com")
+	want := duckv1.SubscriberSpec{
+		UID:           "abc123",
+		Generation:    1,
+		SubscriberURI: u,
+	}
+
+	ch.Spec.Subscribers = append(ch.Spec.Subscribers, want)
+
+	patchChannelable(ctx, t, original, ch)
+
+	var updated *duckv1.Channelable
+	interval, timeout := environment.PollTimingsFromContext(ctx)
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		updated = getChannelable(ctx, t)
+		if updated.Status.ObservedGeneration != updated.Generation {
+			// keep polling.
+			return false, nil
+		}
+		if len(updated.Status.Subscribers) == len(ch.Spec.Subscribers) {
+			for _, got := range updated.Status.Subscribers {
+				// want should be Ready.
+				if got.UID == want.UID {
+					if want := corev1.ConditionTrue; got.Ready == want {
+						// Synced!
+						return true, nil
+					}
+				}
+			}
+		}
+		// keep polling.
+		return false, nil
+	})
+	if err != nil {
+		t.Fatalf("failed waiting for channel subscribers to sync", err)
+	}
+
+	if len(updated.Spec.Subscribers) <= 0 {
+		t.Errorf("subscriber was not saved")
+	}
+
+	found := false
+	for _, got := range updated.Spec.Subscribers {
+		if got.UID == want.UID {
+			found = true
+			if diff := cmp.Diff(want, got); diff != "" {
+				t.Error("Round trip Subscriber has a delta, (-want, +got) =", diff)
+			}
+		}
+	}
+	if !found {
+		t.Error("Round trip Subscriber failed.")
+	}
+
+	if len(updated.Status.Subscribers) != 1 {
+		t.Error("Subscribers not in status.")
+	} else {
+		for _, got := range updated.Status.Subscribers {
+			// want should be Ready.
+			if got.UID == want.UID {
+				if want := corev1.ConditionTrue; got.Ready != want {
+					t.Error("Expected subscriber to be %q, got %q", want, got.Ready)
+				}
+			}
+		}
+	}
+}
+
+func readyChannelIsAddressable(ctx context.Context, t feature.T) {
+	var ch *duckv1.Channelable
+
+	// Poll for a ready channel.
+	interval, timeout := environment.PollTimingsFromContext(ctx)
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		ch = getChannelable(ctx, t)
+		if c := ch.Status.GetCondition(apis.ConditionReady); c.IsTrue() {
+			return true, nil
+		}
+		return false, nil
+	})
+	if err != nil {
+		t.Fatalf("failed to get a ready channel", err)
+	}
+
+	// Confirm the channel is ready, and addressable.
+	if c := ch.Status.GetCondition(apis.ConditionReady); c.IsTrue() {
+		if ch.Status.Address.URL == nil {
+			t.Errorf("channel is not addressable")
+		}
+		// Success!
+	} else {
+		t.Errorf("channel was not ready, reason: %s", ch.Status.GetCondition(apis.ConditionReady).Reason)
+	}
+}
+
+func readyChannelWithDLSHaveStatusUpdated(ctx context.Context, t feature.T) {
+	ch := getChannelable(ctx, t)
+
+	// Confirm the channel is ready, and has the status.deadLetterSinkURI set.
+	if c := ch.Status.GetCondition(apis.ConditionReady); c.IsTrue() {
+		if ch.Status.DeadLetterSinkURI == nil {
+			t.Errorf("channel DLS not resolved but resource reported ready")
+		}
+		// Success!
+	} else {
+		t.Errorf("channel was not ready, reason: %s", ch.Status.GetCondition(apis.ConditionReady).Reason)
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go b/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go
new file mode 100644
index 000000000..8e9ae7cd7
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go
@@ -0,0 +1,219 @@
+/*
+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 channel
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/test/rekt/features/knconf"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/ptr"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/state"
+)
+
+func DataPlaneConformance(channelName string) *feature.FeatureSet {
+	fs := &feature.FeatureSet{
+		Name: "Knative Channel Specification - Data Plane",
+		Features: []feature.Feature{
+			*DataPlaneChannel(channelName),
+		},
+	}
+
+	addControlPlaneDelivery(fs)
+
+	return fs
+}
+
+func DataPlaneChannel(channelName string) *feature.Feature {
+	f := feature.NewFeatureNamed("Conformance")
+
+	f.Setup("Set Channel Name", setChannelableName(channelName))
+
+	f.Requirement("Channel is Ready", channel_impl.IsReady(channelName))
+
+	f.Stable("Input").
+		Must("Every Channel MUST expose either an HTTP or HTTPS endpoint.", todo).
+		Must("The endpoint(s) MUST conform to 0.3 or 1.0 CloudEvents specification.",
+			channelAcceptsCEVersions).
+		MustNot("The Channel MUST NOT perform an upgrade of the passed in version. It MUST emit the event with the same version.", todo).
+		Must("It MUST support both Binary Content Mode and Structured Content Mode of the HTTP Protocol Binding for CloudEvents.", todo).
+		May("When dispatching the event, the channel MAY use a different HTTP Message mode of the one used by the event.", todo).
+		// For example, It MAY receive an event in Structured Content Mode and dispatch in Binary Content Mode.
+		May("The HTTP(S) endpoint MAY be on any port, not just the standard 80 and 443.", todo).
+		May("Channels MAY expose other, non-HTTP endpoints in addition to HTTP at their discretion.", todo)
+		// (e.g. expose a gRPC endpoint to accept events)
+
+	f.Stable("Generic").
+		Must("If a Channel receives an event queueing request and is unable to parse a valid CloudEvent, then it MUST reject the request.", todo)
+
+	f.Stable("HTTP").
+		Must("Channels MUST reject all HTTP event queueing requests with a method other than POST responding with HTTP status code 405 Method Not Supported.", todo).
+		Must("The HTTP event queueing request's URL MUST correspond to a single, unique Channel at any given moment in time.", todo).
+		May("This MAY be done via the host, path, query string, or any combination of these.", todo).
+		Must("If a Channel receives a request that does not correspond to a known channel, then it MUST respond with a 404 Not Found.", todo).
+		Must("The Channel MUST respond with 202 Accepted if the event queueing request is accepted by the server.", todo).
+		Must("If a Channel receives an event queueing request and is unable to parse a valid CloudEvent, then it MUST respond with 400 Bad Request.", todo)
+
+	f.Stable("Output").
+		Must("Channels MUST output CloudEvents.", todo).
+		Must("The output MUST match the CloudEvent version of the Input.", todo).
+		MustNot("Channels MUST NOT alter an event that goes through them.", todo).
+		Must("All CloudEvent attributes, including the data attribute, MUST be received at the subscriber identical to how they were received by the Channel.", todo).
+		Should("Every Channel SHOULD support sending events via Binary Content Mode or Structured Content Mode of the HTTP Protocol Binding for CloudEvents.", todo).
+		Must("Channels MUST send events to all subscribers which are marked with a status of ready: True in the channel's status.subscribers.", todo).
+		Must("The events MUST be sent to the subscriberURI field of spec.subscribers.", todo)
+		// Each channel implementation will have its own quality of service guarantees (e.g. at least once, at most once, etc) which SHOULD be documented.
+
+	f.Stable("Observability").
+		Should("Channels SHOULD expose a variety of metrics: [Number of malformed incoming event queueing events (400 Bad Request responses)]", todo).
+		Should("Channels SHOULD expose a variety of metrics: [Number of accepted incoming event queuing events (202 Accepted responses)]", todo).
+		Should("Channels SHOULD expose a variety of metrics: [Number of egress CloudEvents produced (with the former metric, used to derive channel queue size)]", todo).
+		Should("Metrics SHOULD be enabled by default, with a configuration parameter included to disable them if desired.", todo).
+		Must("The Channel MUST recognize and pass through all tracing information from sender to subscribers using W3C Tracecontext.", todo).
+		Should("The Channel SHOULD sample and write traces to the location specified in config-tracing.", todo).
+		Should("Spans emitted by the Channel SHOULD follow the OpenTelemetry Semantic Conventions for Messaging Systems whenever possible.", todo).
+		Should("Spans emitted by the Channel SHOULD set the knative attributes.", todo)
+		// messaging.system: "knative"
+		// messaging.destination: url to which the event is being routed
+		// messaging.protocol: the name of the underlying transport protocol
+		// messaging.message_id: the event ID
+
+	return f
+}
+
+func channelAcceptsCEVersions(ctx context.Context, t feature.T) {
+	name := state.GetStringOrFail(ctx, t, ChannelableNameKey)
+	knconf.AcceptsCEVersions(ctx, t, channel_impl.GVR(), name)
+}
+
+func addControlPlaneDelivery(fs *feature.FeatureSet) {
+	//Should("Channels SHOULD retry resending CloudEvents when they fail to either connect or send CloudEvents to subscribers.", todo).
+	//Should("Channels SHOULD support various retry configuration parameters: [the maximum number of retries]", todo).
+	//Should("Channels SHOULD support various retry configuration parameters: [the time in-between retries]", todo).
+	//Should("Channels SHOULD support various retry configuration parameters: [the backoff rate]", todo)
+
+	for i, tt := range []struct {
+		name string
+		chDS *v1.DeliverySpec
+		subs []subCfg
+	}{{
+		name: "Channels SHOULD retry resending CloudEvents when they fail to either connect or send CloudEvents to subscribers.",
+		subs: []subCfg{{
+			prefix:         "sub1",
+			hasSub:         true,
+			subFailCount:   1,
+			subReplies:     false,
+			hasReply:       false,
+			replyFailCount: 0,
+			delivery: &v1.DeliverySpec{
+				DeadLetterSink: new(duckv1.Destination),
+				Retry:          ptr.Int32(1),
+			},
+		}},
+	}, {
+		name: "Channels SHOULD support various retry configuration parameters: [the maximum number of retries]",
+		subs: []subCfg{{
+			prefix:         "sub1",
+			hasSub:         true,
+			subFailCount:   1,
+			subReplies:     false,
+			hasReply:       false,
+			replyFailCount: 0,
+			delivery: &v1.DeliverySpec{
+				DeadLetterSink: new(duckv1.Destination),
+				Retry:          ptr.Int32(1),
+			},
+		}, {
+			prefix:         "sub2",
+			hasSub:         true,
+			subFailCount:   1,
+			subReplies:     true,
+			hasReply:       true,
+			replyFailCount: 0,
+			delivery: &v1.DeliverySpec{
+				DeadLetterSink: new(duckv1.Destination),
+				Retry:          ptr.Int32(1),
+			},
+		}, {
+			prefix:         "sub3",
+			hasSub:         true,
+			subFailCount:   1,
+			subReplies:     true,
+			hasReply:       true,
+			replyFailCount: 1,
+			delivery: &v1.DeliverySpec{
+				DeadLetterSink: new(duckv1.Destination),
+				Retry:          ptr.Int32(1),
+			},
+		}, {
+			prefix:         "sub4",
+			hasSub:         true,
+			subFailCount:   1,
+			subReplies:     true,
+			hasReply:       true,
+			replyFailCount: 2,
+			delivery: &v1.DeliverySpec{
+				DeadLetterSink: new(duckv1.Destination),
+				Retry:          ptr.Int32(1),
+			},
+		}},
+	}} {
+		f := feature.NewFeatureNamed("Delivery Spec " + strconv.Itoa(i))
+
+		chName := fmt.Sprintf("dlq-test-%d", i)
+		f.Setup("Set Broker Name", setChannelableName(chName))
+
+		prober := createChannelTopology(f, chName, tt.chDS, tt.subs)
+
+		// Send an event into the matrix and hope for the best
+		prober.SenderFullEvents(1)
+		f.Setup("install source", prober.SenderInstall("source"))
+		f.Requirement("sender is finished", prober.SenderDone("source"))
+
+		// All events have been sent, time to look at the specs and confirm we got them.
+		expected := createExpectedEventPatterns(tt.chDS, tt.subs)
+
+		f.Requirement("wait until done", func(ctx context.Context, t feature.T) {
+			interval, timeout := environment.PollTimingsFromContext(ctx)
+			err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+				gtg := true
+				for prefix, want := range expected {
+					events := prober.ReceivedOrRejectedBy(ctx, prefix)
+					if len(events) != len(want.Success) {
+						gtg = false
+					}
+				}
+				return gtg, nil
+			})
+			if err != nil {
+				t.Failed()
+			}
+		})
+
+		f.Stable("Conformance").Should(tt.name, knconf.AssertEventPatterns(prober, expected))
+
+		// Add this feature to the feature set.
+		fs.Features = append(fs.Features, *f)
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/features.go b/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
new file mode 100644
index 000000000..b4113d872
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
@@ -0,0 +1,191 @@
+/*
+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 channel
+
+import (
+	"context"
+	"fmt"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/binding"
+	"github.com/cloudevents/sdk-go/v2/test"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/eventing/test/rekt/resources/containersource"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/eventing/test/rekt/resources/eventlibrary"
+	"knative.dev/eventing/test/rekt/resources/pingsource"
+	"knative.dev/eventing/test/rekt/resources/source"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+	"knative.dev/reconciler-test/resources/svc"
+)
+
+func ChannelChain(length int, createSubscriberFn func(ref *duckv1.KReference, uri string) manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeature()
+	sink := feature.MakeRandomK8sName("sink")
+	cs := feature.MakeRandomK8sName("containersource")
+
+	var channels []string
+	for i := 0; i < length; i++ {
+		name := feature.MakeRandomK8sName(fmt.Sprintf("channel-%04d", i))
+		channels = append(channels, name)
+		f.Setup("install channel", channel_impl.Install(name))
+		f.Requirement("channel is ready", channel_impl.IsReady(name))
+	}
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+	// attach the first channel to the source
+	f.Setup("install containersource", containersource.Install(cs, pingsource.WithSink(channel_impl.AsRef(channels[0]), "")))
+
+	// use the rest for the chain
+	for i := 0; i < length; i++ {
+		sub := feature.MakeRandomK8sName(fmt.Sprintf("subscription-%04d", i))
+		if i == length-1 {
+			// install the final connection to the sink
+			f.Setup("install sink subscription", subscription.Install(sub,
+				subscription.WithChannel(channel_impl.AsRef(channels[i])),
+				createSubscriberFn(svc.AsKReference(sink), ""),
+			))
+		} else {
+			f.Setup("install subscription", subscription.Install(sub,
+				subscription.WithChannel(channel_impl.AsRef(channels[i])),
+				createSubscriberFn(channel_impl.AsRef(channels[i+1]), ""),
+			))
+		}
+	}
+	f.Requirement("containersource goes ready", containersource.IsReady(cs))
+
+	f.Assert("chained channels relay events", assert.OnStore(sink).MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).AtLeast(1))
+
+	return f
+}
+
+func DeadLetterSink(createSubscriberFn func(ref *duckv1.KReference, uri string) manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeature()
+	sink := feature.MakeRandomK8sName("sink")
+	failer := feature.MakeK8sNamePrefix("failer")
+	cs := feature.MakeRandomK8sName("containersource")
+	name := feature.MakeRandomK8sName("channel")
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+	f.Setup("install failing receiver", eventshub.Install(failer, eventshub.StartReceiver, eventshub.DropFirstN(1)))
+	f.Setup("install channel", channel_impl.Install(name, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
+	f.Setup("install containersource", containersource.Install(cs, source.WithSink(channel_impl.AsRef(name), "")))
+	f.Setup("install subscription", subscription.Install(feature.MakeRandomK8sName("subscription"),
+		subscription.WithChannel(channel_impl.AsRef(name)),
+		createSubscriberFn(svc.AsKReference(failer), ""),
+	))
+
+	f.Requirement("channel is ready", channel_impl.IsReady(name))
+	f.Requirement("containersource is ready", containersource.IsReady(cs))
+
+	f.Assert("dls receives events", assert.OnStore(sink).MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).AtLeast(1))
+
+	return f
+}
+
+func EventTransformation() *feature.Feature {
+	f := feature.NewFeature()
+	lib := feature.MakeRandomK8sName("lib")
+	channel1 := feature.MakeRandomK8sName("channel 1")
+	channel2 := feature.MakeRandomK8sName("channel 2")
+	subscription1 := feature.MakeRandomK8sName("subscription 1")
+	subscription2 := feature.MakeRandomK8sName("subscription 2")
+	prober := eventshub.NewProber()
+	prober.SetTargetResource(channel_impl.GVR(), channel1)
+
+	f.Setup("install events", eventlibrary.Install(lib))
+	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
+	f.Setup("register event expectation", func(ctx context.Context, t feature.T) {
+		if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
+			t.Fatalf("can not find event files: %v", err)
+		}
+	})
+
+	f.Setup("install sink", prober.ReceiverInstall("sink"))
+	f.Setup("install transform service", prober.ReceiverInstall("transform", eventshub.ReplyWithTransformedEvent("transformed", "transformer", "")))
+	f.Setup("install channel 1", channel_impl.Install(channel1))
+	f.Setup("install channel 2", channel_impl.Install(channel2))
+	f.Setup("install subscription 1", subscription.Install(subscription1,
+		subscription.WithChannel(channel_impl.AsRef(channel1)),
+		subscription.WithSubscriber(prober.AsKReference("transform"), ""),
+		subscription.WithReply(channel_impl.AsRef(channel2), ""),
+	))
+	f.Setup("install subscription 2", subscription.Install(subscription2,
+		subscription.WithChannel(channel_impl.AsRef(channel2)),
+		subscription.WithSubscriber(prober.AsKReference("sink"), ""),
+	))
+	f.Setup("subscription 1 is ready", subscription.IsReady(subscription1))
+	f.Setup("subscription 2 is ready", subscription.IsReady(subscription2))
+	f.Setup("channel 1 is ready", channel_impl.IsReady(channel1))
+	f.Setup("channel 2 is ready", channel_impl.IsReady(channel2))
+	f.Setup("install source", prober.SenderInstall("source"))
+	f.Setup("event library is ready", eventlibrary.IsReady(lib))
+
+	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("receiver is finished", prober.ReceiverDone("source", "sink"))
+
+	f.Assert("sink receives events", prober.AssertReceivedAll("source", "sink"))
+	f.Assert("events have passed through transform service", func(ctx context.Context, t feature.T) {
+		events := prober.ReceivedBy(ctx, "sink")
+		if len(events) != 3 {
+			t.Errorf("expected 3 events, got %d", len(events))
+		}
+		for _, e := range events {
+			if e.Event.Type() != "transformed" {
+				t.Errorf(`expected event type to be "transformed", got %q`, e.Event.Type())
+			}
+		}
+	})
+	return f
+}
+
+func SingleEventWithEncoding(encoding binding.Encoding) *feature.Feature {
+	f := feature.NewFeature()
+	channel := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
+	prober := eventshub.NewProber()
+	prober.SetTargetResource(channel_impl.GVR(), channel)
+
+	event := cloudevents.NewEvent()
+	event.SetID(feature.MakeRandomK8sName("test"))
+	event.SetType("myevent")
+	event.SetSource("http://sender.svc/")
+	prober.ExpectEvents([]string{event.ID()})
+
+	f.Setup("install sink", prober.ReceiverInstall("sink"))
+	f.Setup("install channel", channel_impl.Install(channel))
+	f.Setup("install subscription", subscription.Install(sub,
+		subscription.WithChannel(channel_impl.AsRef(channel)),
+		subscription.WithSubscriber(prober.AsKReference("sink"), ""),
+	))
+
+	f.Setup("subscription is ready", subscription.IsReady(sub))
+	f.Setup("channel is ready", channel_impl.IsReady(channel))
+	f.Setup("install source", prober.SenderInstall("source", eventshub.InputEventWithEncoding(event, encoding)))
+
+	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("receiver is finished", prober.ReceiverDone("source", "sink"))
+
+	f.Assert("sink receives events", prober.AssertReceivedAll("source", "sink"))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go b/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go
new file mode 100644
index 000000000..619a1cc29
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go
@@ -0,0 +1,185 @@
+/*
+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 channel
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	authv1 "k8s.io/api/authorization/v1"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/dynamic"
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingclientsetv1 "knative.dev/eventing/pkg/client/clientset/versioned/typed/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/pkg/apis/duck"
+	apiextensionsclient "knative.dev/pkg/client/injection/apiextensions/client"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/injection/clients/dynamicclient"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/state"
+)
+
+func todo(ctx context.Context, t feature.T) {
+	t.Log("TODO, Implement this.")
+}
+
+type EventingClient struct {
+	Channels    messagingclientsetv1.ChannelInterface
+	ChannelImpl dynamic.ResourceInterface
+}
+
+func Client(ctx context.Context) *EventingClient {
+	env := environment.FromContext(ctx)
+
+	mc := eventingclient.Get(ctx).MessagingV1()
+	dc := dynamicclient.Get(ctx)
+
+	return &EventingClient{
+		Channels:    mc.Channels(env.Namespace()),
+		ChannelImpl: dc.Resource(channel_impl.GVR()).Namespace(env.Namespace()),
+	}
+}
+
+const (
+	ChannelableNameKey = "channelableName"
+)
+
+func setChannelableName(name string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		state.SetOrFail(ctx, t, ChannelableNameKey, name)
+	}
+}
+
+func getChannelable(ctx context.Context, t feature.T) *duckv1.Channelable {
+	c := Client(ctx)
+	name := state.GetStringOrFail(ctx, t, ChannelableNameKey)
+
+	obj, err := c.ChannelImpl.Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		t.Errorf("failed to get ChannelImpl, %v", err)
+	}
+
+	channel := &duckv1.Channelable{}
+	if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.Object, channel); err != nil {
+		t.Fatalf("Failed to convert channelImpl to Channelable: %v", err)
+	}
+
+	channel.ResourceVersion = channel_impl.GVR().Version
+	channel.APIVersion = channel_impl.GVR().GroupVersion().String()
+
+	return channel
+}
+
+func patchChannelable(ctx context.Context, t feature.T, before, after *duckv1.Channelable) {
+	patch, err := duck.CreateMergePatch(before, after)
+	if err != nil {
+		t.Fatalf("Failed to create merge patch: %v", err)
+	}
+	// If there is nothing to patch, we are good, just return.
+	// Empty patch is {}, hence we check for that.
+	if len(patch) <= 2 {
+		return
+	}
+
+	_, err = Client(ctx).ChannelImpl.Patch(ctx, before.GetName(), types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		t.Fatal("Failed to patch the ChannelImpl", zap.Error(err), zap.Any("patch", patch))
+	}
+}
+
+func crdOfChannel(ctx context.Context, t feature.T) *apiextv1.CustomResourceDefinition {
+	gvr := channel_impl.GVR()
+	name := strings.Join([]string{gvr.Resource, gvr.Group}, ".")
+
+	crd, err := apiextensionsclient.Get(ctx).ApiextensionsV1().CustomResourceDefinitions().Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		t.Log(err)
+		t.FailNow()
+	}
+	return crd
+}
+
+func crdOfChannelIsNamespaced(ctx context.Context, t feature.T) {
+	crd := crdOfChannel(ctx, t)
+
+	if crd.Spec.Scope != apiextv1.NamespaceScoped {
+		t.Logf("%q CRD is not namespaced", crd.Name)
+		t.Fail()
+	}
+}
+
+func crdOfChannelIsLabeled(key, want string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		crd := crdOfChannel(ctx, t)
+
+		if got, found := crd.Labels[key]; !found {
+			t.Logf("%q CRD does not have label %q", crd.Name, key)
+			t.Fail()
+		} else if got != want {
+			t.Logf("%q CRD label %q expected to be %s, got %s", crd.Name, key, want, got)
+			t.Fail()
+		}
+	}
+}
+
+func crdOfChannelHasCategory(want string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		crd := crdOfChannel(ctx, t)
+
+		for _, got := range crd.Spec.Names.Categories {
+			if got == want {
+				// Success!
+				return
+			}
+		}
+		t.Logf("%q CRD does not have Category %q", crd.Name, want)
+	}
+}
+
+// TODO: move this to framework?
+func ServiceAccountSubjectAccessReviewAllowedOrFail(ctx context.Context, t feature.T, gvr schema.GroupVersionResource, subresource string, saName string, verb string) {
+	env := environment.FromContext(ctx)
+	kube := kubeclient.Get(ctx)
+
+	r, err := kube.AuthorizationV1().SubjectAccessReviews().Create(ctx, &authv1.SubjectAccessReview{
+		Spec: authv1.SubjectAccessReviewSpec{
+			User: fmt.Sprintf("system:serviceaccount:%s:%s", env.Namespace(), saName),
+			ResourceAttributes: &authv1.ResourceAttributes{
+				Verb:        verb,
+				Group:       gvr.Group,
+				Version:     gvr.Version,
+				Resource:    gvr.Resource,
+				Subresource: subresource,
+			},
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatalf("Error while checking if %q is not allowed on %s.%s/%s subresource:%q. err: %q", verb, gvr.Resource, gvr.Group, gvr.Version, subresource, err)
+	}
+	if !r.Status.Allowed {
+		t.Fatalf("Operation %q is not allowed on %s.%s/%s subresource:%q", verb, gvr.Resource, gvr.Group, gvr.Version, subresource)
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/readyness.go b/vendor/knative.dev/eventing/test/rekt/features/channel/readyness.go
new file mode 100644
index 000000000..d620cb7f4
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/readyness.go
@@ -0,0 +1,69 @@
+/*
+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 channel
+
+import (
+	"fmt"
+
+	"knative.dev/eventing/test/rekt/resources/channel"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+// GoesReady returns a feature testing if a Channel becomes ready.
+func GoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeatureNamed("Channel goes ready.")
+
+	f.Setup(fmt.Sprintf("install a Channel named %q", name), channel.Install(name, cfg...))
+
+	f.Requirement("Channel is ready", channel.IsReady(name))
+
+	f.Stable("Channel").
+		Must("be addressable", channel.IsAddressable(name))
+
+	return f
+}
+
+// ImplGoesReady returns a feature testing if a channel impl becomes ready.
+func ImplGoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeatureNamed("ChannelImpl goes ready.")
+
+	f.Setup(fmt.Sprintf("install a %s named %q", channel_impl.GVK().Kind, name), channel_impl.Install(name, cfg...))
+
+	f.Requirement(channel_impl.GVK().Kind+" is ready", channel_impl.IsReady(name))
+
+	f.Stable("ChannelImpl").
+		Must("be addressable", channel_impl.IsAddressable(name))
+
+	return f
+}
+
+// SubscriptionGoesReady returns a feature testing if a Subscription becomes
+// ready.
+func SubscriptionGoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeatureNamed("Subscription goes ready.")
+
+	f.Setup(fmt.Sprintf("install a Subscription named %q", name), subscription.Install(name, cfg...))
+
+	f.Requirement("Subscription is ready", subscription.IsReady(name))
+
+	f.Stable("Subscription")
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/topology.go b/vendor/knative.dev/eventing/test/rekt/features/channel/topology.go
new file mode 100644
index 000000000..086ff2298
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/topology.go
@@ -0,0 +1,193 @@
+/*
+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 channel
+
+import (
+	"fmt"
+	"strconv"
+
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/test/rekt/features/knconf"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/eventing/test/rekt/resources/subscription"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+type subCfg struct {
+	prefix         string
+	hasSub         bool
+	subFailCount   uint
+	subReplies     bool
+	hasReply       bool
+	replyFailCount uint
+	delivery       *v1.DeliverySpec
+}
+
+func (s *subCfg) subName() string {
+	suffix := "sub"
+	return fmt.Sprintf("%s%s", s.prefix, suffix)
+}
+
+func (s *subCfg) replyName() string {
+	suffix := "reply"
+	return fmt.Sprintf("%s%s", s.prefix, suffix)
+}
+
+func (s *subCfg) dlqName() string {
+	suffix := "dlq"
+	return fmt.Sprintf("%s%s", s.prefix, suffix)
+}
+
+// createChannelTopology creates a channel and {n} subscriptions with recorders
+// attached to each endpoint.
+//
+//  source --> [channel (chDS)] --+--[sub1 (sub1DS)]--> sub1sub (optional) --> sub1reply (optional)
+//                        |       |            |
+//                        |       |            +-->sub1dlq (optional)
+//                        |      ...
+//                        |       +-[sub{n} (sub{n}DS)]--> sub{n}sub (optional)--> sub{n}reply (optional)
+//                        |                   |
+//                        |                   +-->sub{n}dlq (optional)
+//                        |
+//                        +--[DLQ]--> chdlq (optional)
+//
+func createChannelTopology(f *feature.Feature, chName string, chDS *v1.DeliverySpec, subs []subCfg) *eventshub.EventProber {
+	prober := eventshub.NewProber()
+	// Install the receivers.
+	f.Setup("install channel DLQ", prober.ReceiverInstall("chdlq"))
+
+	var chOpts []manifest.CfgFn
+	if chDS != nil {
+		if chDS.DeadLetterSink != nil {
+			chOpts = append(chOpts, delivery.WithDeadLetterSink(prober.AsKReference("chdlq"), ""))
+		}
+		if chDS.Retry != nil {
+			chOpts = append(chOpts, delivery.WithRetry(*chDS.Retry, chDS.BackoffPolicy, chDS.BackoffDelay))
+		}
+	}
+	f.Setup("Create Channel Impl", channel_impl.Install(chName, chOpts...))
+	f.Setup("Channel is Ready", channel_impl.IsReady(chName)) // We want to block in the setup phase until the channel is ready to go.
+
+	// Set the prober target.
+	prober.SetTargetResource(channel_impl.GVR(), chName)
+
+	// Install subscriptions.
+	for i, sub := range subs {
+		// Install the expected sinks, they all might not be used.
+
+		subOpts := []eventshub.EventsHubOption{eventshub.DropFirstN(sub.subFailCount)}
+		if sub.subReplies {
+			subOpts = append(subOpts, eventshub.ReplyWithAppendedData(sub.prefix))
+		}
+		f.Setup("install subscription"+strconv.Itoa(i)+" subscriber",
+			prober.ReceiverInstall(sub.subName(), subOpts...))
+
+		f.Setup("install subscription"+strconv.Itoa(i)+" reply",
+			prober.ReceiverInstall(sub.replyName(), eventshub.DropFirstN(sub.replyFailCount)))
+		f.Setup("install subscription"+strconv.Itoa(i)+" DLQ",
+			prober.ReceiverInstall(sub.dlqName()))
+
+		var opts []manifest.CfgFn
+		if sub.hasSub {
+			opts = append(opts, subscription.WithSubscriber(prober.AsKReference(sub.subName()), ""))
+		}
+
+		if sub.hasReply {
+			opts = append(opts, subscription.WithReply(prober.AsKReference(sub.replyName()), ""))
+		}
+
+		if sub.delivery.DeadLetterSink != nil {
+			if sub.delivery != nil {
+				opts = append(opts, subscription.WithDeadLetterSink(prober.AsKReference(sub.dlqName()), ""))
+			}
+			if sub.delivery.Retry != nil {
+				opts = append(opts, subscription.WithRetry(*sub.delivery.Retry, sub.delivery.BackoffPolicy, sub.delivery.BackoffDelay))
+			}
+		}
+		opts = append(opts, subscription.WithChannel(channel_impl.AsRef(chName)))
+		name := feature.MakeRandomK8sName(sub.prefix)
+		f.Setup("install subscription"+strconv.Itoa(i), subscription.Install(name, opts...))
+		f.Setup("subscription"+strconv.Itoa(i)+" is ready", subscription.IsReady(name))
+	}
+
+	return prober
+}
+
+// createExpectedEventPatterns assumes a single event is sent.
+func createExpectedEventPatterns(chDS *v1.DeliverySpec, subs []subCfg) map[string]knconf.EventPattern {
+	// By default, assume that nothing gets anything.
+	p := map[string]knconf.EventPattern{
+		"chdlq": {
+			Success:  []bool{},
+			Interval: []uint{},
+		},
+	}
+
+	chdlq := false
+
+	for _, sub := range subs {
+		p[sub.subName()] = knconf.EventPattern{
+			Success:  []bool{},
+			Interval: []uint{},
+		}
+		p[sub.replyName()] = knconf.EventPattern{
+			Success:  []bool{},
+			Interval: []uint{},
+		}
+		p[sub.dlqName()] = knconf.EventPattern{
+			Success:  []bool{},
+			Interval: []uint{},
+		}
+
+		skipReply := false
+		attempts := knconf.DeliveryAttempts(sub.delivery, chDS)
+		// For subscriber.
+		if sub.hasSub {
+			p[sub.subName()] = knconf.PatternFromEstimates(attempts, sub.subFailCount)
+			if attempts <= sub.subFailCount {
+				skipReply = true
+				if sub.delivery != nil && sub.delivery.DeadLetterSink != nil {
+					p[sub.dlqName()] = knconf.PatternFromEstimates(1, 0)
+				} else {
+					chdlq = true
+				}
+			}
+			if !sub.subReplies {
+				skipReply = true
+			}
+		}
+		// For reply.
+		if !skipReply && sub.hasReply {
+			p[sub.replyName()] = knconf.PatternFromEstimates(attempts, sub.replyFailCount)
+			if attempts <= sub.replyFailCount {
+				if sub.delivery != nil && sub.delivery.DeadLetterSink != nil {
+					p[sub.dlqName()] = knconf.PatternFromEstimates(1, 0)
+				} else {
+					chdlq = true
+				}
+			}
+		}
+	}
+	if chdlq && chDS != nil && chDS.DeadLetterSink != nil {
+		p["chdlq"] = knconf.PatternFromEstimates(1, 0)
+	}
+
+	return p
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/account_role/role.go b/vendor/knative.dev/eventing/test/rekt/resources/account_role/role.go
new file mode 100644
index 000000000..aac3e7165
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/account_role/role.go
@@ -0,0 +1,89 @@
+/*
+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 account_role
+
+import (
+	"context"
+	"embed"
+	"fmt"
+
+	rbacv1 "k8s.io/api/rbac/v1"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+// Install will create a channelable-manipulator bound service account,
+// augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name": name,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func WithRole(role string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["role"] = role
+	}
+}
+
+func WithRules(rules ...rbacv1.PolicyRule) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["rules"]; !set {
+			cfg["rules"] = []map[string]interface{}{}
+		}
+
+		for _, resource := range rules {
+			elem := map[string]interface{}{
+				"apiGroups": resource.APIGroups,
+				"resources": resource.Resources,
+				"verbs":     resource.Verbs,
+				// TODO: skip for now
+				//"resourceNames":   resource.ResourceNames,
+				//"nonResourceURLs": resource.NonResourceURLs,
+			}
+			cfg["rules"] = append(cfg["rules"].([]map[string]interface{}), elem)
+		}
+	}
+}
+
+func WithRoleMatchLabel(matchLabel string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["matchLabel"] = matchLabel
+	}
+}
+
+func AsChannelableManipulator(cfg map[string]interface{}) {
+	WithRole(fmt.Sprintf("channelable-manipulator-collector-%s", cfg["name"]))(cfg)
+	WithRoleMatchLabel("duck.knative.dev/channelable")(cfg)
+}
+
+func AsAddressableResolver(cfg map[string]interface{}) {
+	WithRole(fmt.Sprintf("addressable-resolver-collector-%s", cfg["name"]))(cfg)
+	WithRoleMatchLabel("duck.knative.dev/addressable")(cfg)
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/account_role/role.yaml b/vendor/knative.dev/eventing/test/rekt/resources/account_role/role.yaml
new file mode 100644
index 000000000..91afaedc7
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/account_role/role.yaml
@@ -0,0 +1,73 @@
+# 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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+
+{{- /* only print the ClusterRole if there are any rules or a matchLabel */ -}}
+{{ if or .matchLabel .rules }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ .role }}
+
+{{ if .matchLabel }}
+aggregationRule:
+  clusterRoleSelectors:
+    - matchLabels:
+        {{ .matchLabel }}: "true"
+{{ end }}
+
+{{ if .rules }}
+rules:
+  {{ range $_, $rule := .rules }}
+  - apiGroups:
+    {{ range $_, $apiGroup := $rule.apiGroups }}
+      - {{ $apiGroup }}
+    {{ end }}
+    resources:
+    {{ range $_, $resource := $rule.resources }}
+      - {{ $resource }}
+    {{ end }}
+    verbs:
+    {{ range $_, $verb := $rule.verbs }}
+      - {{ $verb }}
+    {{ end }}
+{{ end }}
+
+{{- /* If no rules defined, print empty rules array */ -}}
+{{ else }}
+rules: []
+{{ end }}
+
+{{ end }}
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: {{ .name }}
+subjects:
+  - kind: ServiceAccount
+    name: {{ .name }}
+    namespace: {{ .namespace }}
+roleRef:
+  kind: ClusterRole
+  name: {{ .role }}
+  apiGroup: rbac.authorization.k8s.io
+
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/channel/channel.go b/vendor/knative.dev/eventing/test/rekt/resources/channel/channel.go
new file mode 100644
index 000000000..a63343a64
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/channel/channel.go
@@ -0,0 +1,83 @@
+/*
+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 channel
+
+import (
+	"context"
+	"embed"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/eventing/test/rekt/resources/addressable"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func GVR() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1", Resource: "channels"}
+}
+
+func GVK() schema.GroupVersionKind {
+	return schema.GroupVersionKind{Group: "messaging.knative.dev", Version: "v1", Kind: "Channel"}
+}
+
+// Install will create a Channel resource, augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name": name,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// IsReady tests to see if a Channel becomes ready within the time given.
+func IsReady(name string, timing ...time.Duration) feature.StepFn {
+	return k8s.IsReady(GVR(), name, timing...)
+}
+
+// IsAddressable tests to see if a Channel becomes addressable within the  time
+// given.
+func IsAddressable(name string, timing ...time.Duration) feature.StepFn {
+	return k8s.IsAddressable(GVR(), name, timing...)
+}
+
+// Address returns a Channel's address.
+func Address(ctx context.Context, name string, timings ...time.Duration) (*apis.URL, error) {
+	return addressable.Address(ctx, GVR(), name, timings...)
+}
+
+// AsRef returns a KRef for a Channel without namespace.
+func AsRef(name string) *duckv1.KReference {
+	apiVersion, kind := GVK().ToAPIVersionAndKind()
+	return &duckv1.KReference{
+		Kind:       kind,
+		APIVersion: apiVersion,
+		Name:       name,
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/channel/channel.yaml b/vendor/knative.dev/eventing/test/rekt/resources/channel/channel.yaml
new file mode 100644
index 000000000..f27d91318
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/channel/channel.yaml
@@ -0,0 +1,56 @@
+# 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.
+
+apiVersion: messaging.knative.dev/v1
+kind: Channel
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+spec:
+  {{ if .channelTemplate }}
+  channelTemplate:
+    apiVersion: {{ .channelTemplate.apiVersion }}
+    kind: {{ .channelTemplate.kind }}
+    {{ if .channelTemplate.spec }}
+    spec:
+      {{ range $key, $value := .channelTemplate.spec }}
+      {{ $key }}: {{ $value }}
+      {{ end }}
+    {{ end }}
+  {{ end }}
+  {{ if .delivery }}
+  delivery:
+    {{ if .delivery.deadLetterSink }}
+    deadLetterSink:
+      {{ if .delivery.deadLetterSink.ref }}
+      ref:
+        kind: {{ .delivery.deadLetterSink.ref.kind }}
+        namespace: {{ .namespace }}
+        name: {{ .delivery.deadLetterSink.ref.name }}
+        apiVersion: {{ .delivery.deadLetterSink.ref.apiVersion }}
+      {{ end }}
+      {{ if .delivery.deadLetterSink.uri }}
+      uri: {{ .delivery.deadLetterSink.uri }}
+      {{ end }}
+    {{ end }}
+    {{ if .delivery.retry }}
+    retry: {{ .delivery.retry}}
+    {{ end }}
+    {{ if .delivery.backoffPolicy }}
+    backoffPolicy: {{ .delivery.backoffPolicy}}
+    {{ end }}
+    {{ if .delivery.backoffDelay }}
+    backoffDelay: "{{ .delivery.backoffDelay }}"
+    {{ end }}
+  {{ end }}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go b/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go
new file mode 100644
index 000000000..5efeaecb1
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go
@@ -0,0 +1,108 @@
+/*
+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 channel_impl
+
+import (
+	"context"
+	"embed"
+	"log"
+	"time"
+
+	"github.com/kelseyhightower/envconfig"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/eventing/test/rekt/resources/addressable"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func GVR() schema.GroupVersionResource {
+	gvr, _ := meta.UnsafeGuessKindToResource(GVK())
+	return gvr
+}
+
+func GVK() schema.GroupVersionKind {
+	return schema.ParseGroupKind(EnvCfg.ChannelGK).WithVersion(EnvCfg.ChannelV)
+}
+
+var EnvCfg EnvConfig
+
+type EnvConfig struct {
+	ChannelGK string `envconfig:"CHANNEL_GROUP_KIND" default:"InMemoryChannel.messaging.knative.dev" required:"true"`
+	ChannelV  string `envconfig:"CHANNEL_VERSION" default:"v1" required:"true"`
+}
+
+func init() {
+	// Process EventingGlobal.
+	if err := envconfig.Process("", &EnvCfg); err != nil {
+		log.Fatal("Failed to process env var", err)
+	}
+}
+
+// Install will create a Channel resource, augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	apiVersion, kind := GVK().ToAPIVersionAndKind()
+	cfg := map[string]interface{}{
+		"name":       name,
+		"kind":       kind,
+		"apiVersion": apiVersion,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// IsReady tests to see if a Channel becomes ready within the time given.
+func IsReady(name string, timing ...time.Duration) feature.StepFn {
+	return k8s.IsReady(GVR(), name, timing...)
+}
+
+// IsAddressable tests to see if a Channel becomes addressable within the  time
+// given.
+func IsAddressable(name string, timing ...time.Duration) feature.StepFn {
+	return k8s.IsAddressable(GVR(), name, timing...)
+}
+
+// Address returns a Channel's address.
+func Address(ctx context.Context, name string, timings ...time.Duration) (*apis.URL, error) {
+	return addressable.Address(ctx, GVR(), name, timings...)
+}
+
+// AsRef returns a KRef for a Channel without namespace.
+func AsRef(name string) *duckv1.KReference {
+	apiVersion, kind := GVK().ToAPIVersionAndKind()
+	return &duckv1.KReference{
+		Kind:       kind,
+		APIVersion: apiVersion,
+		Name:       name,
+	}
+}
+
+// WithDeadLetterSink adds the dead letter sink related config to a Subscription spec.
+var WithDeadLetterSink = delivery.WithDeadLetterSink
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.yaml b/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.yaml
new file mode 100644
index 000000000..34eb667ca
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.yaml
@@ -0,0 +1,45 @@
+# 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.
+
+apiVersion: {{ .apiVersion}}
+kind: {{ .kind }}
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+spec:
+  {{ if .delivery }}
+  delivery:
+    {{ if .delivery.deadLetterSink }}
+    deadLetterSink:
+      {{ if .delivery.deadLetterSink.ref }}
+      ref:
+        kind: {{ .delivery.deadLetterSink.ref.kind }}
+        namespace: {{ .namespace }}
+        name: {{ .delivery.deadLetterSink.ref.name }}
+        apiVersion: {{ .delivery.deadLetterSink.ref.apiVersion }}
+      {{ end }}
+      {{ if .delivery.deadLetterSink.uri }}
+      uri: {{ .delivery.deadLetterSink.uri }}
+      {{ end }}
+    {{ end }}
+    {{ if .delivery.retry }}
+    retry: {{ .delivery.retry}}
+    {{ end }}
+    {{ if .delivery.backoffPolicy }}
+    backoffPolicy: {{ .delivery.backoffPolicy}}
+    {{ end }}
+    {{ if .delivery.backoffDelay }}
+    backoffDelay: "{{ .delivery.backoffDelay }}"
+    {{ end }}
+  {{ end }}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go
new file mode 100644
index 000000000..747fe4bee
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.go
@@ -0,0 +1,81 @@
+/*
+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 containersource
+
+import (
+	"context"
+	"embed"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func init() {
+	environment.RegisterPackage(manifest.ImagesFromFS(yaml)...)
+}
+
+func Gvr() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1", Resource: "containersources"}
+}
+
+// IsReady tests to see if a ContainerSource becomes ready within the time given.
+func IsReady(name string, timing ...time.Duration) feature.StepFn {
+	return k8s.IsReady(Gvr(), name, timing...)
+}
+
+// Install will create a ContainerSource resource, augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name": name,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// WithExtensions adds the ceOverrides related config to a ContainerSource spec.
+func WithExtensions(extensions map[string]interface{}) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["ceOverrides"]; !set {
+			cfg["ceOverrides"] = map[string]interface{}{}
+		}
+		ceOverrides := cfg["ceOverrides"].(map[string]interface{})
+
+		if extensions != nil {
+			if _, set := ceOverrides["extensions"]; !set {
+				ceOverrides["extensions"] = map[string]interface{}{}
+			}
+			ceExt := ceOverrides["extensions"].(map[string]interface{})
+			for k, v := range extensions {
+				ceExt[k] = v
+			}
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml
new file mode 100644
index 000000000..f480b459b
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml
@@ -0,0 +1,52 @@
+# 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.
+
+apiVersion: sources.knative.dev/v1
+kind: ContainerSource
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+spec:
+  {{ if .ceOverrides }}
+  ceOverrides:
+    {{ if .ceOverrides.extensions }}
+    extensions:
+      {{ range $key, $value := .ceOverrides.extensions }}
+      {{ $key }}: {{ $value }}
+      {{ end}}
+    {{ end }}
+  {{ end }}
+  sink:
+    {{ if .sink.ref }}
+    ref:
+      kind: {{ .sink.ref.kind }}
+      namespace: {{ .namespace }}
+      name: {{ .sink.ref.name }}
+      apiVersion: {{ .sink.ref.apiVersion }}
+    {{ end }}
+    {{ if .sink.uri }}
+    uri: {{ .sink.uri }}
+    {{ end }}
+  template:
+    spec:
+      containers:
+      - name: heartbeats
+        image: ko://knative.dev/eventing/test/test_images/heartbeats
+        args:
+        - --period=1
+        env:
+        - name: POD_NAME
+          value: heartbeats
+        - name: POD_NAMESPACE
+          value: {{ .namespace }}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/doc.go b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/doc.go
new file mode 100644
index 000000000..0a7b6a8ab
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/doc.go
@@ -0,0 +1,21 @@
+//go:build cloudevents
+// +build cloudevents
+
+/*
+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 events is here to allow us to vendor the .ce files here.
+package events
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/three.ce b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/three.ce
new file mode 100644
index 000000000..49eaa290d
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/three.ce
@@ -0,0 +1,23 @@
+ContextAttributes:
+  specversion: "1.0"
+  type: a.type
+  id: abc-123-1
+  source: foo/bar
+  datacontenttype: application/json
+Data: '{"hello":"world"}'
+---
+ContextAttributes:
+  specversion: "1.0"
+  type: a.type
+  id: abc-123-2
+  source: foo/bar
+  datacontenttype: application/json
+Data: '{"hello":"world"}'
+---
+ContextAttributes:
+  specversion: "1.0"
+  type: a.type
+  id: abc-123-3
+  source: foo/bar
+  datacontenttype: application/json
+Data: '{"hello":"world"}'
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go b/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go
new file mode 100644
index 000000000..a102ff762
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.go
@@ -0,0 +1,84 @@
+/*
+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 pingsource
+
+import (
+	"context"
+	"embed"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/eventing/test/rekt/resources/source"
+	"knative.dev/reconciler-test/pkg/k8s"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func Gvr() schema.GroupVersionResource {
+	return schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1", Resource: "pingsources"}
+}
+
+// Install will create a Broker resource, augmented with the config fn options.
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name": name,
+	}
+	for _, fn := range opts {
+		fn(cfg)
+	}
+	return func(ctx context.Context, t feature.T) {
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err, cfg)
+		}
+	}
+}
+
+// IsReady tests to see if a PingSource becomes ready within the time given.
+func IsReady(name string, timings ...time.Duration) feature.StepFn {
+	return k8s.IsReady(Gvr(), name, timings...)
+}
+
+// WithSink adds the sink related config to a PingSource spec.
+var WithSink = source.WithSink
+
+// WithData adds the contentType and data config to a PingSource spec.
+func WithData(contentType, data string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if contentType != "" {
+			cfg["contentType"] = contentType
+		}
+		if data != "" {
+			cfg["data"] = data
+		}
+	}
+}
+
+// WithDataBase64 adds the contentType and dataBase64 config to a PingSource spec.
+func WithDataBase64(contentType, dataBase64 string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if contentType != "" {
+			cfg["contentType"] = contentType
+		}
+		if dataBase64 != "" {
+			cfg["dataBase64"] = dataBase64
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.yaml b/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.yaml
new file mode 100644
index 000000000..8a95879fb
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/pingsource/pingsource.yaml
@@ -0,0 +1,45 @@
+# 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.
+
+apiVersion: sources.knative.dev/v1
+kind: PingSource
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+spec:
+  {{ if .schedule }}
+  schedule: '{{ .schedule }}'
+  {{ end }}
+  {{ if .contentType }}
+  contentType: '{{ .contentType }}'
+  {{ end }}
+  {{ if .data }}
+  data: '{{ .data }}'
+  {{ end }}
+  {{ if .dataBase64 }}
+  dataBase64: '{{ .dataBase64 }}'
+  {{ end }}
+  {{if .sink }}
+  sink:
+    {{ if .sink.ref }}
+    ref:
+      kind: {{ .sink.ref.kind }}
+      namespace: {{ .namespace }}
+      name: {{ .sink.ref.name }}
+      apiVersion: {{ .sink.ref.apiVersion }}
+    {{ end }}
+    {{ if .sink.uri }}
+    uri: {{ .sink.uri }}
+    {{ end }}
+  {{ end }}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/source/source.go b/vendor/knative.dev/eventing/test/rekt/resources/source/source.go
new file mode 100644
index 000000000..94c90de08
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/resources/source/source.go
@@ -0,0 +1,46 @@
+/*
+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 source
+
+import (
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+// WithSink adds the sink related config to a duckv1.SourceSpec.
+func WithSink(ref *duckv1.KReference, uri string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["sink"]; !set {
+			cfg["sink"] = map[string]interface{}{}
+		}
+		sink := cfg["sink"].(map[string]interface{})
+
+		if uri != "" {
+			sink["uri"] = uri
+		}
+		if ref != nil {
+			if _, set := sink["ref"]; !set {
+				sink["ref"] = map[string]interface{}{}
+			}
+			sref := sink["ref"].(map[string]interface{})
+			sref["apiVersion"] = ref.APIVersion
+			sref["kind"] = ref.Kind
+			// skip namespace
+			sref["name"] = ref.Name
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/doc.go b/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/doc.go
new file mode 100644
index 000000000..0a7b6a8ab
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/doc.go
@@ -0,0 +1,21 @@
+//go:build cloudevents
+// +build cloudevents
+
+/*
+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 events is here to allow us to vendor the .ce files here.
+package events
diff --git a/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/three.ce b/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/three.ce
new file mode 100644
index 000000000..49eaa290d
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/three.ce
@@ -0,0 +1,23 @@
+ContextAttributes:
+  specversion: "1.0"
+  type: a.type
+  id: abc-123-1
+  source: foo/bar
+  datacontenttype: application/json
+Data: '{"hello":"world"}'
+---
+ContextAttributes:
+  specversion: "1.0"
+  type: a.type
+  id: abc-123-2
+  source: foo/bar
+  datacontenttype: application/json
+Data: '{"hello":"world"}'
+---
+ContextAttributes:
+  specversion: "1.0"
+  type: a.type
+  id: abc-123-3
+  source: foo/bar
+  datacontenttype: application/json
+Data: '{"hello":"world"}'