Repository: knative-sandbox/eventing-kafka-broker
Pull request #701 (2021-03-05 12:54:42)

Implemented delivery order label
Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

Part of #589

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- :gift: Implemented delivery order label
- :bug: Modified default value for delivery order in contract
- :wastebasket: Removed temporarily the usage of upstream features, since they're unstable and causes update issues


<!--
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
Now eventing-kafka-broker implements ordered delivery. You can choose between ordered and unordered delivery through the label kafka.eventing.knative.dev/delivery.order in your triggers. Check out the docs for more details
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

Coming soon in another PR to docs repo

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **slinkydeveloper** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @slinkydeveloper` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["slinkydeveloper"]} -->
slinkydeveloper: I'm going to implement a basic e2e test, although I'm not sure how far I can go with testing this feature in a e2e manner
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/701?src=pr&el=h1) Report
> Merging [#701](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/701?src=pr&el=desc) (5520e4b) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/d8f3e16803ba2bf62e43bc18ba551564f17ebe95?el=desc) (d8f3e16) will **decrease** coverage by `0.02%`.
> The diff coverage is `69.23%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/701/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/701?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             main     #701      +/-   ##
==========================================
- Coverage   71.65%   71.63%   -0.03%     
==========================================
  Files          28       28              
  Lines        1242     1255      +13     
==========================================
+ Hits          890      899       +9     
- Misses        274      277       +3     
- Partials       78       79       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/701?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/trigger/trigger.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/701/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXIuZ28=) | `71.01% <69.23%> (-0.19%)` | :arrow_down: |

------

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

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1367820879706198016/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 84d10343d80abced5584aa3113f89c13f72f3713 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1367820879681032192) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<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 -->
slinkydeveloper: /retest
slinkydeveloper: Because I would love to use reconciler-test for the e2e, I've opened this one https://github.com/knative-sandbox/eventing-kafka-broker/pull/703
pierDipi: /this-is-fine :heart_eyes: 
pierDipi: If you want we can have a status condition to indicate the `DeliveryOrder` configured for a Trigger since it's easy to make a typo and then the `DeliveryOrder` configured is the default one rather than the one users wanted.
pierDipi: /kind enhancement
slinkydeveloper: > If you want we can have a status condition to indicate the DeliveryOrder configured for a Trigger since it's easy to make a typo and then the DeliveryOrder configured is the default one rather than the one users wanted. (A separate PR is fine too)

I don't see the point TBH, there's no need to tell the user the controller read the contract properly IMO
pierDipi: I mean an annotation is weak compared to a key in the spec because we don't have a way to validate that the key is correct for that reason a status condition would be nice to have IMO.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **slinkydeveloper** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["slinkydeveloper"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372192390709776384/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

slinkydeveloper: @pierDipi can we tackle the status issue in another PR? From my understanding (I might be wrong here) we need to setup some alternate condition set, like done in receiver_condition_test.go, right? If that's the case, i think it deserves a separate PR
knative-prow-robot: @slinkydeveloper: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | ee02f6041fa13c0fd5e4999eee421ad79b8fed0d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372192390688804864) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372204758919548928/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372216469919633408/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 62c889e558577f1caff39fe2c9796c8dafd412af | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372216469894467584) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372231983383449600/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372238673684205568/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 393fe9e1ec45e001de404d45a95d96e53e9a34a5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1372238673638068224) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<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 -->
slinkydeveloper: /retest

Something wrong in maven?!
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/701#pullrequestreview-614619217" title="Approved">pierDipi</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @slinkydeveloper: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 393fe9e1ec45e001de404d45a95d96e53e9a34a5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372238673667428352) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<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: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["pierdipi"]} -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["pierdipi"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372553436310016000/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 2cf2ab4fd23149bcf775f647b2186f9f3fdd6640 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372553436284850176) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372566672828796928/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 0c5921e6fedb62741aeab2b6d54463997e2e8014 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372566672778465280) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372582787327987712/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372589521203367936/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372598800593457152/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 413116eca77239863370a4340f6a23a029a08cbe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372598800572485632) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372843377938141184/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-prow-robot: @slinkydeveloper: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 413116eca77239863370a4340f6a23a029a08cbe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372598800572485632) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 75612680810b1411ae5abc17dba0fb5aa98c209c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-build-tests/1372843377862643712) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<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 -->
slinkydeveloper: Seems like ordered delivery is not really ordered, will investigate
slinkydeveloper: /hold
slinkydeveloper: The reason why it's failing is veeeery weird. It doesn't add the delivery order in the json:

```
{
  "generation": "38",
  "resources": [
    {
      "uid": "87585205-e53a-442f-8e4b-90192cf2cc0b",
      "topics": [
        "knative-broker-test-oymfewtk-broker-rtbugxnl"
      ],
      "bootstrapServers": "my-cluster-kafka-bootstrap.kafka:9092",
      "ingress": {
        "path": "/test-oymfewtk/broker-rtbugxnl"
      },
      "egresses": [
        {
          "consumerGroup": "5d4fcef4-2551-4f2e-8f7b-88fdf4697015",
          "destination": "http://sink-ngsdgoqr.test-oymfewtk.svc.cluster.local/",
          "uid": "5d4fcef4-2551-4f2e-8f7b-88fdf4697015"
        }
      ]
    }
  ]
}
```
knative-prow-robot: @slinkydeveloper: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 75612680810b1411ae5abc17dba0fb5aa98c209c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372843377904586752) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 37604c13086a8e0eb2db52bf91eb4b0492d6a78b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-build-tests/1372931959793127424) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372931959860236288/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

slinkydeveloper: I've removed broker_test.go, since upstream movement around features is unstable now...

For me we're ready to go
/unhold
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372940288099618816/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-prow-robot: @slinkydeveloper: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 75612680810b1411ae5abc17dba0fb5aa98c209c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372843377904586752) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 64c978b844322a5a29bf6aee450c07e770517ada | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-build-tests/1372940288019927040) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<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 -->
slinkydeveloper: /hold

one sec, maybe it's better to use a label (that's what other people does :smile: )
slinkydeveloper: /unhold

Ok now we use labels
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372945458892115968/artifacts/line-cov.html#file0) | 85.3% | 85.0% | -0.3

knative-prow-robot: @slinkydeveloper: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 75612680810b1411ae5abc17dba0fb5aa98c209c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1372843377904586752) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 78a2c2d89d37757424117bd47fa3567085f7f565 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-build-tests/1372945458795646976) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1372949599592910848/artifacts/line-cov.html#file0) | 85.3% | 85.0% | -0.3

pierDipi: I see no benefits in using a label here, basically, the difference is that using a label I can query all objects with that label without doing a full scan or building a custom index whereas an annotation requires a full scan or a custom index.

For example, we use a label in our triggers to get fast access to all triggers associated with a broker. 
https://github.com/knative-sandbox/eventing-kafka-broker/blob/1258e973843974a5a7fc42aff651cab4ed2a3f56/control-plane/pkg/reconciler/trigger/controller.go#L141-L157

In this case, a query like "give me all ordered trigger" seems pretty useless.

Of course, the benefit of using a label comes with additional storage, memory, and compute costs, so my preference would be to use an annotation since this value is only useful in a single reconcile cycle.
slinkydeveloper: @pierDipi the reason i went for a label is to align with kafka source: https://knative.dev/docs/eventing/samples/kafka/source/#optional-specify-the-key-deserializer

But yeah i was checking in serving and they use annotations too for "special"/"experimental" features, so i'll revert and go for the annotation

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1373915424294440960/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/701/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1373918692978462720/artifacts/line-cov.html#file0) | 85.3% | 84.8% | -0.5


Patch:
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index e097d349c..36fe425ca 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -79,19 +79,19 @@ func (BackoffPolicy) EnumDescriptor() ([]byte, []int) {
 type DeliveryOrder int32
 
 const (
-	DeliveryOrder_ORDERED   DeliveryOrder = 0
-	DeliveryOrder_UNORDERED DeliveryOrder = 1
+	DeliveryOrder_UNORDERED DeliveryOrder = 0
+	DeliveryOrder_ORDERED   DeliveryOrder = 1
 )
 
 // Enum value maps for DeliveryOrder.
 var (
 	DeliveryOrder_name = map[int32]string{
-		0: "ORDERED",
-		1: "UNORDERED",
+		0: "UNORDERED",
+		1: "ORDERED",
 	}
 	DeliveryOrder_value = map[string]int32{
-		"ORDERED":   0,
-		"UNORDERED": 1,
+		"UNORDERED": 0,
+		"ORDERED":   1,
 	}
 )
 
@@ -417,7 +417,7 @@ func (x *Egress) GetDeliveryOrder() DeliveryOrder {
 	if x != nil {
 		return x.DeliveryOrder
 	}
-	return DeliveryOrder_ORDERED
+	return DeliveryOrder_UNORDERED
 }
 
 type isEgress_ReplyStrategy interface {
@@ -927,8 +927,8 @@ var file_proto_def_contract_proto_rawDesc = []byte{
 	0x0a, 0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12,
 	0x0f, 0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f, 0x6e, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x10, 0x00,
 	0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x10, 0x01, 0x2a, 0x2b, 0x0a, 0x0d,
-	0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x12, 0x0b, 0x0a,
-	0x07, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x55, 0x4e,
+	0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x12, 0x0d, 0x0a,
+	0x09, 0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07,
 	0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x29, 0x0a, 0x0b, 0x43, 0x6f, 0x6e,
 	0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x42, 0x49, 0x4e, 0x41,
 	0x52, 0x59, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x53, 0x54, 0x52, 0x55, 0x43, 0x54, 0x55, 0x52,
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index b381b03fa..90be5ae96 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -19,6 +19,7 @@ package trigger
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -40,6 +41,12 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
+const (
+	deliveryOrderAnnotation = "kafka.eventing.knative.dev/delivery.order"
+	deliveryOrderOrdered    = "ordered"
+	deliveryOrderUnordered  = "unordered"
+)
+
 type Reconciler struct {
 	*base.Reconciler
 
@@ -101,6 +108,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 
 	statusConditionManager.propagateBrokerCondition(broker)
 
+	if !broker.Status.IsReady() {
+		// Trigger will get re-queued once this broker is ready.
+		return nil
+	}
+
 	// Get data plane config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
@@ -279,6 +291,15 @@ func (r *Reconciler) getTriggerConfig(ctx context.Context, trigger *eventing.Tri
 	}
 	egress.EgressConfig = egressConfig
 
+	deliveryOrderAnnotationValue, ok := trigger.Annotations[deliveryOrderAnnotation]
+	if ok {
+		deliveryOrder, err := deliveryOrderFromString(deliveryOrderAnnotationValue)
+		if err != nil {
+			return nil, err
+		}
+		egress.DeliveryOrder = deliveryOrder
+	}
+
 	return egress, nil
 }
 
@@ -297,3 +318,14 @@ func isOurBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass
 }
+
+func deliveryOrderFromString(val string) (contract.DeliveryOrder, error) {
+	switch strings.ToLower(val) {
+	case deliveryOrderOrdered:
+		return contract.DeliveryOrder_ORDERED, nil
+	case deliveryOrderUnordered:
+		return contract.DeliveryOrder_UNORDERED, nil
+	default:
+		return contract.DeliveryOrder_UNORDERED, fmt.Errorf("invalid annotation %s value: %s. Allowed values [ %q | %q ]", deliveryOrderAnnotation, val, deliveryOrderOrdered, deliveryOrderUnordered)
+	}
+}
diff --git a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
index 4b5f6e7ae..656d110f8 100644
--- a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
+++ b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
@@ -152,24 +152,24 @@ public final class DataPlaneContract {
   public enum DeliveryOrder
       implements com.google.protobuf.ProtocolMessageEnum {
     /**
-     * <code>ORDERED = 0;</code>
+     * <code>UNORDERED = 0;</code>
      */
-    ORDERED(0),
+    UNORDERED(0),
     /**
-     * <code>UNORDERED = 1;</code>
+     * <code>ORDERED = 1;</code>
      */
-    UNORDERED(1),
+    ORDERED(1),
     UNRECOGNIZED(-1),
     ;
 
     /**
-     * <code>ORDERED = 0;</code>
+     * <code>UNORDERED = 0;</code>
      */
-    public static final int ORDERED_VALUE = 0;
+    public static final int UNORDERED_VALUE = 0;
     /**
-     * <code>UNORDERED = 1;</code>
+     * <code>ORDERED = 1;</code>
      */
-    public static final int UNORDERED_VALUE = 1;
+    public static final int ORDERED_VALUE = 1;
 
 
     public final int getNumber() {
@@ -196,8 +196,8 @@ public final class DataPlaneContract {
      */
     public static DeliveryOrder forNumber(int value) {
       switch (value) {
-        case 0: return ORDERED;
-        case 1: return UNORDERED;
+        case 0: return UNORDERED;
+        case 1: return ORDERED;
         default: return null;
       }
     }
@@ -2910,7 +2910,7 @@ public final class DataPlaneContract {
       if (egressConfig_ != null) {
         output.writeMessage(7, getEgressConfig());
       }
-      if (deliveryOrder_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder.ORDERED.getNumber()) {
+      if (deliveryOrder_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder.UNORDERED.getNumber()) {
         output.writeEnum(8, deliveryOrder_);
       }
       unknownFields.writeTo(output);
@@ -2946,7 +2946,7 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(7, getEgressConfig());
       }
-      if (deliveryOrder_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder.ORDERED.getNumber()) {
+      if (deliveryOrder_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder.UNORDERED.getNumber()) {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(8, deliveryOrder_);
       }
@@ -10762,8 +10762,8 @@ public final class DataPlaneContract {
       "cret\030\010 \001(\0132\n.ReferenceH\000B\006\n\004Auth\"<\n\010Cont" +
       "ract\022\022\n\ngeneration\030\001 \001(\004\022\034\n\tresources\030\002 " +
       "\003(\0132\t.Resource*,\n\rBackoffPolicy\022\017\n\013Expon" +
-      "ential\020\000\022\n\n\006Linear\020\001*+\n\rDeliveryOrder\022\013\n" +
-      "\007ORDERED\020\000\022\r\n\tUNORDERED\020\001*)\n\013ContentMode" +
+      "ential\020\000\022\n\n\006Linear\020\001*+\n\rDeliveryOrder\022\r\n" +
+      "\tUNORDERED\020\000\022\013\n\007ORDERED\020\001*)\n\013ContentMode" +
       "\022\n\n\006BINARY\020\000\022\016\n\nSTRUCTURED\020\001B[\n*dev.knat" +
       "ive.eventing.kafka.broker.contractB\021Data" +
       "PlaneContractZ\032control-plane/pkg/contrac" +
diff --git a/go.sum b/go.sum
index 34cbaed4f..cd6450ba7 100644
--- a/go.sum
+++ b/go.sum
@@ -705,6 +705,7 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
+github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
@@ -1294,16 +1295,13 @@ k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKU
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/eventing v0.21.1-0.20210319074953-c42772d75cc0 h1:9qEZPdRz+ZH2ZkU+ntqpnxbkHfpiiyrAZM1rXOf8C58=
 knative.dev/eventing v0.21.1-0.20210319074953-c42772d75cc0/go.mod h1:NfgiculSlgtkSOIgUciZ6BxRTzJgy2+VGnsChDW/vSA=
-knative.dev/hack v0.0.0-20210309141825-9b73a256fd9a h1:0UGKyvFzY1Czu0lP9+0IrJhwtKe2oNLI2GYa7MIQ5c0=
 knative.dev/hack v0.0.0-20210309141825-9b73a256fd9a/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210317214554-58edbdc42966 h1:qSnMQkq4gnhD5Hb8Mpfn7IrpkiQFhXXFPOA7uIl2lAc=
 knative.dev/hack v0.0.0-20210317214554-58edbdc42966/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210317214554-58edbdc42966/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20210315160101-6a33a1ab29ac h1:aGbaOcyNqxnvvujRtSMfkwehPER3TTsAw5mgCVqYZ08=
 knative.dev/pkg v0.0.0-20210315160101-6a33a1ab29ac/go.mod h1:7swdJzGy7U6iq8538vt2yFsZJqmnSAxMFexWK9Ktyes=
 knative.dev/pkg v0.0.0-20210318052054-dfeeb1817679 h1:5yTTe0oW2/a13DU2b4U42h7lSHZLWy32a/8DEkkH7C4=
 knative.dev/pkg v0.0.0-20210318052054-dfeeb1817679/go.mod h1:o1aTlbc+h60DP9P6nd/WmP99Ny9Ls3U/JIjL4xMgqpw=
-knative.dev/reconciler-test v0.0.0-20210318201654-cee1961aec9e h1:oDLr53xvxm2+36YgetdA76/iBgmHpCgLU0zULmRaOfQ=
 knative.dev/reconciler-test v0.0.0-20210318201654-cee1961aec9e/go.mod h1:0Rqoen7S2xgblliOrJfSe127JjPI2tMaTnEqhvmjtBI=
 knative.dev/reconciler-test v0.0.0-20210319003554-fe1effbb0013 h1:HT+sPYLpHl0xSK0LQJlViSpuTa6Of0cS6k5Qq1D101Y=
 knative.dev/reconciler-test v0.0.0-20210319003554-fe1effbb0013/go.mod h1:QSd2hWDDwFh+9AHOkiECsj2ZY2DGJBKnj6aT6LITBuw=
diff --git a/hack/tools.go b/hack/tools.go
index 7ef8a1477..896ffd623 100644
--- a/hack/tools.go
+++ b/hack/tools.go
@@ -24,7 +24,6 @@ import (
 	_ "knative.dev/pkg/hack"
 
 	// Test images from eventing
-	_ "knative.dev/eventing/test/test_images/event-flaker"
 	_ "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/proto/def/contract.proto b/proto/def/contract.proto
index d31989d43..2aff2c807 100644
--- a/proto/def/contract.proto
+++ b/proto/def/contract.proto
@@ -45,8 +45,8 @@ message EgressConfig {
 
 // Check dev.knative.eventing.kafka.broker.dispatcher.consumer.DeliveryOrder for more details
 enum DeliveryOrder {
-  ORDERED = 0;
-  UNORDERED = 1;
+  UNORDERED = 0;
+  ORDERED = 1;
 }
 
 message Egress {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index eb18b1a1e..b7d17fcc4 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1130,14 +1130,8 @@ knative.dev/eventing/test/performance/infra/common
 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/knconf
-knative.dev/eventing/test/rekt/resources/broker
-knative.dev/eventing/test/rekt/resources/flaker
 knative.dev/eventing/test/rekt/resources/svc
-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-sender
 knative.dev/eventing/test/test_images/heartbeats
 knative.dev/eventing/test/test_images/performance


Test patch:
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index e9292773a..f9aa522cd 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -153,6 +153,132 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - Trigger with ordered delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderOrdered)),
+				NewService(),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+						},
+					},
+				}, &configs),
+				BrokerDispatcherPod(configs.SystemNamespace, nil),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Egresses: []*contract.Egress{
+								{
+									Destination:   ServiceURL,
+									ConsumerGroup: TriggerUUID,
+									Uid:           TriggerUUID,
+									DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								},
+							},
+						},
+					},
+					Generation: 1,
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerSubscribed(),
+						withSubscriberURI,
+						reconcilertesting.WithTriggerDependencyReady(),
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderOrdered),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - Trigger with unordered delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+				),
+				newTrigger(reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderUnordered)),
+				NewService(),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+						},
+					},
+				}, &configs),
+				BrokerDispatcherPod(configs.SystemNamespace, nil),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Egresses: []*contract.Egress{
+								{
+									Destination:   ServiceURL,
+									ConsumerGroup: TriggerUUID,
+									Uid:           TriggerUUID,
+									DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+								},
+							},
+						},
+					},
+					Generation: 1,
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerSubscribed(),
+						withSubscriberURI,
+						reconcilertesting.WithTriggerDependencyReady(),
+						reconcilertesting.WithTriggerBrokerReady(),
+						reconcilertesting.WithTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, deliveryOrderUnordered),
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - Trigger delivery",
 			Objects: []runtime.Object{
@@ -430,6 +556,31 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				},
 			},
 		},
+		{
+			Name: "Broker not ready",
+			Objects: []runtime.Object{
+				newTrigger(),
+				NewBroker(
+					func(v *eventing.Broker) { v.Status.InitializeConditions() },
+					BrokerConfigNotParsed("wrong"),
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerFailed("wrong", ""),
+					),
+				},
+			},
+		},
 		{
 			Name: "Broker deleted, no broker in config map",
 			Objects: []runtime.Object{
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go b/test/e2e_new/broker/broker.go
similarity index 74%
rename from vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go
rename to test/e2e_new/broker/broker.go
index 8d111f983..f39a6fa03 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.go
+++ b/test/e2e_new/broker/broker.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * 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 broker
 
@@ -33,11 +33,12 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
+// TODO copy pasted from eventing, remove once upstream is clear how templating should work...
+
 var EnvCfg EnvConfig
 
 type EnvConfig struct {
-	BrokerClass        string `envconfig:"BROKER_CLASS" default:"MTChannelBasedBroker" required:"true"`
-	BrokerTemplatesDir string `envconfig:"BROKER_TEMPLATES"`
+	BrokerClass string `envconfig:"BROKER_CLASS" default:"MTChannelBasedBroker" required:"true"`
 }
 
 func init() {
@@ -47,15 +48,6 @@ func init() {
 	}
 }
 
-func WithEnvConfig() []manifest.CfgFn {
-	cfg := []manifest.CfgFn{WithBrokerClass(EnvCfg.BrokerClass)}
-
-	if EnvCfg.BrokerTemplatesDir != "" {
-		cfg = append(cfg, WithBrokerTemplateFiles(EnvCfg.BrokerTemplatesDir))
-	}
-	return cfg
-}
-
 func Gvr() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1", Resource: "brokers"}
 }
@@ -67,9 +59,14 @@ func WithBrokerClass(class string) manifest.CfgFn {
 	}
 }
 
-func WithBrokerTemplateFiles(dir string) manifest.CfgFn {
-	return func(cfg map[string]interface{}) {
-		cfg["__brokerTemplateDir"] = dir
+// WithConfig adds the broker class config to a Broker spec.
+func WithConfig(name string) manifest.CfgFn {
+	return func(templateData map[string]interface{}) {
+		cfg := make(map[string]interface{})
+		cfg["kind"] = "ConfigMap"
+		cfg["apiVersion"] = "v1"
+		cfg["name"] = name
+		templateData["config"] = cfg
 	}
 }
 
@@ -127,13 +124,6 @@ func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 		fn(cfg)
 	}
 
-	if dir, ok := cfg["__brokerTemplateDir"]; ok {
-		return func(ctx context.Context, t feature.T) {
-			if _, err := manifest.InstallYaml(ctx, dir.(string), cfg); err != nil {
-				t.Fatal(err)
-			}
-		}
-	}
 	return func(ctx context.Context, t feature.T) {
 		if _, err := manifest.InstallLocalYaml(ctx, cfg); err != nil {
 			t.Fatal(err)
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/broker/broker.yaml b/test/e2e_new/broker/broker.yaml
similarity index 100%
rename from vendor/knative.dev/eventing/test/rekt/resources/broker/broker.yaml
rename to test/e2e_new/broker/broker.yaml
diff --git a/test/e2e_new/broker_test.go b/test/e2e_new/broker_test.go
deleted file mode 100644
index af430c52d..000000000
--- a/test/e2e_new/broker_test.go
+++ /dev/null
@@ -1,69 +0,0 @@
-// +build e2e
-
-/*
- * Copyright 2020 The Knative Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package e2e_new
-
-import (
-	"testing"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
-
-	"knative.dev/pkg/system"
-	_ "knative.dev/pkg/system/testing"
-	"knative.dev/reconciler-test/pkg/k8s"
-	"knative.dev/reconciler-test/pkg/knative"
-
-	brokerfeatures "knative.dev/eventing/test/rekt/features/broker"
-	"knative.dev/eventing/test/rekt/resources/broker"
-)
-
-func TestBrokerSourceToSink(t *testing.T) {
-	t.Parallel()
-
-	ctx, env := global.Environment(
-		knative.WithKnativeNamespace(system.Namespace()),
-		knative.WithLoggingConfig,
-		knative.WithTracingConfig,
-		k8s.WithEventListener,
-	)
-
-	// Install and wait for a Ready Broker.
-	env.Prerequisite(ctx, t, brokerfeatures.GoesReady("default", broker.WithBrokerClass(kafka.BrokerClass)))
-
-	// Test that a Broker can act as middleware.
-	env.Test(ctx, t, brokerfeatures.SourceToSink("default"))
-
-	env.Finish()
-}
-
-func TestBrokerIngressConformance(t *testing.T) {
-	t.Parallel()
-
-	ctx, env := global.Environment(
-		knative.WithKnativeNamespace(system.Namespace()),
-		knative.WithLoggingConfig,
-		knative.WithTracingConfig,
-		k8s.WithEventListener,
-	)
-
-	for _, f := range brokerfeatures.BrokerIngressConformanceFeatures(kafka.BrokerClass) {
-		env.Test(ctx, t, f)
-	}
-
-	env.Finish()
-}
diff --git a/test/e2e_new/ordered_test.go b/test/e2e_new/ordered_test.go
new file mode 100644
index 000000000..93cf19010
--- /dev/null
+++ b/test/e2e_new/ordered_test.go
@@ -0,0 +1,148 @@
+// +build e2e
+
+/*
+ * 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 e2e_new
+
+import (
+	"context"
+	"sort"
+	"strconv"
+	"testing"
+	"time"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/stretchr/testify/require"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/single_partition_config"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/utils"
+	"knative.dev/eventing/test/rekt/resources/svc"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+
+	"knative.dev/eventing-kafka-broker/test/e2e_new/broker"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/trigger"
+
+	. "knative.dev/reconciler-test/pkg/eventshub/assert"
+)
+
+func SinglePartitionOrderedDelivery() *feature.Feature {
+	f := feature.NewFeature()
+
+	const responseWaitTime = 100 * time.Millisecond
+
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install one partition configuration", single_partition_config.Install)
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		broker.WithBrokerClass(kafka.BrokerClass),
+		broker.WithConfig(single_partition_config.ConfigMapName),
+	))
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
+
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.ResponseWaitTime(responseWaitTime),
+	))
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(svc.AsRef(sinkName), ""),
+		trigger.WithAnnotation("kafka.eventing.knative.dev/delivery.order", "ordered"),
+	))
+	f.Setup("trigger is ready", trigger.IsReady(triggerName))
+
+	f.Setup("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(broker.Gvr(), brokerName),
+		eventshub.InputEventWithEncoding(ev, cloudevents.EncodingBinary),
+		eventshub.AddSequence,
+		eventshub.SendMultipleEvents(20, 100*time.Millisecond),
+	))
+
+	f.Assert("receive events in order", func(ctx context.Context, t feature.T) {
+		events := eventshub.StoreFromContext(ctx, sinkName).AssertExact(
+			20,
+			MatchKind(EventReceived),
+			MatchEvent(cetest.ContainsExtensions("sequence")),
+		)
+
+		// Check we received exactly 20 and no more
+		require.Len(t, events, 20)
+
+		// Now we need to check we received these in order
+		sort.SliceStable(events, func(i, j int) bool {
+			return events[i].Time.Before(events[j].Time)
+		})
+
+		// Test sequence
+		for i, event := range events {
+			expectedSequence := i + 1 // sequence is 1 indexed
+			var actualSequenceStr string
+			err := event.Event.ExtensionAs("sequence", &actualSequenceStr)
+			require.NoError(t, err)
+			actualSequence, err := strconv.Atoi(actualSequenceStr)
+			require.NoError(t, err)
+			require.Equal(t, expectedSequence, actualSequence, "events: %+v", events)
+		}
+
+		// Test timings: because events are ordered, and the receiver pauses for 100 ms,
+		// their time should be > ~100 ms distant
+
+		// Assuming 10ms is the clock skew (highly improbable on a local cluster)
+		const clockSkew = 10 * time.Millisecond
+
+		prev := events[0].Time
+		for _, event := range events[1:] {
+			require.True(t, prev.Before(event.Time), "EventInfo.Time should be before the previous EventInfo.Time: %s < %s", prev, event.Time)
+			require.Greater(t, event.Time.Sub(prev) + clockSkew, responseWaitTime)
+
+			prev = event.Time
+		}
+	})
+
+	f.Teardown("log contract config map", utils.LogContractConfigMap)
+
+	return f
+}
+
+func TestOrderedDelivery(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, SinglePartitionOrderedDelivery())
+}
diff --git a/test/e2e_new/single_partition_config/install.go b/test/e2e_new/single_partition_config/install.go
new file mode 100644
index 000000000..49bd88efb
--- /dev/null
+++ b/test/e2e_new/single_partition_config/install.go
@@ -0,0 +1,34 @@
+/*
+ * 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 single_partition_config
+
+import (
+	"context"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+const ConfigMapName = "single-partition-kafka-broker-config"
+
+func Install(ctx context.Context, t feature.T) {
+	if _, err := manifest.InstallLocalYaml(ctx, map[string]interface{}{
+		"name": ConfigMapName,
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/test/e2e_new/single_partition_config/single_partition_config.yaml b/test/e2e_new/single_partition_config/single_partition_config.yaml
new file mode 100644
index 000000000..b9ba3dfac
--- /dev/null
+++ b/test/e2e_new/single_partition_config/single_partition_config.yaml
@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+data:
+  default.topic.partitions: "1"
+  default.topic.replication.factor: "1"
+  bootstrap.servers: "my-cluster-kafka-bootstrap.kafka:9092"
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go b/test/e2e_new/trigger/trigger.go
similarity index 58%
rename from vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go
rename to test/e2e_new/trigger/trigger.go
index 5fdac6c2a..9479220a5 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.go
+++ b/test/e2e_new/trigger/trigger.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * 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 trigger
 
@@ -27,30 +27,26 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
+// TODO copy pasted from eventing, remove once upstream is clear how templating should work...
+
+type CfgFn func(map[string]interface{})
+
 func gvr() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1", Resource: "triggers"}
 }
 
-// WithFilter adds the filter related config to a Trigger spec.
-func WithFilter(attributes map[string]string) manifest.CfgFn {
+// WithAnnotation adds an annotation
+func WithAnnotation(key string, value string) CfgFn {
 	return func(cfg map[string]interface{}) {
-		if _, set := cfg["filter"]; !set {
-			cfg["filter"] = map[string]interface{}{}
-		}
-		filter := cfg["filter"].(map[string]interface{})
-		if _, set := filter["filter"]; !set {
-			filter["attributes"] = map[string]interface{}{}
-		}
-		attrs := filter["attributes"].(map[string]interface{})
-
-		for k, v := range attributes {
-			attrs[k] = v
+		if _, set := cfg["annotations"]; !set {
+			cfg["annotations"] = map[string]interface{}{}
 		}
+		(cfg["annotations"].(map[string]interface{}))[key] = value
 	}
 }
 
 // WithSubscriber adds the subscriber related config to a Trigger spec.
-func WithSubscriber(ref *duckv1.KReference, uri string) manifest.CfgFn {
+func WithSubscriber(ref *duckv1.KReference, uri string) CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, set := cfg["subscriber"]; !set {
 			cfg["subscriber"] = map[string]interface{}{}
@@ -74,7 +70,7 @@ func WithSubscriber(ref *duckv1.KReference, uri string) manifest.CfgFn {
 }
 
 // Install will create a Trigger resource, augmented with the config fn options.
-func Install(name, brokerName string, opts ...manifest.CfgFn) feature.StepFn {
+func Install(name, brokerName string, opts ...CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name": name,
 	}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml b/test/e2e_new/trigger/trigger.yaml
similarity index 89%
rename from vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
rename to test/e2e_new/trigger/trigger.yaml
index 40accda18..6b512395e 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/trigger/trigger.yaml
+++ b/test/e2e_new/trigger/trigger.yaml
@@ -1,4 +1,4 @@
-# Copyright 2020 The Knative Authors
+# 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.
@@ -17,6 +17,10 @@ kind: Trigger
 metadata:
   name: {{ .name }}
   namespace: {{ .namespace }}
+  annotations:
+  {{ range $key, $value := .annotations }}
+    {{ $key }}: "{{ $value }}"
+  {{ end }}
 spec:
   {{ if .brokerName }}
   broker: {{ .brokerName }}
diff --git a/test/e2e_new/utils/log_cm.go b/test/e2e_new/utils/log_cm.go
new file mode 100644
index 000000000..746104367
--- /dev/null
+++ b/test/e2e_new/utils/log_cm.go
@@ -0,0 +1,48 @@
+/*
+ * 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 utils
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+// LogContractConfigMap logs the broker config map
+func LogContractConfigMap(ctx context.Context, t feature.T) {
+	LogConfigMap("kafka-broker-brokers-triggers", knative.KnativeNamespaceFromContext(ctx))(ctx, t)
+}
+
+// LogConfigMap logs the provided config map
+func LogConfigMap(name string, namespace string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		cm, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(namespace).Get(ctx, name, metav1.GetOptions{})
+		require.NoError(t, err)
+
+		t.Logf("Config map %s/%s", cm.Name, cm.Namespace)
+		for k, v := range cm.Data {
+			t.Logf("%s: %s", k, v)
+		}
+		for k, v := range cm.BinaryData {
+			t.Logf("%s: %s", k, string(v))
+		}
+	}
+}
diff --git a/third_party/VENDOR-LICENSE/knative.dev/reconciler-test/pkg/LICENSE b/third_party/VENDOR-LICENSE/knative.dev/reconciler-test/pkg/LICENSE
new file mode 100644
index 000000000..261eeb9e9
--- /dev/null
+++ b/third_party/VENDOR-LICENSE/knative.dev/reconciler-test/pkg/LICENSE
@@ -0,0 +1,201 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   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.
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go b/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
deleted file mode 100644
index 0c817133c..000000000
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
+++ /dev/null
@@ -1,448 +0,0 @@
-/*
-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 broker
-
-import (
-	"context"
-	"encoding/json"
-	"strings"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
-	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
-	eventingclientsetv1 "knative.dev/eventing/pkg/client/clientset/versioned/typed/eventing/v1"
-	eventingclient "knative.dev/eventing/pkg/client/injection/client"
-	"knative.dev/eventing/test/rekt/features/knconf"
-	brokerresources "knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/svc"
-	triggerresources "knative.dev/eventing/test/rekt/resources/trigger"
-	"knative.dev/reconciler-test/pkg/environment"
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/state"
-)
-
-func ControlPlaneConformance(brokerName string) *feature.FeatureSet {
-	fs := &feature.FeatureSet{
-		Name: "Knative Broker Specification - Control Plane",
-		Features: []feature.Feature{
-			*ControlPlaneBroker(brokerName),
-			*ControlPlaneTrigger_GivenBroker(brokerName),
-			*ControlPlaneTrigger_GivenBrokerTriggerReady(brokerName),
-			*ControlPlaneTrigger_WithBrokerLifecycle(),
-			*ControlPlaneTrigger_WithValidFilters(brokerName),
-			*ControlPlaneTrigger_WithInvalidFilters(brokerName),
-			*ControlPlaneDelivery(brokerName),
-		},
-	}
-	// TODO: This is not a control plane test, or at best it is a blend with data plane.
-	// Must("Events that pass the attributes filter MUST include context or extension attributes that match all key-value pairs exactly.", todo)
-
-	return fs
-}
-
-func setBrokerName(name string) feature.StepFn {
-	return func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, BrokerNameKey, name)
-	}
-}
-
-func ControlPlaneBroker(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Broker")
-
-	f.Setup("Set Broker Name", setBrokerName(brokerName))
-
-	f.Stable("Conformance").
-		Should("Broker objects SHOULD include a Ready condition in their status",
-			brokerHasReadyInConditions).
-		Should("The Broker SHOULD indicate Ready=True when its ingress is available to receive events.",
-			readyBrokerHasIngressAvailable).
-		Should("While a Broker is Ready, it SHOULD be a valid Addressable and its `status.address.url` field SHOULD indicate the address of its ingress.",
-			readyBrokerIsAddressable).
-		Should("The class of a Broker object SHOULD be immutable.",
-			brokerClassIsImmutable)
-	return f
-}
-
-func ControlPlaneTrigger_GivenBroker(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Trigger, Given Broker")
-	f.Setup("Set Broker Name", setBrokerName(brokerName))
-
-	subscriberName := feature.MakeRandomK8sName("sub")
-	f.Setup("Install Subscriber", svc.Install(subscriberName, "bad", "svc"))
-
-	triggerName := feature.MakeRandomK8sName("trigger")
-	f.Setup("Create a Trigger", triggerresources.Install(triggerName, brokerName,
-		triggerresources.WithSubscriber(svc.AsRef(subscriberName), ""),
-	))
-
-	f.Setup("Set Trigger Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, TriggerNameKey, triggerName)
-	})
-
-	f.Stable("Conformance").
-		Should("Triggers SHOULD include a Ready condition in their status.",
-			triggerHasReadyInConditions).
-		Should("The Trigger SHOULD indicate Ready=True when events can be delivered to its subscriber.",
-			readyTriggerCanDeliver).
-		Must("Triggers MUST be assigned to exactly one Broker.",
-			triggerHasOneBroker).
-		Must("The assigned Broker of a Trigger SHOULD be immutable.",
-			triggerSpecBrokerIsImmutable)
-
-	return f
-}
-
-func ControlPlaneTrigger_GivenBrokerTriggerReady(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Trigger, Given Broker")
-	f.Setup("Set Broker Name", setBrokerName(brokerName))
-
-	subscriberName := feature.MakeRandomK8sName("sub")
-	f.Setup("Install Subscriber", svc.Install(subscriberName, "bad", "svc"))
-
-	triggerName := feature.MakeRandomK8sName("trigger")
-	f.Setup("Create a Trigger", triggerresources.Install(triggerName, brokerName,
-		triggerresources.WithSubscriber(svc.AsRef(subscriberName), ""),
-	))
-
-	f.Setup("Set Trigger Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, TriggerNameKey, triggerName)
-	})
-
-	f.Requirement("The Trigger is Ready", triggerresources.IsReady(triggerName))
-
-	f.Stable("Conformance").
-		Should("While a Trigger is Ready, it SHOULD indicate its subscriber's URI via the `status.subscriberUri` field.",
-			readyTriggerHasSubscriberURI)
-
-	return f
-}
-
-func ControlPlaneTrigger_WithBrokerLifecycle() *feature.Feature {
-	f := feature.NewFeatureNamed("Trigger, With Broker Lifecycle")
-
-	subscriberName := feature.MakeRandomK8sName("sub")
-	f.Setup("Install Subscriber", svc.Install(subscriberName, "bad", "svc"))
-
-	brokerName := feature.MakeRandomK8sName("broker")
-
-	triggerName := feature.MakeRandomK8sName("trigger")
-	f.Setup("Create a Trigger", triggerresources.Install(triggerName, brokerName,
-		triggerresources.WithSubscriber(svc.AsRef(subscriberName), ""),
-	))
-
-	f.Setup("Set Trigger Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, TriggerNameKey, triggerName)
-	})
-
-	f.Stable("Conformance").
-		May("A Trigger MAY be created before its assigned Broker exists.",
-			triggerHasOneBroker).
-		Should("A Trigger SHOULD progress to Ready when its assigned Broker exists and is Ready.",
-			func(ctx context.Context, t feature.T) {
-				brokerresources.Install(brokerName, brokerresources.WithEnvConfig()...)(ctx, t) // Default broker from Env.
-				brokerresources.IsReady(brokerName)(ctx, t)
-				triggerresources.IsReady(triggerName)(ctx, t)
-			})
-	return f
-}
-
-func ControlPlaneTrigger_WithValidFilters(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Trigger, With Filters")
-	f.Setup("Set Broker Name", setBrokerName(brokerName))
-
-	subscriberName := feature.MakeRandomK8sName("sub")
-	f.Setup("Install Subscriber", svc.Install(subscriberName, "bad", "svc"))
-
-	// CloudEvents attribute names MUST consist of lower-case letters ('a' to 'z') or digits ('0' to '9') from the ASCII character set. Attribute names SHOULD be descriptive and terse and SHOULD NOT exceed 20 characters in length.
-	filters := map[string]string{
-		"source":               "a source",
-		"id":                   "an id",
-		"specversion":          "the spec version",
-		"type":                 "the type",
-		"subject":              "a subject",
-		"time":                 "a time",
-		"datacontenttype":      "a datacontenttype",
-		"dataschema":           "a dataschema",
-		"aaa":                  "bbb",
-		"c1d2e3":               "123",
-		"abcdefghijklmnopqrst": "max length",
-	}
-
-	triggerName := feature.MakeRandomK8sName("trigger")
-	f.Setup("Create a Trigger", triggerresources.Install(triggerName, brokerName,
-		triggerresources.WithSubscriber(svc.AsRef(subscriberName), ""),
-		triggerresources.WithFilter(filters),
-	))
-
-	f.Setup("Set Trigger Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, TriggerNameKey, triggerName)
-	})
-
-	f.Stable("Conformance").
-		Must("The attributes filter specifying a list of key-value pairs MUST be supported by Trigger.",
-			// Compare the passed filters with what is found on the control plane.
-			func(ctx context.Context, t feature.T) {
-				trigger := getTrigger(ctx, t)
-				got := make(map[string]string)
-				for k, v := range trigger.Spec.Filter.Attributes {
-					got[k] = v
-				}
-				want := filters
-				if diff := cmp.Diff(want, got, cmpopts.SortMaps(func(a, b string) bool {
-					return a < b
-				})); diff != "" {
-					t.Error("Filters do not match (-want, +got) =", diff)
-				}
-			})
-
-	return f
-}
-
-func ControlPlaneTrigger_WithInvalidFilters(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Trigger, With Filters")
-	f.Setup("Set Broker Name", setBrokerName(brokerName))
-
-	subscriberName := feature.MakeRandomK8sName("sub")
-	f.Setup("Install Subscriber", svc.Install(subscriberName, "bad", "svc"))
-
-	// CloudEvents attribute names MUST consist of lower-case letters ('a' to 'z') or digits ('0' to '9') from the ASCII character set. Attribute names SHOULD be descriptive and terse and SHOULD NOT exceed 20 characters in length.
-	filters := map[string]string{
-		"SOURCE":              "not lower case letters, all",
-		"Source":              "not lower case letters, first",
-		"souRce":              "not lower case letters, not first",
-		"s pace s":            "no spaces",
-		"s_pace_s":            "no underscores",
-		"s-pace-s":            "no dashes",
-		"123":                 "just numbers",
-		"😊":                   "unicode not supported",
-		"!@#$%^&*()-_=_`~+\\": "other non-(a-z,0-9) type chars, top row",
-		"{}[];':\"<>,./?":     "other non-(a-z,0-9) type chars, brackets",
-	}
-
-	triggerName := feature.MakeRandomK8sName("trigger")
-	f.Setup("Create a Trigger", triggerresources.Install(triggerName, brokerName,
-		triggerresources.WithSubscriber(svc.AsRef(subscriberName), ""),
-	))
-
-	f.Setup("Set Trigger Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, TriggerNameKey, triggerName)
-	})
-
-	asserter := f.Stable("Conformance - Negatives - The attributes filter specifying a list of key-value pairs MUST be supported by Trigger.")
-
-	for key, value := range filters {
-		k := key
-		v := value
-		asserter.Must("Reject invalid filter - "+k+" - "+v,
-			// Compare the passed filters with what is found on the control plane.
-			func(ctx context.Context, t feature.T) {
-				trigger := getTrigger(ctx, t)
-
-				if trigger.Spec.Filter == nil {
-					trigger.Spec.Filter = &eventingv1.TriggerFilter{
-						Attributes: map[string]string{},
-					}
-				} else if trigger.Spec.Filter.Attributes == nil {
-					trigger.Spec.Filter.Attributes = map[string]string{}
-				}
-
-				trigger.Spec.Filter.Attributes[k] = v
-
-				_, err := Client(ctx).Triggers.Update(ctx, trigger, metav1.UpdateOptions{})
-				if err != nil {
-					// We expect an error.
-					// Success!
-				} else {
-					t.Error("expected Trigger to reject the spec.filter update.")
-				}
-			})
-	}
-	return f
-}
-
-func ControlPlaneDelivery(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Delivery Spec")
-
-	f.Setup("Set Broker Name", setBrokerName(brokerName))
-
-	f.Stable("Conformance").
-		Should("When `BrokerSpec.Delivery` and `TriggerSpec.Delivery` are both not configured, no delivery spec SHOULD be used.",
-			todo).
-		Should("When `BrokerSpec.Delivery` is configured, but not the specific `TriggerSpec.Delivery`, then the `BrokerSpec.Delivery` SHOULD be used.",
-			todo).
-		Should("When `TriggerSpec.Delivery` is configured, then `TriggerSpec.Delivery` SHOULD be used.",
-			todo)
-
-	return f
-}
-
-type EventingClient struct {
-	Brokers  eventingclientsetv1.BrokerInterface
-	Triggers eventingclientsetv1.TriggerInterface
-}
-
-func Client(ctx context.Context) *EventingClient {
-	ec := eventingclient.Get(ctx).EventingV1()
-	env := environment.FromContext(ctx)
-
-	return &EventingClient{
-		Brokers:  ec.Brokers(env.Namespace()),
-		Triggers: ec.Triggers(env.Namespace()),
-	}
-}
-
-const (
-	BrokerNameKey  = "brokerName"
-	TriggerNameKey = "triggerName"
-)
-
-func getBroker(ctx context.Context, t feature.T) *eventingv1.Broker {
-	c := Client(ctx)
-	name := state.GetStringOrFail(ctx, t, BrokerNameKey)
-
-	broker, err := c.Brokers.Get(ctx, name, metav1.GetOptions{})
-	if err != nil {
-		t.Errorf("failed to get Broker, %v", err)
-	}
-	return broker
-}
-
-func getTrigger(ctx context.Context, t feature.T) *eventingv1.Trigger {
-	c := Client(ctx)
-	name := state.GetStringOrFail(ctx, t, TriggerNameKey)
-
-	trigger, err := c.Triggers.Get(ctx, name, metav1.GetOptions{})
-	if err != nil {
-		t.Errorf("failed to get Trigger, %v", err)
-	}
-	return trigger
-}
-
-func brokerHasReadyInConditions(ctx context.Context, t feature.T) {
-	var broker *eventingv1.Broker
-
-	interval, timeout := environment.PollTimingsFromContext(ctx)
-	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
-		broker = getBroker(ctx, t)
-		if broker.Status.ObservedGeneration != 0 {
-			return true, nil
-		}
-		return false, nil
-	})
-	if err != nil {
-		t.Errorf("unable to get a reconciled Broker (status.observedGeneration != 0)")
-	}
-
-	knconf.HasReadyInConditions(ctx, t, broker.Status.Status)
-}
-
-func readyBrokerHasIngressAvailable(ctx context.Context, t feature.T) {
-	// TODO: I am not sure how to test this from the outside.
-}
-
-func readyBrokerIsAddressable(ctx context.Context, t feature.T) {
-	broker := getBroker(ctx, t)
-
-	if broker.Status.IsReady() {
-		if broker.Status.Address.URL == nil {
-			t.Errorf("broker is not addressable")
-		}
-		// Success!
-	} else {
-		t.Errorf("broker was not ready")
-	}
-}
-
-func brokerClassIsImmutable(ctx context.Context, t feature.T) {
-	broker := getBroker(ctx, t)
-
-	if broker.Annotations == nil {
-		broker.Annotations = map[string]string{}
-	}
-	// update annotations
-	broker.Annotations[eventingv1.BrokerClassAnnotationKey] = "Rekt.brokerClassIsImmutable"
-
-	if _, err := Client(ctx).Brokers.Update(ctx, broker, metav1.UpdateOptions{}); err != nil {
-		// Success!
-		t.Log("broker class is immutable")
-	} else {
-		t.Errorf("broker class is mutable")
-	}
-}
-
-func triggerHasReadyInConditions(ctx context.Context, t feature.T) {
-	var trigger *eventingv1.Trigger
-
-	interval, timeout := environment.PollTimingsFromContext(ctx)
-	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
-		trigger = getTrigger(ctx, t)
-		if trigger.Status.ObservedGeneration != 0 {
-			return true, nil
-		}
-		return false, nil
-	})
-	if err != nil {
-		t.Errorf("unable to get a reconciled Trigger (status.observedGeneration != 0)")
-	}
-
-	knconf.HasReadyInConditions(ctx, t, trigger.Status.Status)
-}
-
-func readyTriggerCanDeliver(ctx context.Context, t feature.T) {
-	trigger := getTrigger(ctx, t)
-	_ = trigger
-	// TODO: I am not sure how to test this from the outside.
-}
-
-func readyTriggerHasSubscriberURI(ctx context.Context, t feature.T) {
-	trigger := getTrigger(ctx, t)
-	if trigger.Status.IsReady() {
-		if trigger.Status.SubscriberURI == nil {
-			t.Errorf("trigger did not have subscriber uri in status")
-		}
-		// Success!
-	} else {
-		j, _ := json.Marshal(trigger)
-		t.Errorf("trigger was not ready, \n%s", string(j))
-	}
-}
-
-func triggerHasOneBroker(ctx context.Context, t feature.T) {
-	trigger := getTrigger(ctx, t)
-	if trigger.Spec.Broker == "" {
-		t.Error("broker is empty")
-	}
-	if strings.Contains(trigger.Spec.Broker, ",") {
-		t.Errorf("more than one broker specified: %q", trigger.Spec.Broker)
-	}
-}
-
-func triggerSpecBrokerIsImmutable(ctx context.Context, t feature.T) {
-	trigger := getTrigger(ctx, t)
-
-	// Update spec.broker
-	trigger.Spec.Broker = "Rekt.BrokerImmutable"
-
-	if _, err := Client(ctx).Triggers.Update(ctx, trigger, metav1.UpdateOptions{}); err != nil {
-		// Success!
-		t.Log("Trigger spec.broker is immutable")
-	} else {
-		t.Errorf("Trigger spec.broker is mutable")
-	}
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/data_plane.go b/vendor/knative.dev/eventing/test/rekt/features/broker/data_plane.go
deleted file mode 100644
index 97b5b054f..000000000
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/data_plane.go
+++ /dev/null
@@ -1,208 +0,0 @@
-/*
-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 broker
-
-import (
-	"context"
-	"errors"
-
-	"github.com/google/uuid"
-
-	. "github.com/cloudevents/sdk-go/v2/test"
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/reconciler-test/pkg/eventshub"
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/state"
-	eventshubmain "knative.dev/reconciler-test/pkg/test_images/eventshub"
-)
-
-func DataPlaneConformance(brokerName string) *feature.FeatureSet {
-	return &feature.FeatureSet{
-		Name: "Knative Broker Specification - Data Plane",
-		Features: []feature.Feature{
-			*DataPlaneIngress(brokerName),
-			*DataPlaneDelivery(brokerName),
-			*DataPlaneObservability(brokerName),
-		},
-	}
-}
-
-func DataPlaneIngress(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Ingress")
-
-	f.Setup("Set Broker Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, "brokerName", brokerName)
-	})
-
-	f.Stable("Conformance").
-		Should("A Broker SHOULD expose either an HTTP or HTTPS endpoint as ingress. It MAY expose both.",
-			todo).
-		Must("The ingress endpoint(s) MUST conform to at least one of the following versions of the specification: 0.3, 1.0",
-			brokerAcceptsCEVersions).
-		May("Other versions MAY be rejected.",
-			todo).
-		ShouldNot("The Broker SHOULD NOT perform an upgrade of the produced event's CloudEvents version.",
-			todo).
-		Should("It SHOULD support both Binary Content Mode and Structured Content Mode of the HTTP Protocol Binding for CloudEvents.",
-			todo).
-		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).
-		Must("Brokers MUST reject all HTTP produce requests with a method other than POST responding with HTTP status code `405 Method Not Supported`.",
-			todo).
-		Must("The Broker MUST respond with a 200-level HTTP status code if a produce request is accepted.",
-			todo).
-		Must("If a Broker receives a produce request and is unable to parse a valid CloudEvent, then it MUST reject the request with HTTP status code `400 Bad Request`.",
-			todo)
-
-	return f
-}
-
-func DataPlaneDelivery(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Delivery")
-
-	f.Setup("Set Broker Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, "brokerName", brokerName)
-	})
-
-	f.Stable("Conformance").
-		Must("Delivered events MUST conform to the CloudEvents specification.",
-			todo).
-		Should("All CloudEvent attributes set by the producer, including the data and specversion attributes, SHOULD be received at the subscriber identical to how they were received by the Broker.",
-			todo).
-		Should("The Broker SHOULD support delivering events via Binary Content Mode or Structured Content Mode of the HTTP Protocol Binding for CloudEvents.",
-			todo).
-		Should("Events accepted by the Broker SHOULD be delivered at least once to all subscribers of all Triggers*.",
-			todo).
-		//*
-		//1. are Ready when the produce request was received,
-		//1. specify filters that match the event, and
-		//1. exist when the event is able to be delivered.
-		May("Events MAY additionally be delivered to Triggers that become Ready after the event was accepted.",
-			todo).
-		May("Events MAY be enqueued or delayed between acceptance from a producer and delivery to a subscriber.",
-			todo).
-		May("The Broker MAY choose not to deliver an event due to persistent unavailability of a subscriber or limitations such as storage capacity.",
-			todo).
-		Should("The Broker SHOULD attempt to notify the operator in this case.",
-			todo).
-		May("The Broker MAY forward these events to an alternate endpoint or storage mechanism such as a dead letter queue.",
-			todo).
-		May("If no ready Trigger would match an accepted event, the Broker MAY drop that event without notifying the producer.",
-			todo).
-		May("If multiple Triggers reference the same subscriber, the subscriber MAY be expected to acknowledge successful delivery of an event multiple times.",
-			todo).
-		Should("Events contained in delivery responses SHOULD be published to the Broker ingress and processed as if the event had been produced to the Broker's addressable endpoint.",
-			todo).
-		Should("Events contained in delivery responses that are malformed SHOULD be treated as if the event delivery had failed.",
-			todo).
-		May("The subscriber MAY receive a confirmation that a reply event was accepted by the Broker.",
-			todo).
-		Should("If the reply event was not accepted, the initial event SHOULD be redelivered to the subscriber.",
-			todo)
-
-	return f
-}
-
-func DataPlaneObservability(brokerName string) *feature.Feature {
-	f := feature.NewFeatureNamed("Observability")
-
-	f.Setup("Set Broker Name", func(ctx context.Context, t feature.T) {
-		state.SetOrFail(ctx, t, "brokerName", brokerName)
-	})
-
-	f.Stable("Conformance").
-		Should("The Broker SHOULD expose a variety of metrics*.",
-			todo).
-		// * including, but not limited to:
-		//- Number of malformed produce requests (400-level responses)
-		//- Number of accepted produce requests (200-level responses)
-		//- Number of events delivered
-
-		Should("Metrics SHOULD be enabled by default, with a configuration parameter included to disable them if desired.",
-			todo).
-		Should("Upon receiving an event with context attributes defined in the CloudEvents Distributed Tracing extension the Broker SHOULD preserve that trace header on delivery to subscribers and on reply events, unless the reply is sent with a different set of tracing attributes.",
-			todo).
-		Should("Forwarded trace headers SHOULD be updated with any intermediate spans emitted by the broker.",
-			todo).
-		Should("Spans emitted by the Broker SHOULD follow the OpenTelemetry Semantic Conventions for Messaging System*.",
-			todo)
-	// *
-	//- messaging.system: "knative"
-	//- messaging.destination: broker:name.namespace or trigger:name.namespace with
-	//the Broker or Trigger 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 todo(ctx context.Context, t feature.T) {
-	t.Log("TODO, Implement this.")
-}
-
-func brokerAcceptsCEVersions(ctx context.Context, t feature.T) {
-	brokerName := state.GetStringOrFail(ctx, t, "brokerName")
-
-	u, err := broker.Address(ctx, brokerName)
-	if err != nil || u == nil {
-		t.Error("failed to get the address of the broker", brokerName, err)
-	}
-
-	opts := []eventshub.EventsHubOption{eventshub.StartSenderURL(u.String())}
-	uuids := map[string]string{
-		uuid.New().String(): "1.0",
-		uuid.New().String(): "0.3",
-	}
-	for uuid, version := range uuids {
-		// We need to use a different source name, otherwise, it will try to update
-		// the pod, which is immutable.
-		source := feature.MakeRandomK8sName("source")
-		event := FullEvent()
-		event.SetID(uuid)
-		event.SetSpecVersion(version)
-		opts = append(opts, eventshub.InputEvent(event))
-
-		eventshub.Install(source, opts...)(ctx, t)
-		store := eventshub.StoreFromContext(ctx, source)
-		// We are looking for two events, one of them is the sent event and the other
-		// is Response. We want to make sure the event was sent and that it was accepted.
-		// TODO: Figure out how to send events via eventshub that
-		events := store.AssertAtLeast(2, sentEventMatcher(uuid))
-		for _, e := range events {
-			if e.Kind == eventshubmain.EventResponse {
-				// Make sure HTTP response code is 2XX
-				if e.StatusCode < 200 || e.StatusCode > 299 {
-					t.Errorf("Expected statuscode 2XX got %d", e.StatusCode)
-				}
-			}
-		}
-	}
-}
-
-func sentEventMatcher(uuid string) func(eventshubmain.EventInfo) error {
-	return func(ei eventshubmain.EventInfo) error {
-		if ei.Kind == eventshubmain.EventSent && ei.Event.ID() == uuid {
-			return nil
-		}
-		if ei.Kind == eventshubmain.EventResponse {
-			return nil
-		}
-		return errors.New("no match")
-	}
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/ingress.go b/vendor/knative.dev/eventing/test/rekt/features/broker/ingress.go
deleted file mode 100644
index 2e116f2fc..000000000
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/ingress.go
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
-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 broker
-
-import (
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	cetest "github.com/cloudevents/sdk-go/v2/test"
-	"knative.dev/reconciler-test/pkg/eventshub"
-	"knative.dev/reconciler-test/pkg/feature"
-
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/svc"
-	"knative.dev/eventing/test/rekt/resources/trigger"
-
-	. "github.com/cloudevents/sdk-go/v2/test"
-	. "knative.dev/reconciler-test/pkg/eventshub/assert"
-)
-
-func BrokerIngressConformanceFeatures(brokerClass string) []*feature.Feature {
-	var feats []*feature.Feature
-	for _, version := range []string{cloudevents.VersionV03, cloudevents.VersionV1} {
-		for _, enc := range []cloudevents.Encoding{cloudevents.EncodingBinary, cloudevents.EncodingStructured} {
-			feats = append(feats, brokerIngressConformanceFeature(brokerClass, version, enc))
-		}
-	}
-	feats = append(feats, brokerIngressConformanceBadEvent(brokerClass))
-	return feats
-}
-
-func brokerIngressConformanceFeature(brokerClass string, version string, enc cloudevents.Encoding) *feature.Feature {
-	sourceName := feature.MakeRandomK8sName("source")
-	sinkName := feature.MakeRandomK8sName("sink")
-	triggerName := feature.MakeRandomK8sName("trigger")
-	brokerName := feature.MakeRandomK8sName("broker")
-
-	event := cetest.FullEvent()
-	event.SetSpecVersion(version)
-
-	f := new(feature.Feature)
-	f.Name = "BrokerIngress" + version + enc.String()
-
-	f.Setup("install sink", eventshub.Install(sinkName, eventshub.StartReceiver))
-	f.Setup("install broker", broker.Install(brokerName, broker.WithBrokerClass(brokerClass)))
-	f.Setup("install trigger", trigger.Install(triggerName, brokerName, trigger.WithSubscriber(svc.AsRef(sinkName), "")))
-	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
-
-	f.Setup("install source", eventshub.Install(
-		sourceName,
-		eventshub.StartSenderToResource(broker.Gvr(), brokerName),
-		eventshub.InputEventWithEncoding(event, enc),
-	))
-
-	f.Stable("ingress supports v"+version).
-		Must("accept the event", OnStore(sourceName).Match(
-			MatchKind(EventResponse),
-			MatchStatusCode(202),
-		).AtLeast(1)).
-		Must("deliver the event",
-			OnStore(sinkName).MatchEvent(
-				HasId(event.ID()),
-				HasSpecVersion(event.SpecVersion()),
-			).AtLeast(1))
-
-	return f
-}
-
-func brokerIngressConformanceBadEvent(brokerClass string) *feature.Feature {
-	sourceName := feature.MakeRandomK8sName("source")
-	sinkName := feature.MakeRandomK8sName("sink")
-	triggerName := feature.MakeRandomK8sName("trigger")
-	brokerName := feature.MakeRandomK8sName("broker")
-
-	eventID := "four-hundred-on-bad-ce"
-
-	f := new(feature.Feature)
-	f.Name = "BrokerIngressConformanceBadEvent"
-
-	f.Setup("install sink", eventshub.Install(sinkName, eventshub.StartReceiver))
-	f.Setup("install broker", broker.Install(brokerName, broker.WithBrokerClass(brokerClass)))
-	f.Setup("install trigger", trigger.Install(triggerName, brokerName, trigger.WithSubscriber(svc.AsRef(sinkName), "")))
-	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
-
-	f.Setup("install source", eventshub.Install(sourceName,
-		eventshub.StartSenderToResource(broker.Gvr(), brokerName),
-		eventshub.InputHeader("ce-specversion", "9000.1"),
-		eventshub.InputHeader("ce-type", "sometype"),
-		eventshub.InputHeader("ce-source", "400.request.sender.test.knative.dev"),
-		eventshub.InputHeader("ce-id", eventID),
-		eventshub.InputBody(";la}{kjsdf;oai2095{}{}8234092349807asdfashdf"),
-	))
-
-	f.Stable("ingress").
-		Must("respond with 400 on bad event", OnStore(sourceName).Match(
-			MatchKind(EventResponse),
-			MatchStatusCode(400),
-		).AtLeast(1)).
-		Must("must not propagate bad event",
-			OnStore(sinkName).MatchEvent(HasId(eventID)).Not())
-
-	return f
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
deleted file mode 100644
index e785f95a0..000000000
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package broker
-
-import (
-	"fmt"
-
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/manifest"
-
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/svc"
-	"knative.dev/eventing/test/rekt/resources/trigger"
-)
-
-// TriggerGoesReady returns a feature that tests after the creation of a
-// Trigger, it becomes ready. This feature assumes the Broker already exists.
-func TriggerGoesReady(name, brokerName string) *feature.Feature {
-	cfg := []manifest.CfgFn(nil)
-
-	f := new(feature.Feature)
-
-	// The test needs a subscriber.
-	sub := feature.MakeRandomK8sName("sub")
-	f.Setup("install a service", svc.Install(sub, "app", "rekt"))
-
-	// Point the Trigger subscriber to the service installed in setup.
-	cfg = append(cfg, trigger.WithSubscriber(svc.AsRef(name), ""))
-
-	// Install the trigger
-	f.Setup(fmt.Sprintf("install trigger %q", name), trigger.Install(name, brokerName, cfg...))
-
-	// Wait for a ready broker.
-	f.Requirement("broker is ready", broker.IsReady(brokerName))
-
-	f.Stable("trigger").
-		Must("be ready", trigger.IsReady(name))
-
-	return f
-}
-
-// GoesReady returns a feature that will create a Broker of the given
-// name and class, and confirm it becomes ready with an address.
-func GoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
-	f := new(feature.Feature)
-
-	f.Setup(fmt.Sprintf("install broker %q", name), broker.Install(name, cfg...))
-
-	f.Stable("broker").
-		Must("be ready", broker.IsReady(name)).
-		Must("be addressable", broker.IsAddressable(name))
-
-	return f
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
deleted file mode 100644
index 434d43c13..000000000
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
+++ /dev/null
@@ -1,182 +0,0 @@
-/*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package broker
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/google/uuid"
-
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/flaker"
-	"knative.dev/eventing/test/rekt/resources/svc"
-	"knative.dev/eventing/test/rekt/resources/trigger"
-	"knative.dev/reconciler-test/pkg/environment"
-	"knative.dev/reconciler-test/pkg/eventshub"
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/manifest"
-
-	. "github.com/cloudevents/sdk-go/v2/test"
-	. "knative.dev/reconciler-test/pkg/eventshub/assert"
-)
-
-// SourceToSink tests to see if a Ready Broker acts as middleware.
-// LoadGenerator --> in [Broker] out --> Recorder
-func SourceToSink(brokerName string) *feature.Feature {
-	source := feature.MakeRandomK8sName("source")
-	sink := feature.MakeRandomK8sName("sink")
-	via := feature.MakeRandomK8sName("via")
-	event := FullEvent()
-
-	f := new(feature.Feature)
-
-	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
-
-	// Point the Trigger subscriber to the sink svc.
-	cfg := []manifest.CfgFn{trigger.WithSubscriber(svc.AsRef(sink), "")}
-
-	// Install the trigger
-	f.Setup("install trigger", trigger.Install(via, brokerName, cfg...))
-
-	f.Setup("trigger goes ready", trigger.IsReady(via))
-
-	f.Setup("install source", func(ctx context.Context, t feature.T) {
-		u, err := broker.Address(ctx, brokerName)
-		if err != nil || u == nil {
-			t.Error("failed to get the address of the broker", brokerName, err)
-		}
-		eventshub.Install(source, eventshub.StartSenderURL(u.String()), eventshub.InputEvent(event))(ctx, t)
-	})
-
-	f.Stable("broker as middleware").
-		Must("deliver an event",
-			OnStore(sink).MatchEvent(HasId(event.ID())).Exact(1))
-
-	return f
-}
-
-// SourceToSinkWithDLQ tests to see if a Ready Broker acts as middleware.
-//
-// source ---> broker --[trigger]--> bad uri
-//                  |
-//                  +--[DLQ]--> recorder
-//
-func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
-	source := feature.MakeRandomK8sName("source")
-	sink := feature.MakeRandomK8sName("sink")
-	dlq := feature.MakeRandomK8sName("dlq")
-	via := feature.MakeRandomK8sName("via")
-	event := FullEvent()
-
-	f := new(feature.Feature)
-
-	f.Setup("install sink", svc.Install(sink, "bad", "svc"))
-
-	f.Setup("install dlq", eventshub.Install(dlq, eventshub.StartReceiver))
-
-	f.Setup("update broker with DLQ", broker.Install(brokerName, broker.WithDeadLetterSink(svc.AsRef(dlq), "")))
-
-	// Point the Trigger subscriber to the sink svc.
-	cfg := []manifest.CfgFn{trigger.WithSubscriber(svc.AsRef(sink), "")}
-
-	// Install the trigger
-	f.Setup("install trigger", trigger.Install(via, brokerName, cfg...))
-
-	f.Setup("trigger goes ready", trigger.IsReady(via))
-
-	f.Setup("install source", func(ctx context.Context, t feature.T) {
-		u, err := broker.Address(ctx, brokerName)
-		if err != nil || u == nil {
-			t.Error("failed to get the address of the broker", brokerName, err)
-		}
-		eventshub.Install(source, eventshub.StartSenderURL(u.String()), eventshub.InputEvent(event))(ctx, t)
-	})
-
-	f.Stable("broker with DQL").
-		Must("deliver event to DLQ",
-			OnStore(dlq).MatchEvent(HasId(event.ID())).Exact(1))
-
-	return f
-}
-
-// SourceToSinkWithFlakyDLQ tests to see if a Ready Broker acts as middleware.
-//
-// source ---> broker --[trigger]--> flake 1/3 --> recorder
-//                  |
-//                  +--[DLQ]--> recorder
-//
-func SourceToSinkWithFlakyDLQ(brokerName string) *feature.Feature {
-	source := feature.MakeRandomK8sName("source")
-	sink := feature.MakeRandomK8sName("sink")
-	flake := feature.MakeRandomK8sName("flake")
-	dlq := feature.MakeRandomK8sName("dlq")
-	via := feature.MakeRandomK8sName("via")
-
-	uuids := []string{
-		uuid.New().String(),
-		uuid.New().String(),
-		uuid.New().String(),
-	}
-
-	f := new(feature.Feature)
-
-	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
-
-	f.Setup("install flake", func(ctx context.Context, t feature.T) {
-		env := environment.FromContext(ctx)
-		u := fmt.Sprintf("%s.%s.svc.cluster.local", sink, env.Namespace()) // HACK HACK HACK, could replace with SinkBinding.
-		flaker.Install(flake, u)(ctx, t)
-	})
-
-	f.Setup("install dlq", eventshub.Install(dlq, eventshub.StartReceiver))
-
-	f.Setup("update broker with DLQ", broker.Install(brokerName, broker.WithDeadLetterSink(svc.AsRef(dlq), "")))
-
-	// Point the Trigger subscriber to the sink svc.
-	cfg := []manifest.CfgFn{trigger.WithSubscriber(flaker.AsRef(flake), "")}
-
-	// Install the trigger
-	f.Setup("install trigger", trigger.Install(via, brokerName, cfg...))
-
-	f.Setup("trigger goes ready", trigger.IsReady(via))
-
-	f.Setup("install source", func(ctx context.Context, t feature.T) {
-		u, err := broker.Address(ctx, brokerName)
-		if err != nil || u == nil {
-			t.Error("failed to get the address of the broker", brokerName, err)
-		}
-
-		opts := []eventshub.EventsHubOption{eventshub.StartSenderURL(u.String())}
-		for _, id := range uuids {
-			event := FullEvent()
-			event.SetID(id)
-			opts = append(opts, eventshub.InputEvent(event))
-		}
-		eventshub.Install(source, opts...)(ctx, t)
-	})
-
-	f.Stable("broker with DQL").
-		Must("deliver event flaky sent to DLQ event[0]",
-			OnStore(dlq).MatchEvent(HasId(uuids[0])).Exact(1)).
-		Must("deliver event flaky sent to DLQ event[1]",
-			OnStore(dlq).MatchEvent(HasId(uuids[1])).Exact(1)).
-		Must("deliver event sink receiver got event[2]",
-			OnStore(sink).MatchEvent(HasId(uuids[2])).Exact(1))
-
-	return f
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/knconf/control_plane.go b/vendor/knative.dev/eventing/test/rekt/features/knconf/control_plane.go
deleted file mode 100644
index 85078bd32..000000000
--- a/vendor/knative.dev/eventing/test/rekt/features/knconf/control_plane.go
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
-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 knconf
-
-import (
-	"context"
-	"strings"
-
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/reconciler-test/pkg/feature"
-)
-
-func HasReadyInConditions(_ context.Context, t feature.T, status duckv1.Status) {
-	found := []string(nil)
-	for _, c := range status.Conditions {
-		if c.Type == "Ready" {
-			// Success!
-			return
-		}
-		found = append(found, string(c.Type))
-	}
-	t.Errorf(`does not have "Ready" condition, has: [%s]`, strings.Join(found, ","))
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/flaker/flaker.go b/vendor/knative.dev/eventing/test/rekt/resources/flaker/flaker.go
deleted file mode 100644
index 5c910ae75..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/flaker/flaker.go
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
-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 flaker
-
-import (
-	"context"
-
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/reconciler-test/pkg/environment"
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/manifest"
-)
-
-func init() {
-	environment.RegisterPackage(manifest.ImagesLocalYaml()...)
-}
-
-// Install
-func Install(name, sink string) feature.StepFn {
-	cfg := map[string]interface{}{
-		"name": name,
-		"sink": sink,
-	}
-
-	return func(ctx context.Context, t feature.T) {
-		if _, err := manifest.InstallLocalYaml(ctx, cfg); err != nil {
-			t.Fatal(err)
-		}
-	}
-}
-
-// TODO: add With* methods as we need them.
-
-// AsRef returns a KRef for a Service without namespace.
-func AsRef(name string) *duckv1.KReference {
-	return &duckv1.KReference{
-		Kind:       "Service",
-		APIVersion: "v1",
-		Name:       name,
-	}
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/flaker/flaker.yaml b/vendor/knative.dev/eventing/test/rekt/resources/flaker/flaker.yaml
deleted file mode 100644
index 207c3d681..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/flaker/flaker.yaml
+++ /dev/null
@@ -1,43 +0,0 @@
-# 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: Pod
-metadata:
-  name: {{ .name }}
-  namespace: {{ .namespace }}
-  labels:
-    app: flaker-{{ .name }}
-spec:
-  restartPolicy: "Never"
-  containers:
-    - name: flaker
-      image: ko://knative.dev/eventing/test/test_images/event-flaker
-      imagePullPolicy: "IfNotPresent"
-      env:
-        - name: "K_SINK"
-          value: "{{ .sink }}"
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: {{ .name }}
-  namespace: {{ .namespace }}
-spec:
-  selector:
-    app: flaker-{{ .name }}
-  ports:
-    - protocol: TCP
-      port: 80
-      targetPort: 8080
diff --git a/vendor/knative.dev/eventing/test/test_images/event-flaker/main.go b/vendor/knative.dev/eventing/test/test_images/event-flaker/main.go
deleted file mode 100644
index 02f1f86d5..000000000
--- a/vendor/knative.dev/eventing/test/test_images/event-flaker/main.go
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
-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 main
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"os"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	"github.com/kelseyhightower/envconfig"
-	_ "knative.dev/pkg/system/testing"
-)
-
-type envConfig struct {
-	Reject int    `envconfig:"REJECT_MOD" default:"3"`
-	Sink   string `envconfig:"K_SINK" required:"true"`
-}
-
-func main() {
-	var env envConfig
-	if err := envconfig.Process("", &env); err != nil {
-		log.Printf("[ERROR] Failed to process env var: %s", err)
-		os.Exit(1)
-	}
-
-	p, err := cloudevents.NewHTTP(cloudevents.WithTarget(env.Sink))
-	if err != nil {
-		log.Fatalf("failed to create protocol: %s", err.Error())
-	}
-
-	c, err := cloudevents.NewClient(p, cloudevents.WithUUIDs(), cloudevents.WithTimeNow())
-	if err != nil {
-		log.Fatalf("failed to create client: %s", err.Error())
-	}
-
-	count := 0
-
-	if err := c.StartReceiver(context.Background(), func(ctx context.Context, event cloudevents.Event) error {
-		count++
-		if count%env.Reject == 0 {
-			if result := c.Send(context.Background(), event); !cloudevents.IsACK(result) {
-				log.Printf("failed to send cloudevent: %s", result.Error())
-			}
-			return nil
-		}
-		return fmt.Errorf("rejected %d", count)
-	}); err != nil {
-		log.Fatalf("failed to start receiver: %s", err.Error())
-	}
-}