Repository: knative-sandbox/eventing-kafka-broker
Pull request #1731 (2022-01-12 15:32:49)

Run rekt tests from knative-eventing repo with KafkaChannel
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Run rekt tests from knative-eventing repo with KafkaChannel: https://github.com/knative/eventing/blob/68d441013ca8c65c5b9a7f97d8f8be7e922d8a13/test/rekt/channel_test.go
- Except 2 non-compliant tests (comments inline)

Created separate commits for easier review

<!--
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 **APPROVED**

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

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

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

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

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

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

> :exclamation: Current head 3e0ba73 differs from pull request most recent head a401f89. Consider uploading reports for the commit a401f89 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731/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/1731?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    #1731      +/-   ##
============================================
+ Coverage     72.61%   72.67%   +0.06%     
+ Complexity      606      604       -2     
============================================
  Files           124      124              
  Lines          4604     4604              
  Branches        176      176              
============================================
+ Hits           3343     3346       +3     
+ Misses          984      983       -1     
+ Partials        277      275       -2     
```

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

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `87.50% <0.00%> (-6.25%)` | :arrow_down: |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | `100.00% <0.00%> (+4.76%)` | :arrow_up: |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `75.00% <0.00%> (+15.00%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731?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/1731?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [a9400eb...a401f89](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1731?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: @aliok: 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-upgrade-tests | a401f89d51bfea519338004f544dfbe81e7b15ae | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1481288467559223296) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-upgrade-tests | a401f89d51bfea519338004f544dfbe81e7b15ae | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1481288467559223296) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484417167460929536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-build-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484417167460929536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1484417167578370048) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-build-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484417167460929536) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1484417167578370048) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484417167515455488) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1484417167578370048) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 34cf88eb845a007936edbbf91b9f6d23639c135b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484417167515455488) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484431365934944256) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-build-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484431365934944256) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484431365993664512) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-integration-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484431365993664512) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a2d21f77f1e93daf258684778cb0f20c5d8706a9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484450426928500736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-integration-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484431365993664512) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a2d21f77f1e93daf258684778cb0f20c5d8706a9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484450426928500736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5c5974353532da11b1c63a49952533145ef1e9ce | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1484452793749082112) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-integration-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484431365993664512) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a2d21f77f1e93daf258684778cb0f20c5d8706a9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484450426928500736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5c5974353532da11b1c63a49952533145ef1e9ce | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1484452793749082112) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5c5974353532da11b1c63a49952533145ef1e9ce | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1484452793631641600) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-integration-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484431365993664512) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a2d21f77f1e93daf258684778cb0f20c5d8706a9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484450426928500736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5c5974353532da11b1c63a49952533145ef1e9ce | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1484452793631641600) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 492689a647489682e78961c165a3b74b4a13c259 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1484453937951019008) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: The following tests **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-integration-tests | ed124e5802711661c9011293b50de8d0789b5747 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1484431365993664512) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a2d21f77f1e93daf258684778cb0f20c5d8706a9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-build-tests/1484450426928500736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 492689a647489682e78961c165a3b74b4a13c259 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1484453937951019008) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 492689a647489682e78961c165a3b74b4a13c259 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1484453937825189888) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @aliok: 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-upgrade-tests | d6ba7e04d89cce19fe7762437c34aafeab9da2b0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1731/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1484487463391989760) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`

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

<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 -->
aliok: /test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests

Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index 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/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/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/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/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index e6f9d1e9a..e7db0e0eb 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -21,8 +21,11 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
@@ -81,6 +84,24 @@ func NewConsumerSpec(opts ...ConsumerSpecOption) kafkainternals.ConsumerSpec {
 	return *spec
 }
 
+func NewConsumerSpecDelivery(order internals.DeliveryOrdering) *kafkainternals.DeliverySpec {
+	return &kafkainternals.DeliverySpec{
+		Ordering: order,
+	}
+}
+
+func NewConsumerSpecAuth() *kafkainternals.Auth {
+	return &kafkainternals.Auth{
+		NetSpec: &v1beta1.KafkaNetSpec{},
+	}
+}
+
+func NewConsumerSpecFilters() *kafkainternals.Filters {
+	return &kafkainternals.Filters{
+		Filter: &v1.TriggerFilter{},
+	}
+}
+
 func ConsumerSpec(spec kafkainternals.ConsumerSpec) ConsumerOption {
 	return func(cg *kafkainternals.Consumer) {
 		cg.Spec = spec
@@ -130,3 +151,33 @@ func ConsumerVReplicas(vreplicas int32) ConsumerSpecOption {
 		c.VReplicas = &vreplicas
 	}
 }
+
+func ConsumerAuth(auth *kafkainternals.Auth) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Auth = auth
+	}
+}
+
+func ConsumerDelivery(delivery *kafkainternals.DeliverySpec) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Delivery = delivery
+	}
+}
+
+func ConsumerSubscriber(dest duckv1.Destination) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Subscriber = dest
+	}
+}
+
+func ConsumerCloudEventOverrides(ce *duckv1.CloudEventOverrides) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.CloudEventOverrides = ce
+	}
+}
+
+func ConsumerFilters(filters *kafkainternals.Filters) ConsumerSpecOption {
+	return func(c *kafkainternals.ConsumerSpec) {
+		c.Filters = filters
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index a62463d2a..72f40c1be 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -17,8 +17,10 @@
 package testing
 
 import (
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
@@ -68,6 +70,45 @@ func NewConsumerGroup(opts ...ConsumerGroupOption) *kafkainternals.ConsumerGroup
 	return cg
 }
 
+func ConsumerGroupReady(cg *kafkainternals.ConsumerGroup) {
+	cg.Status.Conditions = duckv1.Conditions{
+		{
+			Type:   apis.ConditionReady,
+			Status: corev1.ConditionTrue,
+		},
+	}
+}
+
+func WithConsumerGroupFailed(reason string, msg string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.GetConditionSet().Manage(cg.GetStatus()).MarkFalse(kafkainternals.ConditionConsumerGroupConsumers, reason, msg)
+	}
+}
+
+func WithConsumerGroupName(name string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.ObjectMeta.Name = name
+	}
+}
+
+func WithConsumerGroupNamespace(namespace string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.ObjectMeta.Namespace = namespace
+	}
+}
+
+func WithConsumerGroupOwnerRef(ownerref *metav1.OwnerReference) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.ObjectMeta.OwnerReferences = []metav1.OwnerReference{*ownerref}
+	}
+}
+
+func WithConsumerGroupLabels(labels map[string]string) ConsumerGroupOption {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		cg.Spec.Template.Labels = labels
+	}
+}
+
 func ConsumerGroupReplicas(replicas int32) ConsumerGroupOption {
 	return func(cg *kafkainternals.ConsumerGroup) {
 		cg.Spec.Replicas = pointer.Int32Ptr(replicas)
@@ -85,3 +126,13 @@ func ConsumerGroupConsumerSpec(spec kafkainternals.ConsumerSpec) ConsumerGroupOp
 		cg.Spec.Template.Spec = spec
 	}
 }
+
+func WithDeadLetterSinkURI(uri string) func(cg *kafkainternals.ConsumerGroup) {
+	return func(cg *kafkainternals.ConsumerGroup) {
+		u, err := apis.ParseURL(uri)
+		if err != nil {
+			panic(err)
+		}
+		cg.Status.DeadLetterSinkURI = u
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 17feb9077..04ce2639f 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -135,8 +135,13 @@ func InitSourceConditions(obj duckv1.KRShaped) {
 func StatusSourceSinkResolved(uri string) KRShapedOption {
 	return func(obj duckv1.KRShaped) {
 		ks := obj.(*sources.KafkaSource)
-		ks.Status.SinkURI, _ = apis.ParseURL(uri)
-		ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(sources.KafkaConditionSinkProvided)
+		res, _ := apis.ParseURL(uri)
+		ks.Status.SinkURI = res
+		if !res.IsEmpty() {
+			ks.GetConditionSet().Manage(ks.GetStatus()).MarkTrue(sources.KafkaConditionSinkProvided)
+		} else {
+			ks.GetConditionSet().Manage(ks.GetStatus()).MarkUnknown(sources.KafkaConditionSinkProvided, "SinkEmpty", "Sink has resolved to empty.%s", "")
+		}
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
index af1533027..455ac194c 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2_test.go
@@ -31,10 +31,10 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 )
 
-func TestNewControllerV2(t *testing.T) {
+func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	controller := NewControllerV2(ctx, &config.Env{})
+	controller := NewController(ctx, &config.Env{})
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
 	}
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
new file mode 100644
index 000000000..6e6fe7e24
--- /dev/null
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -0,0 +1,588 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v2
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgotesting "k8s.io/client-go/testing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
+	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+
+	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+)
+
+const (
+	// name of the trigger under test
+	triggerName = "test-trigger"
+	// namespace of the trigger under test
+	triggerNamespace = "test-namespace"
+)
+
+var DefaultEnv = &config.Env{
+	DataPlaneConfigMapNamespace: "knative-eventing",
+	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+	GeneralConfigMapName:        "kafka-broker-config",
+	SystemNamespace:             "knative-eventing",
+}
+
+var (
+	url = &apis.URL{
+		Scheme: "http",
+		Host:   "localhost",
+		Path:   "/path",
+	}
+)
+
+func TestReconcileKind(t *testing.T) {
+
+	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with ordered delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered))),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Ordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered)),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with unordered delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered))),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered)),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with invalid delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, "invalid")),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(
+						"invalid annotation %s value: invalid. Allowed values [ \"ordered\" | \"unordered\" ]",
+						deliveryOrderAnnotation,
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerDependencyFailed("failed to reconcile consumer group", "invalid annotation kafka.eventing.knative.dev/delivery.order value: invalid. Allowed values [ \"ordered\" | \"unordered\" ]"),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, "invalid"),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupReady,
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(TriggerUUID),
+						WithConsumerGroupNamespace(triggerNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(TriggerUUID),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+							ConsumerFilters(NewConsumerSpecFilters()),
+						)),
+						ConsumerGroupReady,
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyReady(),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg with update but not ready",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+				),
+			},
+			Key:         testKey,
+			WantCreates: []runtime.Object{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewConsumerGroup(
+						WithConsumerGroupName(TriggerUUID),
+						WithConsumerGroupNamespace(triggerNamespace),
+						WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+						WithConsumerGroupLabels(nil),
+						ConsumerGroupConsumerSpec(NewConsumerSpec(
+							ConsumerTopics(),
+							ConsumerConfigs(
+								ConsumerGroupIdConfig(TriggerUUID),
+							),
+							ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+							ConsumerFilters(NewConsumerSpecFilters()),
+						)),
+					),
+				},
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+					ConsumerGroupReady,
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyReady(),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg without update but not ready",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - existing cg but failed",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+					WithConsumerGroupFailed("failed", "failed"),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyFailed("failed", "failed"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal with dead letter sink uri",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(),
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupLabels(nil),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(internals.Unordered)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+					)),
+					WithDeadLetterSinkURI(url.String()),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						withDeadLetterSinkURI(url.String()),
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker not found",
+			Objects: []runtime.Object{
+				newTrigger(),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(
+						"failed to get broker: brokers.eventing.knative.dev \"%s\" not found",
+						BrokerName,
+					),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker not ready",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewBroker(
+					func(v *eventing.Broker) { v.Status.InitializeConditions() },
+					StatusBrokerConfigNotParsed("wrong"),
+				),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+		{
+			Name: "Broker deleted",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewDeletedBroker(),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerNotConfigured(),
+					),
+				},
+			},
+		},
+		{
+			Name: "Don't reconcile trigger associated with a broker with a different broker class",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewBroker(func(b *eventing.Broker) {
+					b.Annotations = map[string]string{
+						eventing.BrokerClassAnnotationKey: "MTChannelBasedBroker",
+					}
+				}),
+			},
+			Key: testKey,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+					),
+				},
+			},
+		},
+	}
+
+	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		logger := logging.FromContext(ctx)
+
+		reconciler := &Reconciler{
+			BrokerLister:        listers.GetBrokerLister(),
+			EventingClient:      eventingclient.Get(ctx),
+			Env:                 env,
+			ConsumerGroupLister: listers.GetConsumerGroupLister(),
+			InternalsClient:     fakeconsumergroupinformer.Get(ctx),
+		}
+
+		return triggerreconciler.NewReconciler(
+			ctx,
+			logger,
+			eventingclient.Get(ctx),
+			listers.GetTriggerLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+	}))
+}
+
+func newTrigger(options ...reconcilertesting.TriggerOption) *eventing.Trigger {
+	return reconcilertesting.NewTrigger(
+		triggerName,
+		triggerNamespace,
+		BrokerName,
+		append(
+			options,
+			//reconcilertesting.WithTriggerSubscriberURI(ServiceURL),
+			func(t *eventing.Trigger) {
+				t.UID = TriggerUUID
+			},
+		)...,
+	)
+}
+
+func withDeadLetterSinkURI(uri string) func(trigger *eventing.Trigger) {
+	return func(trigger *eventing.Trigger) {
+		u, err := apis.ParseURL(uri)
+		if err != nil {
+			panic(err)
+		}
+		trigger.Status.DeadLetterSinkURI = u
+		trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+	}
+}
+
+func withTriggerSubscriberResolvedSucceeded() func(*eventing.Trigger) {
+	return func(t *eventing.Trigger) {
+		t.GetConditionSet().Manage(&t.Status).MarkTrue(
+			eventing.TriggerConditionSubscriberResolved,
+		)
+	}
+}
diff --git a/test/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/test/upgrade/continual/verification.go b/test/upgrade/continual/verification.go
index 02a206071..49f63597e 100644
--- a/test/upgrade/continual/verification.go
+++ b/test/upgrade/continual/verification.go
@@ -18,7 +18,6 @@ package continual
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/kelseyhightower/envconfig"
 	eventingkafkaupgrade "knative.dev/eventing-kafka/test/upgrade/continual"
@@ -78,7 +77,6 @@ func configurator(theSut sut.SystemUnderTest, configTemplate string) prober.Conf
 		// TODO: knative/eventing#5176 - this is cumbersome
 		config.ConfigTemplate = fmt.Sprintf("../../../../../../%s",
 			configTemplate)
-		config.FinishedSleep = 5 * time.Minute
 		// envconfig.Process invocation is repeated from within prober.NewConfig to
 		// make sure every knob is configurable, but using defaults from Eventing
 		// Kafka instead of Core. The prefix is also changed.
diff --git a/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"}'