Repository: knative-sandbox/eventing-kafka-broker
Pull request #2931 (2023-01-31 08:36:41)

Migrate `TestBrokerTrigger` and `TestBrokerWithConfig` broker e2e test to new rekt test
This is the first of a series of PRs to migrate the e2e tests in `test/e2e_broker` to the new reconciler-test framework.

## Proposed Changes
- :broom: Migrate `TestBrokerWithConfig` test

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2931#" title="Author self-approved">creydr</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign pierdipi for approval by writing `/assign @pierdipi` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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[bot]: @creydr: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 3bb6fee2adb3af7fbb818509451e2626af217607 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2931/unit-tests_eventing-kafka-broker_main/1620340291112275968) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2931](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (3bb6fee) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/ef3c197aa9a255ab9aeb1224ee3f1ef24897ac43?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ef3c197) will **decrease** coverage by `5.23%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 3bb6fee differs from pull request most recent head 0cd59f6. Consider uploading reports for the commit 0cd59f6 to get more accurate results

```diff
@@             Coverage Diff              @@
##               main    #2931      +/-   ##
============================================
- Coverage     63.79%   58.57%   -5.23%     
============================================
  Files           155       78      -77     
  Lines         10928     8254    -2674     
  Branches        231        0     -231     
============================================
- Hits           6972     4835    -2137     
+ Misses         3450     3054     -396     
+ Partials        506      365     -141     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...security/secrets\_provider\_legacy\_channel\_secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0c19wcm92aWRlcl9sZWdhY3lfY2hhbm5lbF9zZWNyZXQuZ28=) | `25.00% <0.00%> (-11.93%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `52.63% <0.00%> (-1.10%)` | :arrow_down: |
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | `60.96% <0.00%> (-0.82%)` | :arrow_down: |
| [control-plane/pkg/reconciler/source/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvY29udHJvbGxlci5nbw==) | `100.00% <0.00%> (ø)` | |
| [...venting/kafka/broker/receiver/IngressProducer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9JbmdyZXNzUHJvZHVjZXIuamF2YQ==) | | |
| [...ispatcher/impl/http/WebClientCloudEventSender.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9odHRwL1dlYkNsaWVudENsb3VkRXZlbnRTZW5kZXIuamF2YQ==) | | |
| [...her/impl/filter/subscriptionsapi/SuffixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9TdWZmaXhGaWx0ZXIuamF2YQ==) | | |
| [...atcher/impl/filter/subscriptionsapi/AnyFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9BbnlGaWx0ZXIuamF2YQ==) | | |
| [...er/impl/consumer/InvalidCloudEventInterceptor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudEludGVyY2VwdG9yLmphdmE=) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| ... and [76 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2931?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

creydr: ```
Running go test with args: -count=1 -race -tags=e2e -timeout=1h ./test/e2e_new_broker/...
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Setup/Create_broker_config (0.13s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Setup/Install_broker (0.15s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Broker_ready (3.13s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Topic_is_ready (0.06s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Replication_factor (0.04s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Number_of_partitions (0.04s)
PASS test/e2e_new_broker.TestBrokerWithConfig (5.80s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Create_broker_config (0.14s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Install_broker (0.18s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Install_sink (25.67s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Create_trigger (0.15s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Requirement/Send_matching_event (3.57s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Requirement/Send_non-matching_event (3.56s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Broker_ready (0.04s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Trigger_ready (0.04s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Matching_event_received (0.00s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Non_matching_event_is_not_received (20.00s)
PASS test/e2e_new_broker.TestBrokerTrigger (55.41s)
PASS test/e2e_new_broker

DONE 18 tests in 58.421s
Finished run, return code is 0
```
creydr: /assign @aliok @pierDipi @matzew 
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2931#" title="Author self-approved">creydr</a>*
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from aliok by writing `/assign @aliok` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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

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

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

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 99fbbd0fce2b08b48254b0cb239b3c197888dfe3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2931/unit-tests_eventing-kafka-broker_main/1620400561289433088) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
creydr: /test unit-tests
knative-prow[bot]: @creydr: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 99fbbd0fce2b08b48254b0cb239b3c197888dfe3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2931/unit-tests_eventing-kafka-broker_main/1620401433947934720) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
creydr: /test unit-tests
knative-prow[bot]: @creydr: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 99fbbd0fce2b08b48254b0cb239b3c197888dfe3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2931/unit-tests_eventing-kafka-broker_main/1620401433947934720) | true | `/test unit-tests`
integration-tests_eventing-kafka-broker_main | 99fbbd0fce2b08b48254b0cb239b3c197888dfe3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2931/integration-tests_eventing-kafka-broker_main/1620400563801821184) | true | `/test integration-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
creydr: /test integration-tests
creydr: From [reconciler-tests_eventing-kafka-broker_main/1620400568868540416](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2931/reconciler-tests_eventing-kafka-broker_main/1620400568868540416):

```
Running go test with args: -count=1 -race -tags=e2e -timeout=1h ./test/e2e_new_broker/...
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Setup/Create_broker_config (0.14s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Setup/Install_broker (0.19s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Setup/Install_sink (23.89s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Setup/Create_trigger (0.16s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Requirement/Send_event (4.44s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Broker_ready (0.05s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Trigger_ready (0.05s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Event_received (4.00s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Topic_is_ready (0.07s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Replication_factor (0.04s)
PASS test/e2e_new_broker.TestBrokerWithConfig/Trigger_filters_events/Assert/[Stable/MUST]Number_of_partitions (0.04s)
PASS test/e2e_new_broker.TestBrokerWithConfig (35.15s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Create_broker_config (0.14s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Install_broker (0.18s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Install_sink (23.93s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Setup/Create_trigger (0.16s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Requirement/Send_matching_event (4.85s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Requirement/Send_non-matching_event (3.59s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Broker_ready (0.04s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Trigger_ready (0.04s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Matching_event_received (0.00s)
PASS test/e2e_new_broker.TestBrokerTrigger/Trigger_filters_events/Assert/[Stable/MUST]Non_matching_event_is_not_received (20.00s)
PASS test/e2e_new_broker.TestBrokerTrigger (55.16s)
PASS test/e2e_new_broker

DONE 23 tests in 58.356s
```

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index f1b19ce18..e390d8059 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -4,7 +4,6 @@
 aliases:
   async-component-approvers:
   - AngeloDanducci
-  - julz
   - maximilien
   client-wg-leads:
   - dsimansk
@@ -15,7 +14,6 @@ aliases:
   - navidshaikh
   - rhuss
   container-freezer-approvers:
-  - julz
   - psschwei
   control-protocol-approvers:
   - eric-sap
@@ -148,7 +146,6 @@ aliases:
   - dsimansk
   - houshengbo
   - maximilien
-  - omerbensaadon
   - rhuss
   kn-plugin-quickstart-approvers:
   - dsimansk
@@ -198,40 +195,24 @@ aliases:
   - knative-prow-updater-robot
   - knative-test-reporter-robot
   kperf-approvers:
-  - chzhyang
   - maximilien
-  - nader-ziada
   - psschwei
-  - zhanggbj
-  net-certmanager-approvers:
-  - ZhiminXiang
-  - carlisia
-  - tcnghia
+  net-certmanager-approvers: []
   net-contour-approvers:
+  - KauzClay
   - carlisia
   - dprotaso
-  - tcnghia
   net-gateway-api-approvers:
-  - carlisia
+  - dprotaso
   - nak3
-  - tcnghia
-  net-http01-approvers:
-  - carlisia
-  - tcnghia
+  net-http01-approvers: []
   net-istio-approvers:
   - AngeloDanducci
-  - JRBANCEL
-  - ZhiminXiang
-  - carlisia
   - nak3
   - skonto
-  - tcnghia
-  - vagababov
   net-kourier-approvers:
-  - carlisia
   - nak3
   - skonto
-  - tcnghia
   networking-wg-leads: []
   operations-wg-leads:
   - houshengbo
@@ -247,8 +228,6 @@ aliases:
   - upodroid
   productivity-writers:
   - cardil
-  - chaodaiG
-  - coryrc
   - kvmware
   - psschwei
   - upodroid
@@ -263,11 +242,22 @@ aliases:
   - evankanderson
   security-writers:
   - evankanderson
+  serving-approvers:
+  - nak3
+  serving-reviewers:
+  - KauzClay
+  - carlisia
+  - izabelacg
+  - jsanin-vmw
+  - kauana
+  - retocode
+  - skonto
   serving-wg-leads:
   - dprotaso
   - psschwei
   serving-writers:
   - dprotaso
+  - nak3
   - psschwei
   steering-committee:
   - csantanapr
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 03e598d81..60cad486f 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -30,7 +30,6 @@ import (
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
-	v1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/network"
 	"knative.dev/pkg/reconciler"
@@ -301,7 +300,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return err
 	}
 
-	broker.Status.AddressStatus = v1.AddressStatus{}
+	broker.Status.Address = nil
 
 	ingressHost := network.GetServiceHostname(r.Env.IngressName, r.Reconciler.DataPlaneNamespace)
 
diff --git a/go.mod b/go.mod
index e7ec0b93a..9ce7f3e0b 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.18
 require (
 	github.com/Shopify/sarama v1.37.0
 	github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.4.1
-	github.com/cloudevents/sdk-go/v2 v2.12.0
+	github.com/cloudevents/sdk-go/v2 v2.13.0
 	github.com/google/go-cmp v0.5.8
 	github.com/google/uuid v1.3.0
 	github.com/hashicorp/errwrap v1.1.0
@@ -39,10 +39,10 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/google/gofuzz v1.2.0
 	github.com/kedacore/keda/v2 v2.8.1
-	knative.dev/eventing v0.35.1-0.20230110013451-6ef3f1dfd05d
-	knative.dev/hack v0.0.0-20230110013548-af8745e34e08
-	knative.dev/pkg v0.0.0-20230110013450-dc20e472128f
-	knative.dev/reconciler-test v0.0.0-20221222133658-cf080d097e77
+	knative.dev/eventing v0.36.2
+	knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9
+	knative.dev/pkg v0.0.0-20230117181655-247510c00e9d
+	knative.dev/reconciler-test v0.0.0-20230202091130-91d5fb069a48
 	sigs.k8s.io/controller-runtime v0.12.3
 )
 
@@ -70,8 +70,8 @@ require (
 	github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/cloudevents/conformance v0.2.0 // indirect
-	github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.12.0 // indirect
-	github.com/cloudevents/sdk-go/sql/v2 v2.8.0 // indirect
+	github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.13.0 // indirect
+	github.com/cloudevents/sdk-go/sql/v2 v2.13.0 // indirect
 	github.com/eapache/go-resiliency v1.3.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
diff --git a/go.sum b/go.sum
index aeb24d6a8..21b3a6190 100644
--- a/go.sum
+++ b/go.sum
@@ -153,15 +153,15 @@ github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMn
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudevents/conformance v0.2.0 h1:NvSXOKlagcsOWMEbi8U7Ex/0oQ4JZE1HQ45bnxYf2zk=
 github.com/cloudevents/conformance v0.2.0/go.mod h1:rHKDwylBH89Rns6U3wL9ww8bg9/4GbwRCDNuyoC6bcc=
-github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.12.0 h1:iMJy7/VX/+/ZImJo7dffvg/ZPwrp4PprnnyKPK1mOok=
-github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.12.0/go.mod h1:g7VsRXXYILOchM36AReyfd2bJFJyyE+PMuYa65CxjGo=
+github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.13.0 h1:Mf5y5GYVusfOpPQsKHOvr9c3Y76fZnSZzuZo+LQr/aU=
+github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.13.0/go.mod h1:vgBrMXc1h8htR8PUlGViBcNEkri4fw98nY8Tqsgdtfs=
 github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.4.1 h1:OdBCW2/cxHaGzY9JI8wfFPI0OcmpZ8y97e1ZxWNsVFc=
 github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.4.1/go.mod h1:8aqqE69pYzEgN2sCa6+8fTQ+FpE7QMwYnLkDKyKVtYw=
 github.com/cloudevents/sdk-go/sql/v2 v2.0.0-20220930150014-52b12276cc4a h1:jwFoLdIzNYc92s2ZwhU4yOdoPP8zoALcKOzenIALHJM=
 github.com/cloudevents/sdk-go/sql/v2 v2.0.0-20220930150014-52b12276cc4a/go.mod h1:SPV4fdKh1YqSNn2TTUVqUcw1El9Er0HI/yzay5wqsFk=
 github.com/cloudevents/sdk-go/v2 v2.4.1/go.mod h1:MZiMwmAh5tGj+fPFvtHv9hKurKqXtdB9haJYMJ/7GJY=
-github.com/cloudevents/sdk-go/v2 v2.12.0 h1:p1k+ysVOZtNiXfijnwB3WqZNA3y2cGOiKQygWkUHCEI=
-github.com/cloudevents/sdk-go/v2 v2.12.0/go.mod h1:xDmKfzNjM8gBvjaF8ijFjM1VYOVUEeUfapHMUX1T5To=
+github.com/cloudevents/sdk-go/v2 v2.13.0 h1:2zxDS8RyY1/wVPULGGbdgniGXSzLaRJVl136fLXGsYw=
+github.com/cloudevents/sdk-go/v2 v2.13.0/go.mod h1:xDmKfzNjM8gBvjaF8ijFjM1VYOVUEeUfapHMUX1T5To=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
@@ -1514,14 +1514,14 @@ k8s.io/utils v0.0.0-20200912215256-4140de9c8800/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2 h1:GfD9OzL11kvZN5iArC6oTS7RTj7oJOIfnislxYlqTj8=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-knative.dev/eventing v0.35.1-0.20230110013451-6ef3f1dfd05d h1:t9uKLiKoyLzjLPR3f0YNjUIEPA0qtjNn/F4t2wEO2ic=
-knative.dev/eventing v0.35.1-0.20230110013451-6ef3f1dfd05d/go.mod h1:Z7U8AjFpxwCVubg5ri1evHAGenl090PHtezRJpjxMko=
-knative.dev/hack v0.0.0-20230110013548-af8745e34e08 h1:X+iFjElmlVCizUG+ZNLpH7bky5G24fUphR++5DHAfC0=
-knative.dev/hack v0.0.0-20230110013548-af8745e34e08/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
-knative.dev/pkg v0.0.0-20230110013450-dc20e472128f h1:vORnIzNNy3efqsjBYwkHw1IZf/mC3J4sFfhurtJosh8=
-knative.dev/pkg v0.0.0-20230110013450-dc20e472128f/go.mod h1:IeUSNPPUpQnM35SjpnfCx0w5/V2RpEc+nmke6oPwpD0=
-knative.dev/reconciler-test v0.0.0-20221222133658-cf080d097e77 h1:75JTbBbRmw3exnxFkKQTYzd3XEqq2mCG+MKXUe3+7Ss=
-knative.dev/reconciler-test v0.0.0-20221222133658-cf080d097e77/go.mod h1:R0/G3KxvSgKUrfNWU6nVG/0WipBhfjt7K+Fx9dTmFRQ=
+knative.dev/eventing v0.36.2 h1:XrEnMUUTveZ/zEsikVggwC6J06pnrcMvrX3RfNwF8WE=
+knative.dev/eventing v0.36.2/go.mod h1:Qka5Z6+LeMoHGL1QAznVdmq5LAu21b4F3rgxc2AMgRg=
+knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9 h1:CDa7s9KspEZqPhk7cN68ZypRLuAvSgr+knoOaXSsrHk=
+knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
+knative.dev/pkg v0.0.0-20230117181655-247510c00e9d h1:pjKDcvHoMib8nRp56eISRmMj/pFMzJljnzvMvGCIReI=
+knative.dev/pkg v0.0.0-20230117181655-247510c00e9d/go.mod h1:VO/fcEsq43seuONRQxZyftWHjpMabYzRHDtpSEQ/eoQ=
+knative.dev/reconciler-test v0.0.0-20230202091130-91d5fb069a48 h1:keolT8Eh8T/z+IyoXAtsxQmryrJNfEtktTbEtkfjr9Y=
+knative.dev/reconciler-test v0.0.0-20230202091130-91d5fb069a48/go.mod h1:ZPUBldkrSIOe0c57sDcnuM4bzQy16iTtvMV59nMPpfs=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/hack/tools.go b/hack/tools.go
index d7b7b0f4b..b2a23d8e2 100644
--- a/hack/tools.go
+++ b/hack/tools.go
@@ -29,8 +29,6 @@ import (
 
 	// Test images from eventing
 	_ "knative.dev/eventing/test/test_images/event-flaker"
-	_ "knative.dev/eventing/test/test_images/event-library"
-	_ "knative.dev/eventing/test/test_images/event-library/kodata/events"
 	_ "knative.dev/eventing/test/test_images/event-sender"
 	_ "knative.dev/eventing/test/test_images/heartbeats"
 	_ "knative.dev/eventing/test/test_images/performance"
diff --git a/vendor/github.com/cloudevents/sdk-go/v2/alias.go b/vendor/github.com/cloudevents/sdk-go/v2/alias.go
index a30980a4b..ed64b4c0c 100644
--- a/vendor/github.com/cloudevents/sdk-go/v2/alias.go
+++ b/vendor/github.com/cloudevents/sdk-go/v2/alias.go
@@ -21,7 +21,7 @@ import (
 
 // Client
 
-type ClientOption client.Option
+type ClientOption = client.Option
 type Client = client.Client
 
 // Event
@@ -42,7 +42,7 @@ type URIRef = types.URIRef
 
 // HTTP Protocol
 
-type HTTPOption http.Option
+type HTTPOption = http.Option
 
 type HTTPProtocol = http.Protocol
 
diff --git a/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/abuse_protection.go b/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/abuse_protection.go
index 89222a20c..48f03fb6c 100644
--- a/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/abuse_protection.go
+++ b/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/abuse_protection.go
@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 )
 
 type WebhookConfig struct {
@@ -23,6 +24,7 @@ type WebhookConfig struct {
 
 const (
 	DefaultAllowedRate = 1000
+	DefaultTimeout     = time.Second * 600
 )
 
 // TODO: implement rate limiting.
diff --git a/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/protocol_lifecycle.go b/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/protocol_lifecycle.go
index dacfd30f6..04ef96915 100644
--- a/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/protocol_lifecycle.go
+++ b/vendor/github.com/cloudevents/sdk-go/v2/protocol/http/protocol_lifecycle.go
@@ -38,8 +38,10 @@ func (p *Protocol) OpenInbound(ctx context.Context) error {
 	}
 
 	p.server = &http.Server{
-		Addr:    listener.Addr().String(),
-		Handler: attachMiddleware(p.Handler, p.middleware),
+		Addr:         listener.Addr().String(),
+		Handler:      attachMiddleware(p.Handler, p.middleware),
+		ReadTimeout:  DefaultTimeout,
+		WriteTimeout: DefaultTimeout,
 	}
 
 	// Shutdown
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/apiserver_types.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/apiserver_types.go
index b70516d49..cfe41a956 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1/apiserver_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/apiserver_types.go
@@ -80,6 +80,11 @@ type ApiServerSourceSpec struct {
 	// source. Defaults to default if not set.
 	// +optional
 	ServiceAccountName string `json:"serviceAccountName,omitempty"`
+
+	// NamespaceSelector is a label selector to capture the namespaces that
+	// should be watched by the source.
+	// +optional
+	NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty"`
 }
 
 // ApiServerSourceStatus defines the observed state of ApiServerSource
@@ -92,6 +97,9 @@ type ApiServerSourceStatus struct {
 	// * SinkURI - the current active sink URI that has been configured for the
 	//   Source.
 	duckv1.SourceStatus `json:",inline"`
+
+	// Namespaces show the namespaces currently watched by the ApiServerSource
+	Namespaces []string `json:"namespaces"`
 }
 
 // APIVersionKind is an APIVersion and Kind tuple.
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go
index fd7b8b717..551322eab 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1/zz_generated.deepcopy.go
@@ -140,6 +140,11 @@ func (in *ApiServerSourceSpec) DeepCopyInto(out *ApiServerSourceSpec) {
 		*out = new(APIVersionKind)
 		**out = **in
 	}
+	if in.NamespaceSelector != nil {
+		in, out := &in.NamespaceSelector, &out.NamespaceSelector
+		*out = new(metav1.LabelSelector)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -157,6 +162,11 @@ func (in *ApiServerSourceSpec) DeepCopy() *ApiServerSourceSpec {
 func (in *ApiServerSourceStatus) DeepCopyInto(out *ApiServerSourceStatus) {
 	*out = *in
 	in.SourceStatus.DeepCopyInto(&out.SourceStatus)
+	if in.Namespaces != nil {
+		in, out := &in.Namespaces, &out.Namespaces
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
index df0dcfdf7..556c22080 100644
--- a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
+++ b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
@@ -173,7 +173,7 @@ func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1
 	// Quite an expensive operation but safe and simple.
 	state, err := s.stateAccessor.State(s.reserved)
 	if err != nil {
-		logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+		logger.Debug("error while refreshing scheduler state (will retry)", zap.Error(err))
 		return nil, err
 	}
 
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 645d1a6bd..f6a3818c4 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -65,14 +65,14 @@ github.com/cespare/xxhash/v2
 ## explicit; go 1.14
 github.com/cloudevents/conformance/pkg/event
 github.com/cloudevents/conformance/pkg/http
-# github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.12.0
+# github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.13.0
 ## explicit; go 1.17
 github.com/cloudevents/sdk-go/observability/opencensus/v2/client
 github.com/cloudevents/sdk-go/observability/opencensus/v2/http
 # github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.4.1
 ## explicit; go 1.14
 github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2
-# github.com/cloudevents/sdk-go/sql/v2 v2.8.0 => github.com/cloudevents/sdk-go/sql/v2 v2.0.0-20220930150014-52b12276cc4a
+# github.com/cloudevents/sdk-go/sql/v2 v2.13.0 => github.com/cloudevents/sdk-go/sql/v2 v2.0.0-20220930150014-52b12276cc4a
 ## explicit; go 1.17
 github.com/cloudevents/sdk-go/sql/v2
 github.com/cloudevents/sdk-go/sql/v2/expression
@@ -81,7 +81,7 @@ github.com/cloudevents/sdk-go/sql/v2/gen
 github.com/cloudevents/sdk-go/sql/v2/parser
 github.com/cloudevents/sdk-go/sql/v2/runtime
 github.com/cloudevents/sdk-go/sql/v2/utils
-# github.com/cloudevents/sdk-go/v2 v2.12.0
+# github.com/cloudevents/sdk-go/v2 v2.13.0
 ## explicit; go 1.17
 github.com/cloudevents/sdk-go/v2
 github.com/cloudevents/sdk-go/v2/binding
@@ -1221,7 +1221,7 @@ k8s.io/utils/net
 k8s.io/utils/pointer
 k8s.io/utils/strings/slices
 k8s.io/utils/trace
-# knative.dev/eventing v0.35.1-0.20230110013451-6ef3f1dfd05d
+# knative.dev/eventing v0.36.2
 ## explicit; go 1.18
 knative.dev/eventing/cmd/event_display
 knative.dev/eventing/pkg/apis/config
@@ -1343,8 +1343,6 @@ knative.dev/eventing/test/rekt/resources/channel
 knative.dev/eventing/test/rekt/resources/channel_impl
 knative.dev/eventing/test/rekt/resources/containersource
 knative.dev/eventing/test/rekt/resources/delivery
-knative.dev/eventing/test/rekt/resources/eventlibrary
-knative.dev/eventing/test/rekt/resources/eventlibrary/events
 knative.dev/eventing/test/rekt/resources/flaker
 knative.dev/eventing/test/rekt/resources/pingsource
 knative.dev/eventing/test/rekt/resources/source
@@ -1353,8 +1351,6 @@ 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-library
-knative.dev/eventing/test/test_images/event-library/kodata/events
 knative.dev/eventing/test/test_images/event-sender
 knative.dev/eventing/test/test_images/heartbeats
 knative.dev/eventing/test/test_images/performance
@@ -1374,11 +1370,11 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/hack v0.0.0-20230110013548-af8745e34e08
+# knative.dev/hack v0.0.0-20230113013652-c7cfcb062de9
 ## explicit; go 1.18
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20230110013450-dc20e472128f
+# knative.dev/pkg v0.0.0-20230117181655-247510c00e9d
 ## explicit; go 1.18
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1461,6 +1457,7 @@ knative.dev/pkg/test/ghutil
 knative.dev/pkg/test/helpers
 knative.dev/pkg/test/ingress
 knative.dev/pkg/test/logging
+knative.dev/pkg/test/logstream/v2
 knative.dev/pkg/test/mako
 knative.dev/pkg/test/mako/alerter
 knative.dev/pkg/test/mako/alerter/github
@@ -1486,7 +1483,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20221222133658-cf080d097e77
+# knative.dev/reconciler-test v0.0.0-20230202091130-91d5fb069a48
 ## explicit; go 1.18
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment
@@ -1507,6 +1504,7 @@ knative.dev/reconciler-test/pkg/knative
 knative.dev/reconciler-test/pkg/logging
 knative.dev/reconciler-test/pkg/manifest
 knative.dev/reconciler-test/pkg/milestone
+knative.dev/reconciler-test/pkg/resources/job
 knative.dev/reconciler-test/pkg/state
 knative.dev/reconciler-test/resources/svc
 # sigs.k8s.io/controller-runtime v0.12.3


Test patch:
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 188b27a23..c94d3ea45 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -330,7 +330,7 @@ func NamespacedBrokerAddressable(env *config.Env) func(broker *eventing.Broker)
 }
 
 func brokerAddressable(broker *eventing.Broker, serviceName, serviceNamespace string) {
-	broker.Status.AddressStatus.Address = &duckv1.Addressable{
+	broker.Status.Address = &duckv1.Addressable{
 		URL: &apis.URL{
 			Scheme: "http",
 			Host:   network.GetServiceHostname(serviceName, serviceNamespace),
diff --git a/test/e2e_broker/broker_trigger_sink_test.go b/test/e2e_broker/broker_trigger_sink_test.go
deleted file mode 100644
index c9ddfcba0..000000000
--- a/test/e2e_broker/broker_trigger_sink_test.go
+++ /dev/null
@@ -1,136 +0,0 @@
-//go:build e2e
-// +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_broker
-
-import (
-	"context"
-	"testing"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	"github.com/stretchr/testify/assert"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/utils/pointer"
-	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/resources"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-
-	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
-	eventingv1alpha1clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
-	"knative.dev/eventing-kafka-broker/test/pkg/addressable"
-	brokertest "knative.dev/eventing-kafka-broker/test/pkg/broker"
-	"knative.dev/eventing-kafka-broker/test/pkg/sink"
-)
-
-/*
-+---------+     +--------+     +---------+   +---------------------------+
-|  Broker +---->+ Trigger+---->+KafkaSink+-->+kafka consumer (test image)|
-+----+----+     +--------+     +----+----+   +---------------------------+
-
-	|                              ^
-	|          +--------+          |
-	+--------->+ Trigger+----------+
-	           +--------+
-*/
-func TestBrokerV1TriggersV1SinkV1Alpha1(t *testing.T) {
-	testingpkg.RunMultipleN(t, 10, func(t *testing.T) {
-
-		ctx := context.Background()
-
-		client := testlib.Setup(t, true)
-		defer testlib.TearDown(client)
-
-		clientSet, err := eventingv1alpha1clientset.NewForConfig(client.Config)
-		assert.Nil(t, err)
-
-		// Create a KafkaSink with the following spec.
-		kss := eventingv1alpha1.KafkaSinkSpec{
-			Topic:             "kafka-sink-" + client.Namespace,
-			NumPartitions:     pointer.Int32(10),
-			ReplicationFactor: func(rf int16) *int16 { return &rf }(1),
-			BootstrapServers:  testingpkg.BootstrapServersPlaintextArr,
-		}
-
-		createFunc := sink.CreatorV1Alpha1(clientSet, kss)
-
-		kafkaSink, err := createFunc(types.NamespacedName{
-			Namespace: client.Namespace,
-			Name:      "kafka-sink",
-		})
-		assert.Nil(t, err)
-
-		client.WaitForResourceReadyOrFail(kafkaSink.Name, &kafkaSink.TypeMeta)
-
-		// Create a Kafka Broker
-		brokerName := brokertest.Creator(client, "v1")
-
-		// Create 2 Triggers with the same subscriber
-
-		sinkReference := &duckv1.KReference{
-			Kind:       eventingv1alpha1.Kind("KafkaSink").Kind,
-			APIVersion: eventingv1alpha1.SchemeGroupVersion.String(),
-			Namespace:  client.Namespace,
-			Name:       "kafka-sink",
-		}
-
-		client.CreateTriggerOrFail(
-			"trigger-1",
-			resources.WithBroker(brokerName),
-			resources.WithAttributesTriggerFilter("", "trigger1", nil),
-			func(trigger *eventing.Trigger) {
-				trigger.Spec.Subscriber.Ref = sinkReference
-			},
-		)
-
-		client.CreateTriggerOrFail(
-			"trigger-2",
-			resources.WithBroker(brokerName),
-			resources.WithAttributesTriggerFilter("", "trigger2", nil),
-			func(trigger *eventing.Trigger) {
-				trigger.Spec.Subscriber.Ref = sinkReference
-			},
-		)
-
-		client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-		brokerAddressable := addressable.Addressable{
-			NamespacedName: types.NamespacedName{
-				Namespace: client.Namespace,
-				Name:      brokerName,
-			},
-			TypeMeta: *testlib.BrokerTypeMeta,
-		}
-
-		// Send events to the Broker for both triggers.
-
-		setTypeMutator := func(t string) addressable.EventMutator {
-			return func(event *cloudevents.Event) {
-				event.SetType(t)
-			}
-		}
-
-		idsTrigger1 := addressable.Send(t, brokerAddressable, setTypeMutator("trigger1"))
-		idsTrigger2 := addressable.Send(t, brokerAddressable, setTypeMutator("trigger2"))
-
-		// Read events from the topic.
-		sink.Verify(t, client, eventingv1alpha1.ModeStructured, kss.Topic, append(idsTrigger1, idsTrigger2...))
-	})
-}
diff --git a/test/e2e_broker/broker_trigger_test.go b/test/e2e_broker/broker_trigger_test.go
deleted file mode 100644
index 0191e4b23..000000000
--- a/test/e2e_broker/broker_trigger_test.go
+++ /dev/null
@@ -1,273 +0,0 @@
-//go:build e2e
-// +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_broker
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	. "github.com/cloudevents/sdk-go/v2/test"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	"github.com/google/uuid"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
-	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
-	brokertest "knative.dev/eventing-kafka-broker/test/pkg/broker"
-	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
-	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/recordevents"
-	"knative.dev/eventing/test/lib/resources"
-)
-
-func TestBrokerTrigger(t *testing.T) {
-	testingpkg.RunMultiple(t, func(t *testing.T) {
-
-		ctx := context.Background()
-
-		client := testlib.Setup(t, true)
-		defer testlib.TearDown(client)
-
-		const (
-			senderName  = "sender"
-			triggerName = "trigger"
-			subscriber  = "subscriber"
-
-			defaultNumPartitions     = 10
-			defaultReplicationFactor = 3
-			verifierName             = "num-partitions-replication-factor-verifier"
-
-			eventType       = "type1"
-			eventSource     = "source1"
-			eventBody       = `{"msg":"e2e-eventtransformation-body"}`
-			extension1      = "ext1"
-			valueExtension1 = "value1"
-		)
-
-		nonMatchingEventId := uuid.New().String()
-		eventId := uuid.New().String()
-
-		brokerName := brokertest.Creator(client, "v1")
-
-		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, subscriber)
-
-		client.CreateTriggerOrFail(
-			triggerName,
-			resources.WithBroker(brokerName),
-			resources.WithSubscriberServiceRefForTrigger(subscriber),
-			func(trigger *eventing.Trigger) {
-				trigger.Spec.Filter = &eventing.TriggerFilter{
-					Attributes: map[string]string{
-						"source":   eventSource,
-						extension1: valueExtension1,
-						"type":     "",
-					},
-				}
-			},
-		)
-
-		client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-		t.Logf("Sending events to %s/%s", client.Namespace, brokerName)
-
-		nonMatchingEvent := cloudevents.NewEvent()
-		nonMatchingEvent.SetID(eventId)
-		nonMatchingEvent.SetType(eventType)
-		nonMatchingEvent.SetSource(eventSource)
-		nonMatchingEvent.SetExtension(extension1, valueExtension1+"a")
-		if err := nonMatchingEvent.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
-			t.Fatalf("Cannot set the payload of the event: %s", err.Error())
-		}
-
-		client.SendEventToAddressable(
-			ctx,
-			senderName+"non-matching",
-			brokerName,
-			testlib.BrokerTypeMeta,
-			nonMatchingEvent,
-		)
-
-		eventToSend := cloudevents.NewEvent()
-		eventToSend.SetID(eventId)
-		eventToSend.SetType(eventType)
-		eventToSend.SetSource(eventSource)
-		eventToSend.SetExtension(extension1, valueExtension1)
-		if err := eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
-			t.Fatalf("Cannot set the payload of the event: %s", err.Error())
-		}
-
-		client.SendEventToAddressable(
-			ctx,
-			senderName+"matching",
-			brokerName,
-			testlib.BrokerTypeMeta,
-			eventToSend,
-		)
-
-		eventTracker.AssertAtLeast(1, recordevents.MatchEvent(
-			HasId(eventId),
-			HasSource(eventSource),
-			HasType(eventType),
-			HasData([]byte(eventBody)),
-		))
-
-		eventTracker.AssertNot(recordevents.MatchEvent(
-			HasId(nonMatchingEventId),
-		))
-
-		config := &kafkatest.Config{
-			BootstrapServers:  testingpkg.BootstrapServersPlaintext,
-			ReplicationFactor: defaultReplicationFactor,
-			NumPartitions:     defaultNumPartitions,
-			Topic: kafka.BrokerTopic(broker.TopicPrefix, &eventing.Broker{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      brokerName,
-					Namespace: client.Namespace,
-				},
-			}),
-		}
-
-		err := kafkatest.VerifyNumPartitionAndReplicationFactor(
-			client.Kube,
-			client.Tracker,
-			client.Namespace,
-			verifierName,
-			config,
-		)
-		if err != nil {
-			t.Errorf("failed to verify num partitions and replication factors: %v", err)
-		}
-	})
-}
-
-func TestBrokerWithConfig(t *testing.T) {
-	testingpkg.RunMultiple(t, func(t *testing.T) {
-
-		ctx := context.Background()
-
-		client := testlib.Setup(t, true)
-		defer testlib.TearDown(client)
-
-		const (
-			configMapName = "br-config"
-			senderName    = "sender"
-			triggerName   = "trigger"
-			subscriber    = "subscriber"
-
-			numPartitions     = 20
-			replicationFactor = 1
-			verifierName      = "num-partitions-replication-factor-verifier"
-
-			eventType       = "type1"
-			eventSource     = "source1"
-			eventBody       = `{"msg":"e2e-body"}`
-			extension1      = "ext1"
-			valueExtension1 = "value1"
-		)
-
-		eventId := uuid.New().String()
-
-		brokerName := brokertest.CreatorWithOptions(client, "v1",
-			[]resources.BrokerOption{
-				brokertest.WithBrokerClassFromEnvVar,
-			},
-			[]brokertest.ConfigOptions{
-				func(data map[string]string) {
-					data[kafka.DefaultTopicNumPartitionConfigMapKey] = fmt.Sprintf("%d", numPartitions)
-					data[kafka.DefaultTopicReplicationFactorConfigMapKey] = fmt.Sprintf("%d", replicationFactor)
-					data[kafka.BootstrapServersConfigMapKey] = testingpkg.BootstrapServersPlaintext
-				},
-			})
-
-		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, subscriber)
-
-		client.CreateTriggerOrFail(
-			triggerName,
-			resources.WithBroker(brokerName),
-			resources.WithSubscriberServiceRefForTrigger(subscriber),
-			func(trigger *eventing.Trigger) {
-				trigger.Spec.Filter = &eventing.TriggerFilter{
-					Attributes: map[string]string{
-						"source":   eventSource,
-						extension1: valueExtension1,
-						"type":     "",
-					},
-				}
-			},
-		)
-
-		client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-		t.Logf("Sending events to %s/%s", client.Namespace, brokerName)
-
-		eventToSend := cloudevents.NewEvent()
-		eventToSend.SetID(eventId)
-		eventToSend.SetType(eventType)
-		eventToSend.SetSource(eventSource)
-		eventToSend.SetExtension(extension1, valueExtension1)
-		if err := eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody)); err != nil {
-			t.Fatalf("Cannot set the payload of the event: %s", err.Error())
-		}
-
-		client.SendEventToAddressable(
-			ctx,
-			senderName+"matching",
-			brokerName,
-			testlib.BrokerTypeMeta,
-			eventToSend,
-		)
-
-		eventTracker.AssertAtLeast(1, recordevents.MatchEvent(
-			HasId(eventId),
-			HasSource(eventSource),
-			HasType(eventType),
-			HasData([]byte(eventBody)),
-		))
-
-		t.Logf("Verify num partitions and replication factor")
-
-		config := &kafkatest.Config{
-			BootstrapServers:  testingpkg.BootstrapServersPlaintext,
-			ReplicationFactor: replicationFactor,
-			NumPartitions:     numPartitions,
-			Topic: kafka.BrokerTopic(broker.TopicPrefix, &eventing.Broker{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      brokerName,
-					Namespace: client.Namespace,
-				},
-			}),
-		}
-
-		err := kafkatest.VerifyNumPartitionAndReplicationFactor(
-			client.Kube,
-			client.Tracker,
-			client.Namespace,
-			verifierName,
-			config,
-		)
-		if err != nil {
-			t.Errorf("failed to verify num partitions and replication factors: %v", err)
-		}
-	})
-}
diff --git a/test/e2e_new/broker_config_test.go b/test/e2e_new/broker_config_test.go
new file mode 100644
index 000000000..f5a42ebaa
--- /dev/null
+++ b/test/e2e_new/broker_config_test.go
@@ -0,0 +1,45 @@
+//go:build e2e
+// +build e2e
+
+/*
+ * Copyright 2023 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/test/rekt/features"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+// TestBrokerWithConfig tests a brokers values for ReplicationFactor and number
+// of partitions if set in the broker configmap.
+func TestBrokerWithConfig(t *testing.T) {
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+	)
+
+	env.Test(ctx, t, features.BrokerWithCustomReplicationFactorAndNumPartitions(env))
+}
diff --git a/test/e2e_new/broker_conformance_test.go b/test/e2e_new/broker_conformance_test.go
index 051a6085b..89a078d69 100644
--- a/test/e2e_new/broker_conformance_test.go
+++ b/test/e2e_new/broker_conformance_test.go
@@ -22,17 +22,18 @@ package e2e_new
 import (
 	"testing"
 
-	"knative.dev/eventing-kafka-broker/test/rekt/features"
 	"knative.dev/reconciler-test/pkg/feature"
 
+	"knative.dev/eventing-kafka-broker/test/rekt/features"
+
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
 	"knative.dev/pkg/system"
 	_ "knative.dev/pkg/system/testing"
 
-	brokerfeatures "knative.dev/eventing/test/rekt/features/broker"
-	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/features/broker"
+	b "knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
@@ -52,14 +53,10 @@ func TestBrokerConformance(t *testing.T) {
 	env.Prerequisite(ctx, t, features.BrokerConfigmapCreated(configName))
 
 	// Install and wait for a Ready Broker.
-	env.Prerequisite(ctx, t,
-		brokerfeatures.GoesReady(
-			"default",
-			broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
-			broker.WithConfig(configName),
-		),
-	)
+	cfg := b.WithEnvConfig()
+	cfg = append(cfg, b.WithConfig(configName))
+	env.Prerequisite(ctx, t, broker.GoesReady("default", cfg...))
 
-	env.TestSet(ctx, t, brokerfeatures.ControlPlaneConformance("default", broker.WithConfig(configName)))
-	env.TestSet(ctx, t, brokerfeatures.DataPlaneConformance("default"))
+	env.TestSet(ctx, t, broker.ControlPlaneConformance("default", cfg...))
+	env.TestSet(ctx, t, broker.DataPlaneConformance("default"))
 }
diff --git a/test/e2e_new/broker_delete_config_map_test.go b/test/e2e_new/broker_delete_config_map_test.go
index 75eef792c..7ba1a8866 100644
--- a/test/e2e_new/broker_delete_config_map_test.go
+++ b/test/e2e_new/broker_delete_config_map_test.go
@@ -82,12 +82,12 @@ func BrokerDeleteContractConfigMap() *feature.Feature {
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
 	// Let's delete the contract config map and then wait for it to be recreated
-	f.Setup("delete config map", configmap.DeleteContract)
-	f.Setup("wait config map", configmap.ExistsContract)
-	f.Setup("trigger is ready after deleting the contract config map", trigger.IsReady(triggerName))
+	f.Requirement("delete config map", configmap.DeleteContract)
+	f.Requirement("wait config map", configmap.ExistsContract)
+	f.Requirement("trigger is ready after deleting the contract config map", trigger.IsReady(triggerName))
 
 	// Let's send a message and check if it's received
-	f.Setup("install source", eventshub.Install(
+	f.Assert("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
diff --git a/test/e2e_new/broker_trigger_sink_test.go b/test/e2e_new/broker_trigger_sink_test.go
new file mode 100644
index 000000000..e0e157177
--- /dev/null
+++ b/test/e2e_new/broker_trigger_sink_test.go
@@ -0,0 +1,55 @@
+//go:build e2e
+// +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/test/rekt/features"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+/*
++---------+     +--------+     +---------+   +---------------------------+
+|  Broker +---->+ Trigger+---->+KafkaSink+-->+kafka consumer (test image)|
++----+----+     +--------+     +----+----+   +---------------------------+
+
+	|                              ^
+	|          +--------+          |
+	+--------->+ Trigger+----------+
+	           +--------+
+*/
+func TestBrokerTriggersSink(t *testing.T) {
+
+	// Run Test In Parallel With Others
+	t.Parallel()
+
+	//Create The Test Context / Environment
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+	)
+
+	env.Test(ctx, t, features.BrokerWithTriggersAndKafkaSink(env))
+}
diff --git a/test/e2e_new/dls_test.go b/test/e2e_new/dls_test.go
index 83c64dff1..efa26b6b5 100644
--- a/test/e2e_new/dls_test.go
+++ b/test/e2e_new/dls_test.go
@@ -102,7 +102,7 @@ func SendsEventWithRetries() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
@@ -167,7 +167,7 @@ func SendsEventErrorWithoutRetries() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
@@ -229,7 +229,7 @@ func SendsEventNoRetries() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEvent(ev),
diff --git a/test/e2e_new/ordered_test.go b/test/e2e_new/ordered_test.go
index 49a687b7f..c9406b58f 100644
--- a/test/e2e_new/ordered_test.go
+++ b/test/e2e_new/ordered_test.go
@@ -97,7 +97,7 @@ func SinglePartitionOrderedDelivery() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEventWithEncoding(ev, cloudevents.EncodingBinary),
@@ -155,21 +155,21 @@ func MultiplePartitionOrderedDelivery() *feature.Feature {
 	))
 	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install source for events keyed 'a'", eventshub.Install(
+	f.Requirement("install source for events keyed 'a'", eventshub.Install(
 		feature.MakeRandomK8sName("source-a"),
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEventWithEncoding(evA, cloudevents.EncodingBinary),
 		eventshub.AddSequence,
 		eventshub.SendMultipleEvents(20, 100*time.Millisecond),
 	))
-	f.Setup("install source for events keyed 'b'", eventshub.Install(
+	f.Requirement("install source for events keyed 'b'", eventshub.Install(
 		feature.MakeRandomK8sName("source-b"),
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEventWithEncoding(evB, cloudevents.EncodingBinary),
 		eventshub.AddSequence,
 		eventshub.SendMultipleEvents(20, 100*time.Millisecond),
 	))
-	f.Setup("install source for events keyed 'c'", eventshub.Install(
+	f.Requirement("install source for events keyed 'c'", eventshub.Install(
 		feature.MakeRandomK8sName("source-c"),
 		eventshub.StartSenderToResource(broker.GVR(), brokerName),
 		eventshub.InputEventWithEncoding(evC, cloudevents.EncodingBinary),
diff --git a/test/e2e_new_channel/kafka_channel_upstream_test.go b/test/e2e_new_channel/kafka_channel_upstream_test.go
index 5e3901f44..0f7ac9d0f 100644
--- a/test/e2e_new_channel/kafka_channel_upstream_test.go
+++ b/test/e2e_new_channel/kafka_channel_upstream_test.go
@@ -35,9 +35,6 @@ import (
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
-
-	// needed for vendoring the test resource files
-	_ "knative.dev/eventing/test/rekt/resources/eventlibrary/events"
 )
 
 // TestChannelConformance
diff --git a/test/kafka/strimzi-cluster-operator.yaml b/test/kafka/strimzi-cluster-operator.yaml
index 15787eb54..78aa1e899 100644
--- a/test/kafka/strimzi-cluster-operator.yaml
+++ b/test/kafka/strimzi-cluster-operator.yaml
@@ -7377,20 +7377,6 @@ roleRef:
   name: strimzi-cluster-operator-global
   apiGroup: rbac.authorization.k8s.io
 
----
-apiVersion: coordination.k8s.io/v1
-kind: Lease
-metadata:
-  name: strimzi-cluster-operator
-  labels:
-    app: strimzi
-spec:
-  acquireTime: "1970-01-01T00:00:01.000000Z"
-  holderIdentity: ""
-  leaseDurationSeconds: 15
-  leaseTransitions: 0
-  renewTime: "1970-01-01T00:00:01.000000Z"
-
 ---
 apiVersion: v1
 kind: ServiceAccount
@@ -9515,7 +9501,7 @@ spec:
             name: strimzi-cluster-operator
       containers:
         - name: strimzi-cluster-operator
-          image: quay.io/strimzi/operator:0.33.0
+          image: quay.io/strimzi/operator:0.33.1
           ports:
             - containerPort: 8080
               name: http
@@ -9536,53 +9522,53 @@ spec:
             - name: STRIMZI_OPERATION_TIMEOUT_MS
               value: "300000"
             - name: STRIMZI_DEFAULT_TLS_SIDECAR_ENTITY_OPERATOR_IMAGE
-              value: quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+              value: quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_DEFAULT_KAFKA_EXPORTER_IMAGE
-              value: quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+              value: quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_DEFAULT_CRUISE_CONTROL_IMAGE
-              value: quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+              value: quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_KAFKA_IMAGES
               value: |
-                3.2.0=quay.io/strimzi/kafka:0.33.0-kafka-3.2.0
-                3.2.1=quay.io/strimzi/kafka:0.33.0-kafka-3.2.1
-                3.2.3=quay.io/strimzi/kafka:0.33.0-kafka-3.2.3
-                3.3.1=quay.io/strimzi/kafka:0.33.0-kafka-3.3.1
-                3.3.2=quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+                3.2.0=quay.io/strimzi/kafka:0.33.1-kafka-3.2.0
+                3.2.1=quay.io/strimzi/kafka:0.33.1-kafka-3.2.1
+                3.2.3=quay.io/strimzi/kafka:0.33.1-kafka-3.2.3
+                3.3.1=quay.io/strimzi/kafka:0.33.1-kafka-3.3.1
+                3.3.2=quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_KAFKA_CONNECT_IMAGES
               value: |
-                3.2.0=quay.io/strimzi/kafka:0.33.0-kafka-3.2.0
-                3.2.1=quay.io/strimzi/kafka:0.33.0-kafka-3.2.1
-                3.2.3=quay.io/strimzi/kafka:0.33.0-kafka-3.2.3
-                3.3.1=quay.io/strimzi/kafka:0.33.0-kafka-3.3.1
-                3.3.2=quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+                3.2.0=quay.io/strimzi/kafka:0.33.1-kafka-3.2.0
+                3.2.1=quay.io/strimzi/kafka:0.33.1-kafka-3.2.1
+                3.2.3=quay.io/strimzi/kafka:0.33.1-kafka-3.2.3
+                3.3.1=quay.io/strimzi/kafka:0.33.1-kafka-3.3.1
+                3.3.2=quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_KAFKA_MIRROR_MAKER_IMAGES
               value: |
-                3.2.0=quay.io/strimzi/kafka:0.33.0-kafka-3.2.0
-                3.2.1=quay.io/strimzi/kafka:0.33.0-kafka-3.2.1
-                3.2.3=quay.io/strimzi/kafka:0.33.0-kafka-3.2.3
-                3.3.1=quay.io/strimzi/kafka:0.33.0-kafka-3.3.1
-                3.3.2=quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+                3.2.0=quay.io/strimzi/kafka:0.33.1-kafka-3.2.0
+                3.2.1=quay.io/strimzi/kafka:0.33.1-kafka-3.2.1
+                3.2.3=quay.io/strimzi/kafka:0.33.1-kafka-3.2.3
+                3.3.1=quay.io/strimzi/kafka:0.33.1-kafka-3.3.1
+                3.3.2=quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_KAFKA_MIRROR_MAKER_2_IMAGES
               value: |
-                3.2.0=quay.io/strimzi/kafka:0.33.0-kafka-3.2.0
-                3.2.1=quay.io/strimzi/kafka:0.33.0-kafka-3.2.1
-                3.2.3=quay.io/strimzi/kafka:0.33.0-kafka-3.2.3
-                3.3.1=quay.io/strimzi/kafka:0.33.0-kafka-3.3.1
-                3.3.2=quay.io/strimzi/kafka:0.33.0-kafka-3.3.2
+                3.2.0=quay.io/strimzi/kafka:0.33.1-kafka-3.2.0
+                3.2.1=quay.io/strimzi/kafka:0.33.1-kafka-3.2.1
+                3.2.3=quay.io/strimzi/kafka:0.33.1-kafka-3.2.3
+                3.3.1=quay.io/strimzi/kafka:0.33.1-kafka-3.3.1
+                3.3.2=quay.io/strimzi/kafka:0.33.1-kafka-3.3.2
             - name: STRIMZI_DEFAULT_TOPIC_OPERATOR_IMAGE
-              value: quay.io/strimzi/operator:0.33.0
+              value: quay.io/strimzi/operator:0.33.1
             - name: STRIMZI_DEFAULT_USER_OPERATOR_IMAGE
-              value: quay.io/strimzi/operator:0.33.0
+              value: quay.io/strimzi/operator:0.33.1
             - name: STRIMZI_DEFAULT_KAFKA_INIT_IMAGE
-              value: quay.io/strimzi/operator:0.33.0
+              value: quay.io/strimzi/operator:0.33.1
             - name: STRIMZI_DEFAULT_KAFKA_BRIDGE_IMAGE
               value: quay.io/strimzi/kafka-bridge:0.24.0
             - name: STRIMZI_DEFAULT_JMXTRANS_IMAGE
-              value: quay.io/strimzi/jmxtrans:0.33.0
+              value: quay.io/strimzi/jmxtrans:0.33.1
             - name: STRIMZI_DEFAULT_KANIKO_EXECUTOR_IMAGE
-              value: quay.io/strimzi/kaniko-executor:0.33.0
+              value: quay.io/strimzi/kaniko-executor:0.33.1
             - name: STRIMZI_DEFAULT_MAVEN_BUILDER
-              value: quay.io/strimzi/maven-builder:0.33.0
+              value: quay.io/strimzi/maven-builder:0.33.1
             - name: STRIMZI_OPERATOR_NAMESPACE
               valueFrom:
                 fieldRef:
diff --git a/test/pkg/kafka/util.go b/test/pkg/kafka/util.go
index 7e114d14d..77f7164e8 100644
--- a/test/pkg/kafka/util.go
+++ b/test/pkg/kafka/util.go
@@ -19,6 +19,7 @@ package kafka
 import (
 	"context"
 	"fmt"
+	"time"
 
 	batchv1 "k8s.io/api/batch/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -29,6 +30,11 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 )
 
+const (
+	interval = 1 * time.Second
+	timeout  = 2 * time.Minute
+)
+
 func verifyJobSucceeded(
 	ctx context.Context,
 	client kubernetes.Interface,
diff --git a/test/pkg/kafka/verify_num_partitions_replication.go b/test/pkg/kafka/verify_num_partitions_replication.go
deleted file mode 100644
index 2707ad21c..000000000
--- a/test/pkg/kafka/verify_num_partitions_replication.go
+++ /dev/null
@@ -1,101 +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 kafka
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	batchv1 "k8s.io/api/batch/v1"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/utils/pointer"
-	testlib "knative.dev/eventing/test/lib"
-	pkgtest "knative.dev/pkg/test"
-)
-
-const (
-	partitionReplicationVerifierImage = "partitions-replication-verifier"
-
-	interval = 1 * time.Second
-	timeout  = 2 * time.Minute
-)
-
-type Config struct {
-	BootstrapServers  string `required:"true" split_words:"true"`
-	ReplicationFactor int16  `required:"true" split_words:"true"`
-	NumPartitions     int32  `required:"true" split_words:"true"`
-	Topic             string `required:"true" split_words:"true"`
-}
-
-func VerifyNumPartitionAndReplicationFactor(
-	client kubernetes.Interface,
-	tracker *testlib.Tracker,
-	namespace string,
-	name string,
-	config *Config) error {
-
-	ctx := context.Background()
-
-	job := &batchv1.Job{
-		TypeMeta: metav1.TypeMeta{},
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: namespace,
-			Name:      name,
-		},
-		Spec: batchv1.JobSpec{
-			BackoffLimit: pointer.Int32(2),
-			Template: corev1.PodTemplateSpec{
-				Spec: corev1.PodSpec{
-					Containers: []corev1.Container{
-						{
-							Name:                     name,
-							Image:                    pkgtest.ImagePath(partitionReplicationVerifierImage),
-							ImagePullPolicy:          corev1.PullIfNotPresent,
-							TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError,
-							Env: []corev1.EnvVar{
-								{
-									Name:  "BOOTSTRAP_SERVERS",
-									Value: config.BootstrapServers,
-								},
-								{
-									Name:  "REPLICATION_FACTOR",
-									Value: fmt.Sprintf("%d", config.ReplicationFactor),
-								},
-								{
-									Name:  "NUM_PARTITIONS",
-									Value: fmt.Sprintf("%d", config.NumPartitions),
-								},
-								{
-									Name:  "TOPIC",
-									Value: config.Topic,
-								},
-							},
-						},
-					},
-					RestartPolicy: "Never",
-				},
-			},
-			TTLSecondsAfterFinished: nil,
-		},
-		Status: batchv1.JobStatus{},
-	}
-	return verifyJobSucceeded(ctx, client, tracker, types.NamespacedName{Namespace: namespace, Name: name}, job)
-}
diff --git a/test/rekt/features/broker_config.go b/test/rekt/features/broker_config.go
new file mode 100644
index 000000000..c799716c5
--- /dev/null
+++ b/test/rekt/features/broker_config.go
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2023 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 features
+
+import (
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	testpkg "knative.dev/eventing-kafka-broker/test/pkg"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+
+	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	brokerconfigmap "knative.dev/eventing-kafka-broker/test/rekt/resources/configmap/broker"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/test/rekt/resources/broker"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func BrokerWithCustomReplicationFactorAndNumPartitions(env environment.Environment) *feature.Feature {
+	const (
+		numPartitions     = 20
+		replicationFactor = 1
+	)
+
+	f := feature.NewFeatureNamed("Broker KafkaTopic config from Broker configmap")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	brokerConfig := feature.MakeRandomK8sName("brokercfg")
+
+	f.Setup("Create broker config", brokerconfigmap.Install(brokerConfig,
+		brokerconfigmap.WithNumPartitions(numPartitions),
+		brokerconfigmap.WithReplicationFactor(replicationFactor),
+		brokerconfigmap.WithBootstrapServer(testpkg.BootstrapServersPlaintext)))
+
+	f.Setup("Install broker", broker.Install(brokerName, append(
+		broker.WithEnvConfig(),
+		broker.WithConfig(brokerConfig))...,
+	))
+	f.Setup("Broker ready", broker.IsReady(brokerName))
+
+	topic := kafka.BrokerTopic(brokerreconciler.TopicPrefix, &eventing.Broker{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      brokerName,
+			Namespace: env.Namespace(),
+		},
+	})
+	f.Setup("Topic is ready", kafkatopic.IsReady(topic))
+
+	f.Assert("Replication factor", kafkatopic.HasReplicationFactor(topic, replicationFactor))
+	f.Assert("Number of partitions", kafkatopic.HasNumPartitions(topic, numPartitions))
+	return f
+}
diff --git a/test/rekt/features/kafka_sink.go b/test/rekt/features/kafka_sink.go
new file mode 100644
index 000000000..2b2739fa2
--- /dev/null
+++ b/test/rekt/features/kafka_sink.go
@@ -0,0 +1,140 @@
+/*
+ * Copyright 2023 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 features
+
+import (
+	"fmt"
+
+	"github.com/cloudevents/sdk-go/v2/test"
+	"github.com/google/uuid"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	testpkg "knative.dev/eventing-kafka-broker/test/pkg"
+	brokerconfigmap "knative.dev/eventing-kafka-broker/test/rekt/resources/configmap/broker"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasink"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"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/resources/job"
+)
+
+const (
+	kafkaConsumerImage = "ko://knative.dev/eventing-kafka-broker/test/test_images/kafka-consumer"
+)
+
+/*
++---------+     +--------+     +---------+   +---------------------------+
+|  Broker +---->+ Trigger+---->+KafkaSink+-->+kafka consumer (test image)|
++----+----+     +--------+     +----+----+   +---------------------------+
+
+	|                              ^
+	|          +--------+          |
+	+--------->+ Trigger+----------+
+	           +--------+
+*/
+func BrokerWithTriggersAndKafkaSink(env environment.Environment) *feature.Feature {
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	brokerConfig := feature.MakeRandomK8sName("brokercfg")
+	trigger1Name := feature.MakeRandomK8sName("trigger-1")
+	trigger2Name := feature.MakeRandomK8sName("trigger-2")
+	sink := feature.MakeRandomK8sName("sink")
+	verifyMessagesJobName := feature.MakeRandomK8sName("verify-messages-job")
+
+	topic := kafka.BrokerTopic(brokerreconciler.TopicPrefix, &eventingv1.Broker{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      brokerName,
+			Namespace: env.Namespace(),
+		},
+	})
+
+	trigger1FilterType := "trigger-1"
+	trigger2FilterType := "trigger-2"
+
+	eventIdTrigger1 := uuid.New().String()
+	eventTrigger1 := test.MinEvent()
+	eventTrigger1.SetID(eventIdTrigger1)
+	eventTrigger1.SetType(trigger1FilterType)
+
+	eventIdTrigger2 := uuid.New().String()
+	eventTrigger2 := test.MinEvent()
+	eventTrigger2.SetID(eventIdTrigger2)
+	eventTrigger2.SetType(trigger2FilterType)
+
+	f := feature.NewFeatureNamed("Trigger with KafkaSink")
+
+	f.Setup("Create broker config", brokerconfigmap.Install(
+		brokerConfig,
+		brokerconfigmap.WithBootstrapServer(testpkg.BootstrapServersPlaintext),
+		brokerconfigmap.WithNumPartitions(testpkg.NumPartitions),
+		brokerconfigmap.WithReplicationFactor(testpkg.ReplicationFactor),
+	))
+
+	f.Setup("Install broker", broker.Install(brokerName,
+		append(
+			broker.WithEnvConfig(),
+			broker.WithConfig(brokerConfig))...,
+	))
+	f.Setup("Broker is ready", broker.IsReady(brokerName))
+	f.Setup("Topic is ready", kafkatopic.IsReady(topic))
+
+	f.Setup("Install kafkasink", kafkasink.Install(sink, topic, testpkg.BootstrapServersPlaintextArr,
+		kafkasink.WithNumPartitions(10),
+		kafkasink.WithReplicationFactor(1)))
+	f.Setup("KafkaSink is ready", kafkasink.IsReady(sink))
+
+	f.Setup("Create trigger 1", trigger.Install(trigger1Name, brokerName,
+		trigger.WithSubscriber(kafkasink.AsKReference(sink, env.Namespace()), ""),
+		trigger.WithFilter(map[string]string{
+			"type": trigger1FilterType,
+		})))
+	f.Setup("Trigger 1 is ready", trigger.IsReady(trigger1Name))
+
+	f.Setup("Create trigger 2", trigger.Install(trigger2Name, brokerName,
+		trigger.WithSubscriber(kafkasink.AsKReference(sink, env.Namespace()), ""),
+		trigger.WithFilter(map[string]string{
+			"type": trigger2FilterType,
+		})))
+	f.Setup("Trigger 2 is ready", trigger.IsReady(trigger2Name))
+
+	f.Requirement("Send event for trigger 1", eventshub.Install(feature.MakeRandomK8sName("sender-1"),
+		eventshub.InputEvent(eventTrigger1),
+		eventshub.StartSenderToResource(broker.GVR(), brokerName)))
+	f.Requirement("Send event for trigger 2", eventshub.Install(feature.MakeRandomK8sName("sender-2"),
+		eventshub.InputEvent(eventTrigger2),
+		eventshub.StartSenderToResource(broker.GVR(), brokerName)))
+
+	f.Requirement("Install verify-messages job", job.Install(verifyMessagesJobName, kafkaConsumerImage,
+		job.WithEnvs(map[string]string{
+			"BOOTSTRAP_SERVERS": testpkg.BootstrapServersPlaintext,
+			"TOPIC":             topic,
+			"IDS":               fmt.Sprintf("%s,%s", eventIdTrigger1, eventIdTrigger2),
+			"CONTENT_MODE":      "structured",
+		}),
+		job.WithRestartPolicy(corev1.RestartPolicyNever),
+		job.WithBackoffLimit(2),
+		job.WithImagePullPolicy(corev1.PullIfNotPresent)))
+	f.Assert("Verify-messages job succeeded", job.IsSucceeded(verifyMessagesJobName))
+
+	return f
+}
diff --git a/test/rekt/resources/kafkasink/kafkasink.go b/test/rekt/resources/kafkasink/kafkasink.go
index 0fa409894..c75439683 100644
--- a/test/rekt/resources/kafkasink/kafkasink.go
+++ b/test/rekt/resources/kafkasink/kafkasink.go
@@ -19,9 +19,11 @@ package kafkasink
 import (
 	"context"
 	"embed"
+	"fmt"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
@@ -63,7 +65,7 @@ func WithAuthSecretName(name string) manifest.CfgFn {
 }
 
 // Install will create a Trigger resource, augmented with the config fn options.
-func Install(name, topic, bootstrapServers []string, opts ...manifest.CfgFn) feature.StepFn {
+func Install(name, topic string, bootstrapServers []string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name":             name,
 		"topic":            topic,
@@ -83,3 +85,12 @@ func Install(name, topic, bootstrapServers []string, opts ...manifest.CfgFn) fea
 func IsReady(name string, timing ...time.Duration) feature.StepFn {
 	return k8s.IsReady(gvr(), name, timing...)
 }
+
+func AsKReference(name string, namespace string) *duckv1.KReference {
+	return &duckv1.KReference{
+		Kind:       "KafkaSink",
+		APIVersion: fmt.Sprintf("%s/%s", gvr().Group, gvr().Version),
+		Namespace:  namespace,
+		Name:       name,
+	}
+}
diff --git a/test/rekt/resources/kafkatopic/topic.go b/test/rekt/resources/kafkatopic/topic.go
index 5fc12e092..544e547a9 100644
--- a/test/rekt/resources/kafkatopic/topic.go
+++ b/test/rekt/resources/kafkatopic/topic.go
@@ -19,10 +19,16 @@ package kafkatopic
 import (
 	"context"
 	"embed"
+	"fmt"
 	"time"
 
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/injection/clients/dynamicclient"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
@@ -69,6 +75,83 @@ func IsReady(name string, timings ...time.Duration) feature.StepFn {
 	}
 }
 
+// HasReplicationFactor asserts that the Topic has the given replication factor.
+func HasReplicationFactor(name string, replicationFactor int, timings ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		interval, timeout := k8s.PollTimings(ctx, timings)
+
+		err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+			ut, err := dynamicclient.Get(ctx).
+				Resource(GVR()).
+				Namespace(kafkaNamespace).
+				Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				if apierrors.IsNotFound(err) {
+					t.Logf("failed to get topic %s/%s: %v", kafkaNamespace, name, err)
+					// keep polling
+					return false, nil
+				}
+				return false, err
+			}
+
+			topicObj := &topic{}
+			if err := runtime.DefaultUnstructuredConverter.FromUnstructured(ut.UnstructuredContent(), topicObj); err != nil {
+				return false, fmt.Errorf("could not convert unstructured to topic: %w", err)
+			}
+
+			if topicObj.Spec.Replicas != replicationFactor {
+				t.Logf("Kafkatopic %s/%s does not have the expected replication factor of %d. Has %d instead.", name, kafkaNamespace, topicObj.Spec.Replicas, replicationFactor)
+				// keep polling
+				return false, nil
+			} else {
+				return true, nil
+			}
+		})
+
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// HasReplicationFactor asserts that the Topic has the given replication factor.
+func HasNumPartitions(name string, numPartitions int, timings ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		interval, timeout := k8s.PollTimings(ctx, timings)
+
+		err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+			ut, err := dynamicclient.Get(ctx).
+				Resource(GVR()).
+				Namespace(kafkaNamespace).
+				Get(ctx, name, metav1.GetOptions{})
+			if err != nil {
+				if apierrors.IsNotFound(err) {
+					t.Logf("failed to get topic %s/%s: %v", kafkaNamespace, name, err)
+					// keep polling
+					return false, nil
+				}
+				return false, err
+			}
+
+			topicObj := &topic{}
+			if err := runtime.DefaultUnstructuredConverter.FromUnstructured(ut.UnstructuredContent(), topicObj); err != nil {
+				return false, fmt.Errorf("could not convert unstructured to topic: %w", err)
+			}
+
+			if topicObj.Spec.Partitions != numPartitions {
+				t.Logf("Kafkatopic %s/%s does not have the expected number of partitions of %d. Has %d instead.", name, kafkaNamespace, topicObj.Spec.Partitions, numPartitions)
+				// keep polling
+				return false, nil
+			}
+			return true, nil
+		})
+
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
 // WithPartitions overrides the number of partitions (default: 10).
 func WithPartitions(partitions string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
diff --git a/test/rekt/resources/kafkatopic/types.go b/test/rekt/resources/kafkatopic/types.go
new file mode 100644
index 000000000..0db68fe64
--- /dev/null
+++ b/test/rekt/resources/kafkatopic/types.go
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 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 kafkatopic
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type topic struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec topicSpec `json:"spec,omitempty"`
+}
+
+type topicSpec struct {
+	Partitions int    `json:"partitions,omitempty"`
+	Replicas   int    `json:"replicas,omitempty"`
+	TopicName  string `json:"topicName,omitempty"`
+}
diff --git a/test/test_images/partitions-replication-verifier/partitions_replication_verifier.go b/test/test_images/partitions-replication-verifier/partitions_replication_verifier.go
deleted file mode 100644
index 4b0b35eec..000000000
--- a/test/test_images/partitions-replication-verifier/partitions_replication_verifier.go
+++ /dev/null
@@ -1,58 +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 main
-
-import (
-	"log"
-
-	"github.com/Shopify/sarama"
-	"github.com/kelseyhightower/envconfig"
-
-	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
-)
-
-func main() {
-	envConfig := &kafkatest.Config{}
-
-	if err := envconfig.Process("", envConfig); err != nil {
-		log.Fatalf("failed to process env variables: %v", err)
-	}
-
-	config := sarama.NewConfig()
-	config.Version = sarama.MaxVersion
-
-	admin, err := sarama.NewClusterAdmin([]string{envConfig.BootstrapServers}, config)
-	if err != nil {
-		log.Fatalf("failed to create cluster admin: %v", err)
-	}
-
-	topics, err := admin.ListTopics()
-	if err != nil {
-		log.Fatalf("failed to list topics: %v", err)
-	}
-
-	td, ok := topics[envConfig.Topic]
-	if !ok {
-		log.Fatalf("topic %s hasn't been created", envConfig.Topic)
-	}
-	if td.NumPartitions != envConfig.NumPartitions {
-		log.Fatalf("expected num partitions %d got %d", envConfig.NumPartitions, td.NumPartitions)
-	}
-	if td.ReplicationFactor != envConfig.ReplicationFactor {
-		log.Fatalf("expected replication factor %d got %d", envConfig.ReplicationFactor, td.ReplicationFactor)
-	}
-}
diff --git a/test/test_images/partitions-replication-verifier/pod.yaml b/test/test_images/partitions-replication-verifier/pod.yaml
deleted file mode 100644
index 59089d0c3..000000000
--- a/test/test_images/partitions-replication-verifier/pod.yaml
+++ /dev/null
@@ -1,8 +0,0 @@
-apiVersion: v1
-kind: Pod
-metadata:
-  name: partitions_replication_verifier
-spec:
-  containers:
-    - name: partitions_replication_verifier
-      image: ko://knative.dev/eventing-kafka-broker/test/test_images/partitions-replication-verifier
diff --git a/test/upgrade/continual/broker.go b/test/upgrade/continual/broker.go
index 682a646a2..155e91a53 100644
--- a/test/upgrade/continual/broker.go
+++ b/test/upgrade/continual/broker.go
@@ -155,10 +155,7 @@ func (k *kafkaBrokerSut) fetchURL(ctx sut.Context) *apis.URL {
 		ctx.T.Fatal(err)
 	}
 
-	var url *apis.URL
-	if br.Status.AddressStatus.Address != nil {
-		url = br.Status.Address.URL
-	}
+	url := br.Status.Address.URL
 	ctx.Log.Debugf("\"%s\" broker URL for ns %s is %v", k.Broker.Name, namespace, url)
 	return url
 }
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index c2870869b..db245ab3d 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20230129-288c8404c"
+    kafka.eventing.knative.dev/release: "v20230203-109b0159b"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index a9adde0a3..ab33714cf 100644
--- a/third_party/eventing-kafka-latest/source-crd.yaml
+++ b/third_party/eventing-kafka-latest/source-crd.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20230129-288c8404c"
+    kafka.eventing.knative.dev/release: "v20230203-109b0159b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go
index f78ab6a83..04bb90f63 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go
@@ -111,7 +111,7 @@ func WithApiServerSourceSufficientPermissions(s *v1.ApiServerSource) {
 }
 
 func WithApiServerSourceNoSufficientPermissions(s *v1.ApiServerSource) {
-	s.Status.MarkNoSufficientPermissions("", `User system:serviceaccount:testnamespace:default cannot get, list, watch resource "namespaces" in API group ""`)
+	s.Status.MarkNoSufficientPermissions("", `User system:serviceaccount:testnamespace:default cannot get, list, watch resource "namespaces" in API group "" in Namespace "testnamespace"`)
 }
 
 func WithApiServerSourceDeleted(c *v1.ApiServerSource) {
@@ -136,3 +136,15 @@ func WithApiServerSourceObjectMetaGeneration(generation int64) ApiServerSourceOp
 		c.ObjectMeta.Generation = generation
 	}
 }
+
+func WithApiServerSourceNamespaceSelector(nsSelector metav1.LabelSelector) ApiServerSourceOption {
+	return func(c *v1.ApiServerSource) {
+		c.Spec.NamespaceSelector = &nsSelector
+	}
+}
+
+func WithApiServerSourceStatusNamespaces(namespaces []string) ApiServerSourceOption {
+	return func(c *v1.ApiServerSource) {
+		c.Status.Namespaces = namespaces
+	}
+}
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
index b42e75967..aa3690ec0 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
@@ -250,6 +250,12 @@ func MarkNotAddedToChannel(reason, msg string) SubscriptionOption {
 	}
 }
 
+func MarkChannelFailed(reason, msg string) SubscriptionOption {
+	return func(s *v1.Subscription) {
+		s.Status.MarkChannelFailed(reason, msg)
+	}
+}
+
 func MarkReferencesResolved(s *v1.Subscription) {
 	s.Status.MarkReferencesResolved()
 }
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/broker_channel_flow_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/broker_channel_flow_helper.go
index 82427785a..088d51bd0 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/broker_channel_flow_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/broker_channel_flow_helper.go
@@ -51,6 +51,7 @@ Trigger1 filters the orignal event and transforms it to a new event,
 Trigger2 logs all events,
 Trigger3 filters the transformed event and sends it to Channel.
 */
+// Deprecated, use reconciler-test based tests.
 func BrokerChannelFlowWithTransformation(
 	ctx context.Context,
 	t *testing.T,
diff --git a/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go b/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
index ed8903ff0..f907fb205 100644
--- a/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
+++ b/vendor/knative.dev/eventing/test/experimental/features/delivery_timeout/channel.go
@@ -24,16 +24,17 @@ import (
 	"github.com/rickb777/date/period"
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
-	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	eventingclient "knative.dev/eventing/pkg/client/injection/client"
-	"knative.dev/eventing/test/rekt/resources/channel"
-	"knative.dev/eventing/test/rekt/resources/subscription"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
+
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/channel"
+	"knative.dev/eventing/test/rekt/resources/subscription"
 )
 
 // ChannelToSink tests a scenario where the flow is source -> channel -> sink (timeout) -- fallback to -> dead letter sink
@@ -106,7 +107,7 @@ func ChannelToSink() *feature.Feature {
 
 	f.Setup("subscription channel -> Sink is ready", subscription.IsReady(subName))
 
-	f.Setup("install source", eventshub.Install(
+	f.Requirement("install source", eventshub.Install(
 		sourceName,
 		eventshub.StartSenderToResource(channel.GVR(), channelName),
 		eventshub.InputEvent(ev),
diff --git a/vendor/knative.dev/eventing/test/lib/logexporter.go b/vendor/knative.dev/eventing/test/lib/logexporter.go
index 5bd42543e..efe0107e8 100644
--- a/vendor/knative.dev/eventing/test/lib/logexporter.go
+++ b/vendor/knative.dev/eventing/test/lib/logexporter.go
@@ -17,16 +17,21 @@ limitations under the License.
 package lib
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"log"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	pkgtest "knative.dev/pkg/test"
 	"knative.dev/pkg/test/helpers"
+	"knative.dev/pkg/test/logstream/v2"
 	"knative.dev/pkg/test/prow"
 )
 
@@ -95,3 +100,92 @@ func ExportLogs(systemLogsDir, systemNamespace string) {
 		}
 	}
 }
+
+// ExportLogStreamOnError starts a log stream from the given namespace
+// for pods specified via podPrefixes. The log strema is stopped and
+// logs exported upon calling the returned Canceler.
+func ExportLogStreamOnError(t *testing.T, logDir, namespace string, podPrefixes ...string) logstream.Canceler {
+	config, err := pkgtest.Flags.GetRESTConfig()
+	if err != nil {
+		t.Fatalf("Failed to create REST config: %v\n", err)
+	}
+
+	kubeClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		t.Fatalf("Failed to create kube client: %v\n", err)
+	}
+
+	buf := threadSafeBuffer{}
+	callback := func(s string, params ...interface{}) {
+		buf.Write([]byte(fmt.Sprintf(s, params...) + "\n"))
+	}
+
+	sysStream := logstream.New(context.Background(), kubeClient,
+		logstream.WithNamespaces(namespace),
+		logstream.WithLineFiltering(false),
+		logstream.WithPodPrefixes(podPrefixes...))
+	canceler, err := sysStream.StartStream("unfiltered-logs", callback)
+	if err != nil {
+		t.Fatal("Unable to stream logs from namespace", err)
+	}
+
+	return func() {
+		canceler()
+
+		if !t.Failed() {
+			return
+		}
+
+		// Maps pod name prefix to its logs.
+		logs := make(map[string]string)
+		if len(podPrefixes) == 0 {
+			logs["all-in-one"] = buf.String()
+		} else {
+			// Divide logs by podPrefix so that logs for each pod are stored in a separate file.
+			for _, podPrefix := range podPrefixes {
+				for _, line := range strings.Split(buf.String(), "\n") {
+					if strings.Contains(line, podPrefix) {
+						logs[podPrefix] = logs[podPrefix] + line + "\n"
+					}
+				}
+			}
+		}
+
+		dir := filepath.Join(prow.GetLocalArtifactsDir(), logDir)
+		if err := helpers.CreateDir(dir); err != nil {
+			t.Errorf("Error creating directory %q: %v", dir, err)
+		}
+
+		var errs []error
+		for podPrefix, log := range logs {
+			path := filepath.Join(dir, fmt.Sprintf("%s.log", podPrefix))
+			f, err := os.Create(path)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("error creating file %q: %w", path, err))
+				continue
+			}
+			_, err = f.Write([]byte(log))
+			if err != nil {
+				errs = append(errs, fmt.Errorf("error writing logs into file %q: %w", path, err))
+			}
+			_ = f.Close()
+		}
+
+		if len(errs) > 0 {
+			t.Errorf("Failed to write logs: %v", helpers.CombineErrors(errs))
+		}
+	}
+}
+
+// threadSafeBuffer avoids race conditions on bytes.Buffer.
+// See: https://stackoverflow.com/a/36226525/844449
+type threadSafeBuffer struct {
+	bytes.Buffer
+	sync.Mutex
+}
+
+func (b *threadSafeBuffer) Write(p []byte) (n int, err error) {
+	b.Mutex.Lock()
+	defer b.Mutex.Unlock()
+	return b.Buffer.Write(p)
+}
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
index 07d01c8e4..95b20831e 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/control_plane.go
@@ -370,8 +370,7 @@ func addControlPlaneDelivery(fs *feature.FeatureSet, brokerOpts ...manifest.CfgF
 
 		// Send an event into the matrix and hope for the best
 		prober.SenderFullEvents(1)
-		f.Setup("install source", prober.SenderInstall("source"))
-		f.Requirement("sender is finished", prober.SenderDone("source"))
+		f.Requirement("install source", prober.SenderInstall("source"))
 
 		// All events have been sent, time to look at the specs and confirm we got them.
 		expectedEvents := createExpectedEventPatterns(tt.brokerDS, cfg)
@@ -516,8 +515,8 @@ func addControlPlaneEventRouting(fs *feature.FeatureSet, brokerOpts ...manifest.
 		// All the events generated are currently hardcoded into the com.example.FullEvent
 		// so once prober supports more configuration, wire it up here.
 		prober.SenderFullEvents(1)
-		f.Setup("install source", prober.SenderInstall("source"))
-		f.Requirement("sender is finished", prober.SenderDone("source"))
+
+		f.Requirement("install source", prober.SenderInstall("source"))
 
 		// All events have been sent, time to look at the specs and confirm we got them.
 		expectedEvents := createExpectedEventRoutingMap(tt.config, tt.inEvents)
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go b/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go
index 2c2308442..318a47997 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/feature.go
@@ -22,12 +22,17 @@ import (
 	"strings"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
+
 	"github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
 
-	eventingv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/channel"
+	"knative.dev/eventing/test/rekt/resources/subscription"
 	"knative.dev/eventing/test/rekt/resources/trigger"
+	eventingduckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/ptr"
 
 	"knative.dev/reconciler-test/pkg/eventshub"
@@ -37,6 +42,155 @@ import (
 	"knative.dev/reconciler-test/resources/svc"
 )
 
+/*
+BrokerChannelFlowWithTransformation tests the following topology:
+	------------- ----------------------
+	|           | |                    |
+	v	       | v                    |
+EventSource ---> Broker ---> Trigger1 -------> Sink1(Transformation)
+	|
+	|
+	|-------> Trigger2 -------> Sink2(Logger1)
+	|
+	|
+	|-------> Trigger3 -------> Channel --------> Subscription --------> Sink3(Logger2)
+Explanation:
+Trigger1 filters the orignal event and transforms it to a new event,
+Trigger2 logs all events,
+Trigger3 filters the transformed event and sends it to Channel.
+*/
+
+func BrokerChannelFlowWithTransformation(createSubscriberFn func(ref *eventingduckv1.KReference, uri string) manifest.CfgFn) *feature.Feature {
+	f := feature.NewFeatureNamed("Broker topology of transformation")
+
+	source := feature.MakeRandomK8sName("source")
+
+	sink1 := feature.MakeRandomK8sName("sink1")
+	sink2 := feature.MakeRandomK8sName("sink2")
+	sink3 := feature.MakeRandomK8sName("sink3")
+
+	trigger1 := feature.MakeRandomK8sName("trigger1")
+	trigger2 := feature.MakeRandomK8sName("trigger2")
+	trigger3 := feature.MakeRandomK8sName("trigger3")
+
+	// Construct original cloudevent message
+	eventType := "type1"
+	eventSource := "http://source1.com"
+	eventBody := `{"msg":"e2e-brokerchannel-body"}`
+	// Construct cloudevent message after transformation
+	transformedEventType := "type2"
+	transformedEventSource := "http://source2.com"
+	transformedBody := `{"msg":"transformed body"}`
+	// Construct eventToSend
+	eventToSend := cloudevents.NewEvent()
+	eventToSend.SetID(uuid.New().String())
+	eventToSend.SetType(eventType)
+	eventToSend.SetSource(eventSource)
+	eventToSend.SetData(cloudevents.ApplicationJSON, []byte(eventBody))
+
+	//Install the broker
+	brokerName := feature.MakeRandomK8sName("broker")
+	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
+
+	f.Setup("install sink1", eventshub.Install(sink1,
+		eventshub.ReplyWithTransformedEvent(transformedEventType, transformedEventSource, transformedBody),
+		eventshub.StartReceiver),
+	)
+	f.Setup("install sink2", eventshub.Install(sink2, eventshub.StartReceiver))
+	f.Setup("install sink3", eventshub.Install(sink3, eventshub.StartReceiver))
+
+	// filter1 filters the original events
+	filter1 := eventingv1.TriggerFilterAttributes{
+		"type":   eventType,
+		"source": eventSource,
+	}
+	// filter2 filters all events
+	filter2 := eventingv1.TriggerFilterAttributes{
+		"type": eventingv1.TriggerAnyFilter,
+	}
+	// filter3 filters events after transformation
+	filter3 := eventingv1.TriggerFilterAttributes{
+		"type":   transformedEventType,
+		"source": transformedEventSource,
+	}
+
+	// Install the trigger1 point to Broker and transform the original events to new events
+	f.Setup("install trigger1", trigger.Install(
+		trigger1,
+		brokerName,
+		trigger.WithFilter(filter1),
+		trigger.WithSubscriber(svc.AsKReference(sink1), ""),
+	))
+	f.Setup("trigger1 goes ready", trigger.IsReady(trigger1))
+	// Install the trigger2 point to Broker to filter all the events
+	f.Setup("install trigger2", trigger.Install(
+		trigger2,
+		brokerName,
+		trigger.WithFilter(filter2),
+		trigger.WithSubscriber(svc.AsKReference(sink2), ""),
+	))
+	f.Setup("trigger2 goes ready", trigger.IsReady(trigger2))
+
+	// Install the channel and corresponding subscription point to sink3
+	channelName := feature.MakeRandomK8sName("channel")
+	f.Setup("install channel", channel.Install(channelName,
+		channel.WithTemplate(),
+	))
+	sub := feature.MakeRandomK8sName("subscription")
+	f.Setup("install subscription", subscription.Install(sub,
+		subscription.WithChannel(channel.AsRef(channelName)),
+		createSubscriberFn(svc.AsKReference(sink3), ""),
+	))
+	f.Setup("subscription is ready", subscription.IsReady(sub))
+	f.Setup("channel is ready", channel.IsReady(channelName))
+
+	// Install the trigger3 point to Broker to filter the events after transformation point to channel
+	f.Setup("install trigger3", trigger.Install(
+		trigger3,
+		brokerName,
+		trigger.WithFilter(filter3),
+		trigger.WithSubscriber(channel.AsRef(channelName), ""),
+	))
+	f.Setup("trigger3 goes ready", trigger.IsReady(trigger3))
+
+	f.Requirement("install source", eventshub.Install(
+		source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(eventToSend),
+	))
+
+	eventMatcher := eventasssert.MatchEvent(
+		test.HasSource(eventSource),
+		test.HasType(eventType),
+		test.HasData([]byte(eventBody)),
+	)
+	transformEventMatcher := eventasssert.MatchEvent(
+		test.HasSource(transformedEventSource),
+		test.HasType(transformedEventType),
+		test.HasData([]byte(transformedBody)),
+	)
+
+	f.Stable("(Trigger1 point to) sink1 has all the events").
+		Must("delivers original events",
+			eventasssert.OnStore(sink1).Match(eventMatcher).AtLeast(1))
+
+	f.Stable("(Trigger2 point to) sink2 has all the events").
+		Must("delivers original events",
+			eventasssert.OnStore(sink2).Match(eventMatcher).AtLeast(1)).
+		Must("delivers transformation events",
+			eventasssert.OnStore(sink2).Match(transformEventMatcher).AtLeast(1))
+
+	f.Stable("(Trigger3 point to) Channel's subscriber just has events after transformation").
+		Must("delivers transformation events",
+			eventasssert.OnStore(sink3).Match(transformEventMatcher).AtLeast(1)).
+		Must("delivers original events",
+			eventasssert.OnStore(sink3).Match(eventMatcher).Not())
+
+	return f
+}
+
 func BrokerPreferHeaderCheck() *feature.Feature {
 	f := feature.NewFeatureNamed("Broker PreferHeader Check")
 
@@ -114,7 +268,7 @@ func brokerRedeliveryFibonacci(retryNum int32) *feature.Feature {
 	//Install the broker
 	brokerName := feature.MakeRandomK8sName("broker")
 
-	exp := eventingv1.BackoffPolicyLinear
+	exp := duckv1.BackoffPolicyLinear
 	brokerConfig := append(broker.WithEnvConfig(), broker.WithRetry(retryNum, &exp, ptr.String("PT1S")))
 	f.Setup("install broker", broker.Install(brokerName, brokerConfig...))
 	f.Requirement("broker is ready", broker.IsReady(brokerName))
@@ -168,7 +322,7 @@ func brokerRedeliveryDropN(retryNum int32, dropNum uint) *feature.Feature {
 
 	//Install the broker
 	brokerName := feature.MakeRandomK8sName("broker")
-	exp := eventingv1.BackoffPolicyLinear
+	exp := duckv1.BackoffPolicyLinear
 	brokerConfig := append(broker.WithEnvConfig(), broker.WithRetry(retryNum, &exp, ptr.String("PT1S")))
 
 	f.Setup("install broker", broker.Install(brokerName, brokerConfig...))
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
index eb98965e0..25634098a 100644
--- 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
@@ -19,19 +19,20 @@ package broker
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/google/uuid"
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/delivery"
-	"knative.dev/eventing/test/rekt/resources/eventlibrary"
-	"knative.dev/eventing/test/rekt/resources/flaker"
-	"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"
 	"knative.dev/reconciler-test/resources/svc"
 
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/eventing/test/rekt/resources/flaker"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+
 	. "github.com/cloudevents/sdk-go/v2/test"
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
@@ -56,13 +57,10 @@ func SourceToSink(brokerName string) *feature.Feature {
 
 	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.Requirement("install source", eventshub.Install(source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(event),
+	))
 
 	f.Stable("broker as middleware").
 		Must("deliver an event",
@@ -81,43 +79,31 @@ func SourceToSink(brokerName string) *feature.Feature {
 func SourceToSinkWithDLQ() *feature.Feature {
 	f := feature.NewFeature()
 
-	prober := eventshub.NewProber()
 	brokerName := feature.MakeRandomK8sName("broker")
-	dlq := feature.MakeRandomK8sName("dlq")
-	via := feature.MakeRandomK8sName("via")
+	dls := feature.MakeRandomK8sName("dls")
+	triggerName := feature.MakeRandomK8sName("trigger")
 	source := feature.MakeRandomK8sName("source")
 
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
-
-	f.Setup("install dlq", prober.ReceiverInstall(dlq))
+	f.Setup("install dead letter sink service", eventshub.Install(dls, eventshub.StartReceiver))
 
-	brokerConfig := append(broker.WithEnvConfig(), delivery.WithDeadLetterSink(prober.AsKReference(dlq), ""))
+	brokerConfig := append(broker.WithEnvConfig(), delivery.WithDeadLetterSink(svc.AsKReference(dls), ""))
 	f.Setup("install broker", broker.Install(brokerName, brokerConfig...))
-	// Block till broker is ready
 	f.Setup("Broker is ready", broker.IsReady(brokerName))
-	prober.SetTargetResource(broker.GVR(), brokerName)
+	f.Setup("install trigger", trigger.Install(triggerName, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
+	f.Setup("trigger is ready", trigger.IsReady(triggerName))
 
-	f.Setup("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
-	// Trigger is ready.
-	f.Setup("trigger goes ready", trigger.IsReady(via))
-
-	// Install events after data plane is ready.
-	f.Setup("install source", prober.SenderInstall(source))
+	ce := FullEvent()
+	ce.SetID(uuid.New().String())
 
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone(source))
-	f.Requirement("receiver is finished", prober.ReceiverDone(source, dlq))
+	// Send events after data plane is ready.
+	f.Requirement("install source", eventshub.Install(source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(ce),
+	))
 
 	// Assert events ended up where we expected.
-	f.Stable("broker with DLQ").
-		Must("accepted all events", prober.AssertSentAll(source)).
-		Must("deliver event to DLQ", prober.AssertReceivedAll(source, dlq))
+	f.Stable("broker with DLS").
+		Must("deliver event to DLQ", OnStore(dls).MatchEvent(HasId(ce.ID())).AtLeast(1))
 
 	return f
 }
@@ -135,15 +121,7 @@ func SourceToSinkWithFlakyDLQ(brokerName string) *feature.Feature {
 	dlq := feature.MakeRandomK8sName("dlq")
 	via := feature.MakeRandomK8sName("via")
 
-	uuids := []string{
-		uuid.New().String(),
-		uuid.New().String(),
-		uuid.New().String(),
-	}
-
-	f := feature.NewFeature()
-
-	//
+	f := feature.NewFeatureNamed("Source to sink with flaky DLS")
 
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 
@@ -154,39 +132,25 @@ func SourceToSinkWithFlakyDLQ(brokerName string) *feature.Feature {
 	})
 
 	f.Setup("install dlq", eventshub.Install(dlq, eventshub.StartReceiver))
-
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver, eventshub.DropFirstN(2)))
 	f.Setup("update broker with DLQ", broker.Install(brokerName, broker.WithDeadLetterSink(svc.AsKReference(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("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(svc.AsKReference(sink), "")))
 	f.Setup("trigger goes ready", trigger.IsReady(via))
+	f.Setup("broker goes ready", broker.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.Requirement("install source", eventshub.Install(source,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.SendMultipleEvents(3, time.Millisecond),
+		eventshub.EnableIncrementalId,
+	))
 
-	f.Stable("broker with DQL").
+	f.Stable("broker with DLS").
 		Must("deliver event flaky sent to DLQ event[0]",
-			OnStore(dlq).MatchEvent(HasId(uuids[0])).Exact(1)).
+			OnStore(dlq).MatchEvent(HasId("1")).Exact(1)).
 		Must("deliver event flaky sent to DLQ event[1]",
-			OnStore(dlq).MatchEvent(HasId(uuids[1])).Exact(1)).
+			OnStore(dlq).MatchEvent(HasId("2")).Exact(1)).
 		Must("deliver event sink receiver got event[2]",
-			OnStore(sink).MatchEvent(HasId(uuids[2])).Exact(1))
+			OnStore(sink).MatchEvent(HasId("3")).Exact(1))
 
 	return f
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go b/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go
index e60edbd25..4abcab91f 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/control_plane.go
@@ -18,6 +18,7 @@ package channel
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
@@ -89,9 +90,9 @@ func ControlPlaneChannel(channelName string) *feature.Feature {
 			"Each channel CRD MUST have a status subresource which contains [subscribers (as an array)]. "+
 			"The ready field of the subscriber identified by its uid MUST be set to True when the subscription is ready to be processed.",
 			channelAllowsSubscribersAndStatus)
-		// Special note for Channel tests: The array of subscribers MUST NOT be
-		// set directly on the generic Channel custom object, but rather
-		// appended to the backing channel by the subscription itself.
+	// Special note for Channel tests: The array of subscribers MUST NOT be
+	// set directly on the generic Channel custom object, but rather
+	// appended to the backing channel by the subscription itself.
 
 	f.Stable("Status Requirements").
 		Must("observedGeneration MUST be populated if present",
@@ -105,11 +106,12 @@ func ControlPlaneChannel(channelName string) *feature.Feature {
 	cName := feature.MakeRandomK8sName("channel")
 	sink := feature.MakeRandomK8sName("sink")
 
-	f.Setup("Set Channel Name", setChannelableName(cName))
-
 	f.Setup("install a service", svc.Install(sink, "app", "rekt"))
 	f.Setup("update Channel", channel_impl.Install(cName, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
 	f.Setup("Channel goes ready", channel_impl.IsReady(cName))
+	f.Setup("Channel is addressable", channel_impl.IsAddressable(cName))
+
+	f.Requirement("Channel has dead letter sink URI in status", channel_impl.HasDeadLetterSinkURI(cName, channel_impl.GVR()))
 
 	f.Stable("Channel Status").
 		Must("When the channel instance is ready to receive events status.address.url MUST be populated. "+
@@ -117,7 +119,7 @@ func ControlPlaneChannel(channelName string) *feature.Feature {
 			"When the Channel instance is ready to receive events status.address.url status.addressable MUST be set to True",
 			readyChannelIsAddressable).
 		Should("Set the Channel status.deadLetterSinkURI if there is a valid spec.delivery.deadLetterSink defined",
-			readyChannelWithDLSHaveStatusUpdated)
+			readyChannelWithDLSHaveStatusUpdated(cName))
 
 	return f
 }
@@ -250,16 +252,19 @@ func readyChannelIsAddressable(ctx context.Context, t feature.T) {
 	}
 }
 
-func readyChannelWithDLSHaveStatusUpdated(ctx context.Context, t feature.T) {
-	ch := getChannelable(ctx, t)
+func readyChannelWithDLSHaveStatusUpdated(name string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		ch := getChannelableFromName(name, ctx, t)
 
-	// Confirm the channel is ready, and has the status.deadLetterSinkURI set.
-	if c := ch.Status.GetCondition(apis.ConditionReady); c.IsTrue() {
-		if ch.Status.DeadLetterSinkURI == nil {
-			t.Errorf("channel DLS not resolved but resource reported ready")
+		// Confirm the channel is ready, and has the status.deadLetterSinkURI set.
+		if c := ch.Status.GetCondition(apis.ConditionReady); c.IsTrue() {
+			if ch.Status.DeadLetterSinkURI == nil {
+				bytes, _ := json.MarshalIndent(ch, "", " ")
+				t.Errorf("channel DLS not resolved but resource reported ready, state:\n%s", string(bytes))
+			}
+			// Success!
+		} else {
+			t.Errorf("channel was not ready, reason: %s", ch.Status.GetCondition(apis.ConditionReady).Reason)
 		}
-		// Success!
-	} else {
-		t.Errorf("channel was not ready, reason: %s", ch.Status.GetCondition(apis.ConditionReady).Reason)
 	}
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go b/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go
index 76a4967fd..c508804ae 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/data_plane.go
@@ -235,8 +235,7 @@ func addControlPlaneDelivery(fs *feature.FeatureSet) {
 
 		// Send an event into the matrix and hope for the best
 		prober.SenderFullEvents(1)
-		f.Setup("install source", prober.SenderInstall("source"))
-		f.Requirement("sender is finished", prober.SenderDone("source"))
+		f.Requirement("install source", prober.SenderInstall("source"))
 
 		// All events have been sent, time to look at the specs and confirm we got them.
 		expected := createExpectedEventPatterns(tt.chDS, tt.subs)
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/features.go b/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
index f393e2b2b..75d2da134 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/features.go
@@ -32,15 +32,15 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/resources/svc"
 
+	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
+
 	"knative.dev/eventing/test/rekt/resources/channel"
 	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/containersource"
 	"knative.dev/eventing/test/rekt/resources/delivery"
-	"knative.dev/eventing/test/rekt/resources/eventlibrary"
 	"knative.dev/eventing/test/rekt/resources/pingsource"
 	"knative.dev/eventing/test/rekt/resources/source"
 	"knative.dev/eventing/test/rekt/resources/subscription"
-	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
 
 func ChannelChain(length int, createSubscriberFn func(ref *duckv1.KReference, uri string) manifest.CfgFn) *feature.Feature {
@@ -89,19 +89,20 @@ func DeadLetterSink(createSubscriberFn func(ref *duckv1.KReference, uri string)
 	failer := feature.MakeK8sNamePrefix("failer")
 	cs := feature.MakeRandomK8sName("containersource")
 	name := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
 
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 	f.Setup("install failing receiver", eventshub.Install(failer, eventshub.StartReceiver, eventshub.DropFirstN(1)))
 	f.Setup("install channel", channel_impl.Install(name, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
-	f.Setup("install containersource", containersource.Install(cs, source.WithSink(channel_impl.AsRef(name), "")))
-	f.Setup("install subscription", subscription.Install(feature.MakeRandomK8sName("subscription"),
+	f.Setup("install subscription", subscription.Install(sub,
 		subscription.WithChannel(channel_impl.AsRef(name)),
 		createSubscriberFn(svc.AsKReference(failer), ""),
 	))
-
 	f.Setup("channel is ready", channel_impl.IsReady(name))
-	f.Setup("containersource is ready", containersource.IsReady(cs))
+	f.Setup("subscription is ready", subscription.IsReady(sub))
 
+	f.Requirement("install containersource", containersource.Install(cs, source.WithSink(channel_impl.AsRef(name), "")))
+	f.Requirement("containersource is ready", containersource.IsReady(cs))
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(name, channel_impl.GVR()))
 
 	f.Assert("dls receives events", assert.OnStore(sink).
@@ -118,6 +119,7 @@ func DeadLetterSinkGenericChannel(createSubscriberFn func(ref *duckv1.KReference
 	failer := feature.MakeK8sNamePrefix("failer")
 	cs := feature.MakeRandomK8sName("containersource")
 	name := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
 
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 	f.Setup("install failing receiver", eventshub.Install(failer, eventshub.StartReceiver, eventshub.DropFirstN(1)))
@@ -125,15 +127,15 @@ func DeadLetterSinkGenericChannel(createSubscriberFn func(ref *duckv1.KReference
 		channel.WithTemplate(),
 		delivery.WithDeadLetterSink(svc.AsKReference(sink), "")),
 	)
-	f.Setup("install containersource", containersource.Install(cs, source.WithSink(channel.AsRef(name), "")))
-	f.Setup("install subscription", subscription.Install(feature.MakeRandomK8sName("subscription"),
+	f.Setup("install subscription", subscription.Install(sub,
 		subscription.WithChannel(channel.AsRef(name)),
 		createSubscriberFn(svc.AsKReference(failer), ""),
 	))
-
 	f.Setup("channel is ready", channel.IsReady(name))
-	f.Setup("containersource is ready", containersource.IsReady(cs))
+	f.Setup("subscription is ready", subscription.IsReady(sub))
 
+	f.Requirement("install containersource", containersource.Install(cs, source.WithSink(channel.AsRef(name), "")))
+	f.Requirement("containersource is ready", containersource.IsReady(cs))
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(name, channel.GVR()))
 
 	f.Assert("dls receives events", assert.OnStore(sink).
@@ -193,21 +195,13 @@ func AsDeadLetterSink(createSubscriberFn func(ref *duckv1.KReference, uri string
 
 func EventTransformation() *feature.Feature {
 	f := feature.NewFeature()
-	lib := feature.MakeRandomK8sName("lib")
 	channel1 := feature.MakeRandomK8sName("channel 1")
 	channel2 := feature.MakeRandomK8sName("channel 2")
 	subscription1 := feature.MakeRandomK8sName("subscription 1")
 	subscription2 := feature.MakeRandomK8sName("subscription 2")
 	prober := eventshub.NewProber()
 	prober.SetTargetResource(channel_impl.GVR(), channel1)
-
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	f.Setup("register event expectation", func(ctx context.Context, t feature.T) {
-		if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-			t.Fatalf("can not find event files: %v", err)
-		}
-	})
+	prober.SenderFullEvents(3)
 
 	f.Setup("install sink", prober.ReceiverInstall("sink"))
 	f.Setup("install transform service", prober.ReceiverInstall("transform", eventshub.ReplyWithTransformedEvent("transformed", "transformer", "")))
@@ -226,11 +220,8 @@ func EventTransformation() *feature.Feature {
 	f.Setup("subscription 2 is ready", subscription.IsReady(subscription2))
 	f.Setup("channel 1 is ready", channel_impl.IsReady(channel1))
 	f.Setup("channel 2 is ready", channel_impl.IsReady(channel2))
-	f.Setup("install source", prober.SenderInstall("source"))
-	f.Setup("event library is ready", eventlibrary.IsReady(lib))
 
-	f.Requirement("sender is finished", prober.SenderDone("source"))
-	f.Requirement("receiver is finished", prober.ReceiverDone("source", "sink"))
+	f.Requirement("install source", prober.SenderInstall("source"))
 
 	f.Assert("sink receives events", prober.AssertReceivedAll("source", "sink"))
 	f.Assert("events have passed through transform service", func(ctx context.Context, t feature.T) {
@@ -269,10 +260,8 @@ func SingleEventWithEncoding(encoding binding.Encoding) *feature.Feature {
 
 	f.Setup("subscription is ready", subscription.IsReady(sub))
 	f.Setup("channel is ready", channel_impl.IsReady(channel))
-	f.Setup("install source", prober.SenderInstall("source", eventshub.InputEventWithEncoding(event, encoding)))
 
-	f.Requirement("sender is finished", prober.SenderDone("source"))
-	f.Requirement("receiver is finished", prober.ReceiverDone("source", "sink"))
+	f.Requirement("install source", prober.SenderInstall("source", eventshub.InputEventWithEncoding(event, encoding)))
 
 	f.Assert("sink receives events", prober.AssertReceivedAll("source", "sink"))
 
@@ -341,6 +330,7 @@ func channelSubscriberUnreachable(createSubscriberFn func(ref *duckv1.KReference
 
 	sourceName := feature.MakeRandomK8sName("source")
 	channelName := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
 
 	ev := test.FullEvent()
 
@@ -348,10 +338,13 @@ func channelSubscriberUnreachable(createSubscriberFn func(ref *duckv1.KReference
 
 	f.Setup("install channel", channel_impl.Install(channelName, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
 
-	f.Setup("install subscription", subscription.Install(feature.MakeRandomK8sName("subscription"),
+	f.Setup("install subscription", subscription.Install(sub,
 		subscription.WithChannel(channel_impl.AsRef(channelName)),
 		createSubscriberFn(nil, "http://fake.svc.cluster.local"),
 	))
+	f.Setup("channel is ready", channel_impl.IsReady(channelName))
+	f.Setup("channel is addressable", channel_impl.IsAddressable(channelName))
+	f.Setup("subscription is ready", subscription.IsReady(sub))
 
 	f.Requirement("install source", eventshub.Install(
 		sourceName,
@@ -359,9 +352,6 @@ func channelSubscriberUnreachable(createSubscriberFn func(ref *duckv1.KReference
 		eventshub.InputEvent(ev),
 	))
 
-	f.Setup("channel is ready", channel_impl.IsReady(channelName))
-	f.Setup("channel is addressable", channel_impl.IsAddressable(channelName))
-
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(channelName, channel_impl.GVR()))
 
 	f.Assert("Receives dls extensions when subscriber is unreachable", eventasssert.OnStore(sink).
@@ -380,6 +370,7 @@ func channelSubscriberReturnedErrorNoData(createSubscriberFn func(ref *duckv1.KR
 	sourceName := feature.MakeRandomK8sName("source")
 	failer := feature.MakeRandomK8sName("failerWitdata")
 	channelName := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
 
 	ev := test.FullEvent()
 
@@ -392,10 +383,13 @@ func channelSubscriberReturnedErrorNoData(createSubscriberFn func(ref *duckv1.KR
 	))
 	f.Setup("install channel", channel_impl.Install(channelName, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
 
-	f.Setup("install subscription", subscription.Install(feature.MakeRandomK8sName("subscription"),
+	f.Setup("install subscription", subscription.Install(sub,
 		subscription.WithChannel(channel_impl.AsRef(channelName)),
 		createSubscriberFn(svc.AsKReference(failer), ""),
 	))
+	f.Setup("channel is ready", channel_impl.IsReady(channelName))
+	f.Setup("channel is addressable", channel_impl.IsAddressable(channelName))
+	f.Setup("subscription is ready", subscription.IsReady(sub))
 
 	f.Requirement("install source", eventshub.Install(
 		sourceName,
@@ -403,9 +397,6 @@ func channelSubscriberReturnedErrorNoData(createSubscriberFn func(ref *duckv1.KR
 		eventshub.InputEvent(ev),
 	))
 
-	f.Setup("channel is ready", channel_impl.IsReady(channelName))
-	f.Setup("channel is addressable", channel_impl.IsAddressable(channelName))
-
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(channelName, channel_impl.GVR()))
 
 	f.Assert("Receives dls extensions without errordata", assertEnhancedWithKnativeErrorExtensions(
@@ -429,6 +420,7 @@ func channelSubscriberReturnedErrorWithData(createSubscriberFn func(ref *duckv1.
 	sourceName := feature.MakeRandomK8sName("source")
 	failer := feature.MakeRandomK8sName("failerWitdata")
 	channelName := feature.MakeRandomK8sName("channel")
+	sub := feature.MakeRandomK8sName("subscription")
 
 	ev := test.FullEvent()
 
@@ -443,11 +435,13 @@ func channelSubscriberReturnedErrorWithData(createSubscriberFn func(ref *duckv1.
 		eventshub.DropEventsResponseBody(errorData),
 	))
 	f.Setup("install channel", channel_impl.Install(channelName, delivery.WithDeadLetterSink(svc.AsKReference(sink), "")))
-
-	f.Setup("install subscription", subscription.Install(feature.MakeRandomK8sName("subscription"),
+	f.Setup("install subscription", subscription.Install(sub,
 		subscription.WithChannel(channel_impl.AsRef(channelName)),
 		createSubscriberFn(svc.AsKReference(failer), ""),
 	))
+	f.Setup("channel is ready", channel_impl.IsReady(channelName))
+	f.Setup("channel is addressable", channel_impl.IsAddressable(channelName))
+	f.Setup("subscription is ready", subscription.IsReady(sub))
 
 	f.Requirement("install source", eventshub.Install(
 		sourceName,
@@ -455,9 +449,6 @@ func channelSubscriberReturnedErrorWithData(createSubscriberFn func(ref *duckv1.
 		eventshub.InputEvent(ev),
 	))
 
-	f.Setup("channel is ready", channel_impl.IsReady(channelName))
-	f.Setup("channel is addressable", channel_impl.IsAddressable(channelName))
-
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(channelName, channel_impl.GVR()))
 
 	f.Assert("Receives dls extensions with errordata Base64encoding", assertEnhancedWithKnativeErrorExtensions(
diff --git a/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go b/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go
index 619a1cc29..5c52da70b 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/channel/helpers.go
@@ -29,10 +29,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
-	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
-	messagingclientsetv1 "knative.dev/eventing/pkg/client/clientset/versioned/typed/messaging/v1"
-	eventingclient "knative.dev/eventing/pkg/client/injection/client"
-	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/pkg/apis/duck"
 	apiextensionsclient "knative.dev/pkg/client/injection/apiextensions/client"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
@@ -40,6 +36,11 @@ import (
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/state"
+
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingclientsetv1 "knative.dev/eventing/pkg/client/clientset/versioned/typed/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
 )
 
 func todo(ctx context.Context, t feature.T) {
@@ -74,8 +75,12 @@ func setChannelableName(name string) feature.StepFn {
 }
 
 func getChannelable(ctx context.Context, t feature.T) *duckv1.Channelable {
-	c := Client(ctx)
 	name := state.GetStringOrFail(ctx, t, ChannelableNameKey)
+	return getChannelableFromName(name, ctx, t)
+}
+
+func getChannelableFromName(name string, ctx context.Context, t feature.T) *duckv1.Channelable {
+	c := Client(ctx)
 
 	obj, err := c.ChannelImpl.Get(ctx, name, metav1.GetOptions{})
 	if err != nil {
diff --git a/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go b/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go
index 6b2ac4794..c7a03e735 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/trigger/feature.go
@@ -20,14 +20,15 @@ import (
 	"context"
 
 	"github.com/cloudevents/sdk-go/v2/test"
-	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/pingsource"
-	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/resources/svc"
+
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/pingsource"
+	"knative.dev/eventing/test/rekt/resources/trigger"
 )
 
 // This test is for avoiding regressions on the trigger dependency annotation functionality.
@@ -40,8 +41,8 @@ func TriggerDependencyAnnotation() *feature.Feature {
 	//Install the broker
 	brokerName := feature.MakeRandomK8sName("broker")
 	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
-	f.Requirement("broker is ready", broker.IsReady(brokerName))
-	f.Requirement("broker is addressable", broker.IsAddressable(brokerName))
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
 
 	psourcename := "test-ping-source-annotation"
 	dependencyAnnotation := `{"kind":"PingSource","name":"test-ping-source-annotation","apiVersion":"sources.knative.dev/v1"}`
@@ -61,7 +62,7 @@ func TriggerDependencyAnnotation() *feature.Feature {
 
 	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
 
-	f.Setup("install pingsource", func(ctx context.Context, t feature.T) {
+	f.Requirement("install pingsource", func(ctx context.Context, t feature.T) {
 		brokeruri, err := broker.Address(ctx, brokerName)
 		if err != nil {
 			t.Error("failed to get address of broker", err)
@@ -73,7 +74,7 @@ func TriggerDependencyAnnotation() *feature.Feature {
 		}
 		pingsource.Install(psourcename, cfg...)(ctx, t)
 	})
-	f.Setup("PingSource goes ready", pingsource.IsReady(psourcename))
+	f.Requirement("PingSource goes ready", pingsource.IsReady(psourcename))
 
 	f.Stable("pingsource as event source to test trigger with annotations").
 		Must("delivers events on broker with URI", assert.OnStore(sink).MatchEvent(
diff --git a/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
index bc2953c7d..fbdb89dcf 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
@@ -18,14 +18,12 @@ package trigger
 
 import (
 	"context"
-	"fmt"
 
 	"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/delivery"
-	"knative.dev/eventing/test/rekt/resources/eventlibrary"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 )
 
@@ -45,14 +43,7 @@ func SourceToTriggerSinkWithDLS() *feature.Feature {
 
 	prober := eventshub.NewProber()
 	prober.SetTargetResource(broker.GVR(), brokerName)
-
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
+	prober.SenderFullEvents(3)
 
 	f.Setup("install broker", broker.Install(brokerName, broker.WithEnvConfig()...))
 
@@ -70,10 +61,7 @@ func SourceToTriggerSinkWithDLS() *feature.Feature {
 	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
 
 	// Install sender.
-	f.Setup("install source", prober.SenderInstall("source"))
-
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("install source", prober.SenderInstall("source"))
 
 	// Assert events ended up where we expected.
 	f.Stable("trigger with DLS").
@@ -100,14 +88,7 @@ func SourceToTriggerSinkWithDLSDontUseBrokers() *feature.Feature {
 
 	prober := eventshub.NewProber()
 	prober.SetTargetResource(broker.GVR(), brokerName)
-
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
+	prober.SenderFullEvents(3)
 
 	// Setup Probes
 	f.Setup("install trigger recorder", prober.ReceiverInstall(triggerSinkName))
@@ -132,10 +113,7 @@ func SourceToTriggerSinkWithDLSDontUseBrokers() *feature.Feature {
 	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
 
 	// Install events after topology is ready.
-	f.Setup("install source", prober.SenderInstall("source"))
-
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone("source"))
+	f.Requirement("install source", prober.SenderInstall("source"))
 
 	// Assert events ended up where we expected.
 	f.Stable("trigger with a valid DLS ref").
@@ -163,13 +141,7 @@ func BadTriggerDoesNotAffectOkTrigger() *feature.Feature {
 	sink := feature.MakeRandomK8sName("sink")
 	source := feature.MakeRandomK8sName("source")
 
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
+	prober.SenderFullEvents(3)
 
 	// Setup Probes
 	f.Setup("install dlq", prober.ReceiverInstall(dlq))
@@ -192,11 +164,6 @@ func BadTriggerDoesNotAffectOkTrigger() *feature.Feature {
 	// Install events after data plane is ready.
 	f.Requirement("install source", prober.SenderInstall(source))
 
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone(source))
-	f.Requirement("receiver 1 is finished", prober.ReceiverDone(source, dlq))
-	f.Requirement("receiver 2 is finished", prober.ReceiverDone(source, sink))
-
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
 		Must("accepted all events", prober.AssertSentAll(source)).
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go b/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go
index 0783c9e7a..473fa6896 100644
--- a/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go
+++ b/vendor/knative.dev/eventing/test/rekt/resources/channel_impl/channel_impl.go
@@ -19,14 +19,17 @@ package channel_impl
 import (
 	"context"
 	"embed"
+	"encoding/json"
 	"log"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/injection/clients/dynamicclient"
@@ -108,21 +111,35 @@ func IsAddressable(name string, timing ...time.Duration) feature.StepFn {
 func HasDeadLetterSinkURI(name string, gvr schema.GroupVersionResource) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
 		ns := environment.FromContext(ctx).Namespace()
-		ch, err := dynamicclient.Get(ctx).
-			Resource(gvr).
-			Namespace(ns).
-			Get(ctx, name, metav1.GetOptions{})
+		interval, timeout := environment.PollTimingsFromContext(ctx)
+		var lastState *eventingduck.Channelable
+		err := wait.Poll(interval, timeout, func() (done bool, err error) {
+			ch, err := dynamicclient.Get(ctx).
+				Resource(gvr).
+				Namespace(ns).
+				Get(ctx, name, metav1.GetOptions{})
+			if apierrors.IsNotFound(err) {
+				return false, nil
+			}
+			if err != nil {
+				t.Fatalf("failed to get %s/%s channel: %v", ns, name, err)
+			}
+
+			channelable := &eventingduck.Channelable{}
+			if err := runtime.DefaultUnstructuredConverter.FromUnstructured(ch.UnstructuredContent(), channelable); err != nil {
+				t.Fatal(err)
+			}
+			lastState = channelable
+
+			if channelable.Status.DeadLetterSinkURI.String() == "" {
+				return false, nil
+			}
+
+			return true, nil
+		})
 		if err != nil {
-			t.Fatalf("failed to get %s/%s channel: %v", ns, name, err)
-		}
-
-		channelable := &eventingduck.Channelable{}
-		if err := runtime.DefaultUnstructuredConverter.FromUnstructured(ch.UnstructuredContent(), channelable); err != nil {
-			t.Fatal(err)
-		}
-
-		if channelable.Status.DeadLetterSinkURI.String() == "" {
-			t.Fatalf("channel %s/%s has no dead letter sink uri in the status", ns, name)
+			bytes, _ := json.MarshalIndent(lastState, "", "  ")
+			t.Errorf("failed to verify channel has dead letter sink: %w, last state:\n%s", err, string(bytes))
 		}
 	}
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/README.md b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/README.md
deleted file mode 100644
index a96f3c2c3..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/README.md
+++ /dev/null
@@ -1,20 +0,0 @@
-# Event Library
-
-This installs is a simple file server from the test_images/event-library
-package. The local `events` dir is symlinked into the `kodata` dir.
-
-Example local test (from the root of the project):
-
-```shell
-KO_DATA_PATH=./test/test_images/event-library/kodata \
-  go run ./test/test_images/event-library
-```
-
-Then access the files path-ed exactly as they are rooted from
-`resources/eventlibrary` dir:
-
-```shell
-curl localhost:8080/events/three.ce
-```
-
-Results in the contents of `./events/three.ce`.
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/eventlibrary.go b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/eventlibrary.go
deleted file mode 100644
index 9e3815726..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/eventlibrary.go
+++ /dev/null
@@ -1,79 +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 eventlibrary
-
-import (
-	"context"
-	"embed"
-	"log"
-	"path"
-	"runtime"
-
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/reconciler-test/pkg/environment"
-	"knative.dev/reconciler-test/pkg/feature"
-	"knative.dev/reconciler-test/pkg/k8s"
-	"knative.dev/reconciler-test/pkg/manifest"
-	"knative.dev/reconciler-test/resources/svc"
-)
-
-//go:embed *.yaml
-var yaml embed.FS
-
-// Install
-func Install(name string) feature.StepFn {
-	cfg := map[string]interface{}{
-		"name": name,
-	}
-
-	return func(ctx context.Context, t feature.T) {
-		if err := registerImage(ctx); err != nil {
-			t.Fatal(err)
-		}
-		manifest.PodSecurityCfgFn(ctx, t)(cfg)
-		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
-			t.Fatal(err)
-		}
-	}
-}
-
-func IsReady(name string) feature.StepFn {
-	return func(ctx context.Context, t feature.T) {
-		k8s.WaitForPodRunningOrFail(ctx, t, name)
-		k8s.WaitForServiceEndpointsOrFail(ctx, t, name, 1)
-	}
-}
-
-func PathFor(file string) string {
-	_, filename, _, _ := runtime.Caller(0)
-	log.Println("FILENAME: ", filename)
-
-	return path.Join(path.Dir(filename), file)
-}
-
-// AsKReference returns a KReference for the svc the event library uses to be
-// addressable.
-func AsKReference(name string) *duckv1.KReference {
-	return svc.AsKReference(name)
-}
-
-func registerImage(ctx context.Context) error {
-	im := manifest.ImagesFromFS(ctx, yaml)
-	reg := environment.RegisterPackage(im...)
-	_, err := reg(ctx, environment.FromContext(ctx))
-	return err
-}
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/eventlibrary.yaml b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/eventlibrary.yaml
deleted file mode 100644
index f525ec977..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/eventlibrary.yaml
+++ /dev/null
@@ -1,63 +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: library-{{ .name }}
-spec:
-  {{ if .podSecurityContext }}
-  securityContext:
-    runAsNonRoot: {{ .podSecurityContext.runAsNonRoot }}
-    seccompProfile:
-      type: {{ .podSecurityContext.seccompProfile.type }}
-  {{ end }}
-  restartPolicy: "Never"
-  containers:
-    - name: library
-      image: ko://knative.dev/eventing/test/test_images/event-library
-      imagePullPolicy: "IfNotPresent"
-      {{ if .containerSecurityContext }}
-      securityContext:
-        capabilities:
-          {{ if .containerSecurityContext.capabilities.drop }}
-          drop:
-            {{ range $_, $value := .containerSecurityContext.capabilities.drop }}
-            - {{ $value }}
-            {{ end }}
-          {{ end }}
-          {{ if .containerSecurityContext.capabilities.add }}
-          add:
-            {{ range $_, $value := .containerSecurityContext.capabilities.add }}
-            - {{ $value }}
-            {{ end }}
-          {{ end }}
-        allowPrivilegeEscalation: {{ .containerSecurityContext.allowPrivilegeEscalation }}
-      {{ end }}
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: {{ .name }}
-  namespace: {{ .namespace }}
-spec:
-  selector:
-    app: library-{{ .name }}
-  ports:
-    - protocol: TCP
-      port: 80
-      targetPort: 8080
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/doc.go b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/doc.go
deleted file mode 100644
index 0a7b6a8ab..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/doc.go
+++ /dev/null
@@ -1,21 +0,0 @@
-//go:build cloudevents
-// +build cloudevents
-
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// package events is here to allow us to vendor the .ce files here.
-package events
diff --git a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/three.ce b/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/three.ce
deleted file mode 100644
index 49eaa290d..000000000
--- a/vendor/knative.dev/eventing/test/rekt/resources/eventlibrary/events/three.ce
+++ /dev/null
@@ -1,23 +0,0 @@
-ContextAttributes:
-  specversion: "1.0"
-  type: a.type
-  id: abc-123-1
-  source: foo/bar
-  datacontenttype: application/json
-Data: '{"hello":"world"}'
----
-ContextAttributes:
-  specversion: "1.0"
-  type: a.type
-  id: abc-123-2
-  source: foo/bar
-  datacontenttype: application/json
-Data: '{"hello":"world"}'
----
-ContextAttributes:
-  specversion: "1.0"
-  type: a.type
-  id: abc-123-3
-  source: foo/bar
-  datacontenttype: application/json
-Data: '{"hello":"world"}'
diff --git a/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/doc.go b/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/doc.go
deleted file mode 100644
index 0a7b6a8ab..000000000
--- a/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/doc.go
+++ /dev/null
@@ -1,21 +0,0 @@
-//go:build cloudevents
-// +build cloudevents
-
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// package events is here to allow us to vendor the .ce files here.
-package events
diff --git a/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/three.ce b/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/three.ce
deleted file mode 100644
index 49eaa290d..000000000
--- a/vendor/knative.dev/eventing/test/test_images/event-library/kodata/events/three.ce
+++ /dev/null
@@ -1,23 +0,0 @@
-ContextAttributes:
-  specversion: "1.0"
-  type: a.type
-  id: abc-123-1
-  source: foo/bar
-  datacontenttype: application/json
-Data: '{"hello":"world"}'
----
-ContextAttributes:
-  specversion: "1.0"
-  type: a.type
-  id: abc-123-2
-  source: foo/bar
-  datacontenttype: application/json
-Data: '{"hello":"world"}'
----
-ContextAttributes:
-  specversion: "1.0"
-  type: a.type
-  id: abc-123-3
-  source: foo/bar
-  datacontenttype: application/json
-Data: '{"hello":"world"}'
diff --git a/vendor/knative.dev/eventing/test/test_images/event-library/main.go b/vendor/knative.dev/eventing/test/test_images/event-library/main.go
deleted file mode 100644
index ee1388b0c..000000000
--- a/vendor/knative.dev/eventing/test/test_images/event-library/main.go
+++ /dev/null
@@ -1,45 +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 (
-	"fmt"
-	"log"
-	"net/http"
-
-	"github.com/kelseyhightower/envconfig"
-	_ "knative.dev/pkg/system/testing"
-)
-
-type envConfig struct {
-	DataPath string `envconfig:"KO_DATA_PATH" default:"/var/run/ko/" required:"true"`
-	Port     int    `envconfig:"PORT" default:"8080" required:"true"`
-}
-
-var env envConfig
-
-func main() {
-	if err := envconfig.Process("", &env); err != nil {
-		log.Fatalf("failed to process env var: %s", err)
-	}
-
-	m := http.NewServeMux()
-	m.Handle("/", http.FileServer(http.Dir(env.DataPath)))
-
-	log.Println("Listening on", env.Port)
-	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", env.Port), m))
-}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
index 9edd93158..5f4ef72b2 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
@@ -97,7 +97,8 @@ func (p *prober) Verify() (eventErrs []error, eventsSent int) {
 	}
 	for i, t := range report.Thrown.Duplicated {
 		if p.config.OnDuplicate == Warn {
-			p.log.Warn("Duplicate events: ", t)
+			// Print at info level to prevent excessive stacktraces.
+			p.log.Info("WARNING: Duplicate:", t)
 		} else if p.config.OnDuplicate == Error {
 			eventErrs = append(eventErrs, errors.New(t))
 		}
@@ -141,7 +142,7 @@ func (p *prober) getStepNoFromMsg(message string) (string, error) {
 }
 
 func (p *prober) getTraceForStepEvent(eventNo string) []byte {
-	p.log.Infof("Fetching trace for Step event #%s", eventNo)
+	p.log.Debugf("Fetching trace for Step event #%s", eventNo)
 	query := fmt.Sprintf("step=%s and cloudevents.type=%s and target=%s",
 		eventNo, event.StepType, fmt.Sprintf(forwarderTargetFmt, p.client.Namespace))
 	trace, err := event.FindTrace(query)
diff --git a/vendor/knative.dev/pkg/test/logstream/v2/interface.go b/vendor/knative.dev/pkg/test/logstream/v2/interface.go
new file mode 100644
index 000000000..cf2359a5a
--- /dev/null
+++ b/vendor/knative.dev/pkg/test/logstream/v2/interface.go
@@ -0,0 +1,34 @@
+/*
+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 logstream
+
+type (
+	// Canceler is the type of a function returned when a logstream is
+	// started to be deferred so that the logstream can be stopped when
+	// the test is complete.
+	Canceler func()
+
+	// Callback is invoked after pod logs are transformed
+	Callback func(string, ...interface{})
+
+	// Source allows you to create streams for a given resource name
+	Source interface {
+		// Start a log stream for the given resource name and invoke
+		// the callback with the processed log
+		StartStream(name string, l Callback) (Canceler, error)
+	}
+)
diff --git a/vendor/knative.dev/pkg/test/logstream/v2/stream.go b/vendor/knative.dev/pkg/test/logstream/v2/stream.go
new file mode 100644
index 000000000..1a37a54d2
--- /dev/null
+++ b/vendor/knative.dev/pkg/test/logstream/v2/stream.go
@@ -0,0 +1,320 @@
+/*
+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 logstream
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"sync"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/pkg/ptr"
+)
+
+// New creates a new log source. The source namespaces must be configured through
+// log source options.
+func New(ctx context.Context, c kubernetes.Interface, opts ...func(*logSource)) Source {
+	s := &logSource{
+		ctx:         ctx,
+		kc:          c,
+		keys:        make(map[string]Callback, 1),
+		filterLines: true, // Filtering log lines by the watched resource name is enabled by default.
+	}
+	for _, opt := range opts {
+		opt(s)
+	}
+	return s
+}
+
+// WithNamespaces configures namespaces for log stream.
+func WithNamespaces(namespaces ...string) func(*logSource) {
+	return func(s *logSource) {
+		s.namespaces = namespaces
+	}
+}
+
+// WithLineFiltering configures whether log lines will be filtered by
+// the resource name.
+func WithLineFiltering(enabled bool) func(*logSource) {
+	return func(s *logSource) {
+		s.filterLines = enabled
+	}
+}
+
+// WithPodPrefixes specifies which Pods will be included in the
+// log stream through the provided prefixes. If no prefixes are
+// configured then logs from all Pods in the configured namespaces will
+// be streamed.
+func WithPodPrefixes(podPrefixes ...string) func(*logSource) {
+	return func(s *logSource) {
+		s.podPrefixes = podPrefixes
+	}
+}
+
+func FromNamespaces(ctx context.Context, c kubernetes.Interface, namespaces []string, opts ...func(*logSource)) Source {
+	sOpts := []func(*logSource){WithNamespaces(namespaces...)}
+	sOpts = append(sOpts, opts...)
+	return New(ctx, c, sOpts...)
+}
+
+func FromNamespace(ctx context.Context, c kubernetes.Interface, namespace string, opts ...func(*logSource)) Source {
+	return FromNamespaces(ctx, c, []string{namespace}, opts...)
+}
+
+type logSource struct {
+	namespaces []string
+	kc         kubernetes.Interface
+	ctx        context.Context
+
+	m           sync.RWMutex
+	once        sync.Once
+	keys        map[string]Callback
+	filterLines bool
+	podPrefixes []string
+	watchErr    error
+}
+
+func (s *logSource) StartStream(name string, l Callback) (Canceler, error) {
+	s.once.Do(func() { s.watchErr = s.watchPods() })
+	if s.watchErr != nil {
+		return nil, fmt.Errorf("failed to watch pods in one of the namespace(s) %q: %w", s.namespaces, s.watchErr)
+	}
+
+	// Register a key
+	s.m.Lock()
+	defer s.m.Unlock()
+	s.keys[name] = l
+
+	// Return a function that unregisters that key.
+	return func() {
+		s.m.Lock()
+		defer s.m.Unlock()
+		delete(s.keys, name)
+	}, nil
+}
+
+func (s *logSource) watchPods() error {
+	if len(s.namespaces) == 0 {
+		return errors.New("namespaces for logstream not configured")
+	}
+	for _, ns := range s.namespaces {
+		wi, err := s.kc.CoreV1().Pods(ns).Watch(s.ctx, metav1.ListOptions{})
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			defer wi.Stop()
+			watchedPods := sets.NewString()
+
+			for {
+				select {
+				case <-s.ctx.Done():
+					return
+				case ev := <-wi.ResultChan():
+					// We have reports of this being randomly nil.
+					if ev.Object == nil || reflect.ValueOf(ev.Object).IsNil() {
+						continue
+					}
+					p, ok := ev.Object.(*corev1.Pod)
+					if !ok {
+						// The Watch interface can return errors via the channel as *metav1.Status.
+						// Log those to get notified that loglines might be missing but don't crash.
+						s.handleGenericLine([]byte(fmt.Sprintf("unexpected event: %v", p)), "no-pod", "no-container")
+						continue
+					}
+					switch ev.Type {
+					case watch.Deleted:
+						watchedPods.Delete(p.Name)
+					case watch.Added, watch.Modified:
+						if !watchedPods.Has(p.Name) && isPodReady(p) && s.matchesPodPrefix(p.Name) {
+							watchedPods.Insert(p.Name)
+							s.startForPod(p)
+						}
+					}
+
+				}
+			}
+		}()
+	}
+
+	return nil
+}
+
+func (s *logSource) matchesPodPrefix(name string) bool {
+	if len(s.podPrefixes) == 0 {
+		// Pod prefixes are not configured => always match.
+		return true
+	}
+	for _, p := range s.podPrefixes {
+		if strings.Contains(name, p) {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *logSource) startForPod(pod *corev1.Pod) {
+	// Grab data from all containers in the pods.  We need this in case
+	// an envoy sidecar is injected for mesh installs.  This should be
+	// equivalent to --all-containers.
+	for _, container := range pod.Spec.Containers {
+		// Required for capture below.
+		psn, pn, cn := pod.Namespace, pod.Name, container.Name
+
+		handleLine := s.handleLine
+		if wellKnownContainers.Has(cn) || !s.filterLines {
+			// Specialcase logs from chaosduck, queueproxy etc.
+			// - ChaosDuck logs enable easy
+			//   monitoring of killed pods throughout all tests.
+			// - QueueProxy logs enable
+			//   debugging troubleshooting data plane request handling issues.
+			handleLine = s.handleGenericLine
+		}
+
+		go func() {
+			options := &corev1.PodLogOptions{
+				Container: cn,
+				// Follow directs the API server to continuously stream logs back.
+				Follow: true,
+				// Only return new logs (this value is being used for "epsilon").
+				SinceSeconds: ptr.Int64(1),
+			}
+
+			req := s.kc.CoreV1().Pods(psn).GetLogs(pn, options)
+			stream, err := req.Stream(context.Background())
+			if err != nil {
+				s.handleGenericLine([]byte(err.Error()), pn, cn)
+				return
+			}
+			defer stream.Close()
+			// Read this container's stream.
+			for scanner := bufio.NewScanner(stream); scanner.Scan(); {
+				handleLine(scanner.Bytes(), pn, cn)
+			}
+			// Pods get killed with chaos duck, so logs might end
+			// before the test does. So don't report an error here.
+		}()
+	}
+}
+
+func isPodReady(p *corev1.Pod) bool {
+	if p.Status.Phase == corev1.PodRunning && p.DeletionTimestamp == nil {
+		for _, cond := range p.Status.Conditions {
+			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+const (
+	// timeFormat defines a simple timestamp with millisecond granularity
+	timeFormat = "15:04:05.000"
+	// ChaosDuck is the well known name for the chaosduck.
+	ChaosDuck = "chaosduck"
+	// QueueProxy is the well known name for the queueproxy.
+	QueueProxy = "queueproxy"
+)
+
+// Names of well known containers that do not produce nicely formatted logs that
+// could be easily filtered and parsed by handleLine. Logs from these containers
+// are captured without filtering.
+var wellKnownContainers = sets.NewString(ChaosDuck, QueueProxy)
+
+func (s *logSource) handleLine(l []byte, pod string, _ string) {
+	// This holds the standard structure of our logs.
+	var line struct {
+		Level      string    `json:"severity"`
+		Timestamp  time.Time `json:"timestamp"`
+		Controller string    `json:"knative.dev/controller"`
+		Caller     string    `json:"caller"`
+		Key        string    `json:"knative.dev/key"`
+		Message    string    `json:"message"`
+		Error      string    `json:"error"`
+
+		// TODO(mattmoor): Parse out more context.
+	}
+	if err := json.Unmarshal(l, &line); err != nil {
+		// Ignore malformed lines.
+		return
+	}
+	if line.Key == "" {
+		return
+	}
+
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	for name, logf := range s.keys {
+		// TODO(mattmoor): Do a slightly smarter match.
+		if !strings.Contains(line.Key, "/"+name) {
+			continue
+		}
+
+		// We also get logs not from controllers (activator, autoscaler).
+		// So replace controller string in them with their callsite.
+		site := line.Controller
+		if site == "" {
+			site = line.Caller
+		}
+		func() {
+			defer func() {
+				if err := recover(); err != nil {
+					logf("Invalid log format for pod %s: %s", pod, string(l))
+				}
+			}()
+			// E 15:04:05.000 webhook-699b7b668d-9smk2 [route-controller] [default/testroute-xyz] this is my message
+			msg := fmt.Sprintf("%s %s %s [%s] [%s] %s",
+				strings.ToUpper(string(line.Level[0])),
+				line.Timestamp.Format(timeFormat),
+				pod,
+				site,
+				line.Key,
+				line.Message)
+
+			if line.Error != "" {
+				msg += " err=" + line.Error
+			}
+
+			logf(msg)
+		}()
+	}
+}
+
+// handleGenericLine prints the given logline to all active tests as it cannot be parsed
+// and/or doesn't contain any correlation data (like the chaosduck for example).
+func (s *logSource) handleGenericLine(l []byte, pod string, cn string) {
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	for _, logf := range s.keys {
+		// I 15:04:05.000 webhook-699b7b668d-9smk2 this is my message
+		logf("I %s %s %s %s", time.Now().Format(timeFormat), pod, cn, string(l))
+	}
+}
diff --git a/vendor/knative.dev/pkg/test/upgrade/functions.go b/vendor/knative.dev/pkg/test/upgrade/functions.go
index 98bf28340..fe4965d80 100644
--- a/vendor/knative.dev/pkg/test/upgrade/functions.go
+++ b/vendor/knative.dev/pkg/test/upgrade/functions.go
@@ -17,38 +17,39 @@ limitations under the License.
 package upgrade
 
 import (
-	"bytes"
-	"strings"
+	"testing"
 	"time"
 
-	"go.uber.org/zap/zapcore"
-
 	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest"
 )
 
 // Execute the Suite of upgrade tests with a Configuration given.
+// When the suite includes Continual tests the number of logical CPUs
+// usable by the test process must be at least <NUMBER OF CONTINUAL TESTS> + 1.
+// The -parallel test flag or GOMAXPROCS environment variable might be
+// used to adjust the settings.
 func (s *Suite) Execute(c Configuration) {
-	l, err := c.logger(c.T)
-	if err != nil {
-		c.T.Fatal("Failed to build logger:", err)
-		return
-	}
+	l := c.logger(c.T)
 	se := suiteExecution{
 		suite:         enrichSuite(s),
 		configuration: c,
-		failed:        false,
 	}
 	l.Info("🏃 Running upgrade test suite...")
 
 	se.execute()
 
-	if !se.failed {
+	if !c.T.Failed() {
 		l.Info("🥳🎉 Success! Upgrade suite completed without errors.")
 	} else {
 		l.Error("💣🤬💔️ Upgrade suite have failed!")
 	}
 }
 
+func (c Configuration) logger(t *testing.T) *zap.SugaredLogger {
+	return zaptest.NewLogger(t, zaptest.WrapOptions(c.Options...)).Sugar()
+}
+
 // NewOperation creates a new upgrade operation or test.
 func NewOperation(name string, handler func(c Context)) Operation {
 	return &simpleOperation{name: name, handler: handler}
@@ -56,14 +57,14 @@ func NewOperation(name string, handler func(c Context)) Operation {
 
 // NewBackgroundVerification is convenience function to easily setup a
 // background operation that will setup environment and then verify environment
-// status after receiving a StopEvent.
+// status after receiving a stop event.
 func NewBackgroundVerification(name string, setup func(c Context), verify func(c Context)) BackgroundOperation {
 	return NewBackgroundOperation(name, setup, func(bc BackgroundContext) {
 		WaitForStopEvent(bc, WaitForStopEventConfiguration{
 			Name: name,
-			OnStop: func(event StopEvent) {
+			OnStop: func() {
 				verify(Context{
-					T:   event.T,
+					T:   bc.T,
 					Log: bc.Log,
 				})
 			},
@@ -90,8 +91,9 @@ func NewBackgroundOperation(name string, setup func(c Context),
 func WaitForStopEvent(bc BackgroundContext, w WaitForStopEventConfiguration) {
 	for {
 		select {
-		case stopEvent := <-bc.Stop:
-			handleStopEvent(stopEvent, bc, w)
+		case <-bc.Stop:
+			bc.Log.Debugf("%s received a stop event", w.Name)
+			w.OnStop()
 			return
 		default:
 			w.OnWait(bc, w)
@@ -100,32 +102,6 @@ func WaitForStopEvent(bc BackgroundContext, w WaitForStopEventConfiguration) {
 	}
 }
 
-// Name returns a friendly human readable text.
-func (s *StopEvent) Name() string {
-	return s.name
-}
-
-func handleStopEvent(
-	se StopEvent,
-	bc BackgroundContext,
-	wc WaitForStopEventConfiguration,
-) {
-	bc.Log.Debugf("%s have received a stop event: %s", wc.Name, se.Name())
-
-	logFn := se.logger.Info
-	logFn(wrapLog(bc.logBuffer.Dump()))
-
-	defer func() {
-		defer close(se.Finished)
-		if se.T.Failed() {
-			logFn = se.logger.Error
-		}
-		logFn(wrapLog(bc.logBuffer.Dump()))
-	}()
-
-	wc.OnStop(se)
-}
-
 func enrichSuite(s *Suite) *enrichedSuite {
 	es := &enrichedSuite{
 		installations: s.Installations,
@@ -139,7 +115,7 @@ func enrichSuite(s *Suite) *enrichedSuite {
 	for i, test := range s.Tests.Continual {
 		es.tests.continual[i] = stoppableOperation{
 			BackgroundOperation: test,
-			stop:                make(chan StopEvent),
+			stop:                make(chan struct{}),
 		}
 	}
 	return es
@@ -168,64 +144,9 @@ func (s *simpleBackgroundOperation) Setup() func(c Context) {
 
 // Handler will be executed in background while upgrade/downgrade is being
 // executed. It can be used to constantly validate environment during that
-// time and/or wait for StopEvent being sent. After StopEvent is received
+// time and/or wait for a stop event being sent. After a stop event is received
 // user should validate environment, clean up resources, and report found
-// issues to testing.T forwarded in StepEvent.
+// issues to testing.T forwarded in BackgroundContext.
 func (s *simpleBackgroundOperation) Handler() func(bc BackgroundContext) {
 	return s.handler
 }
-
-func (b *threadSafeBuffer) Read(p []byte) (n int, err error) {
-	b.Mutex.Lock()
-	defer b.Mutex.Unlock()
-	return b.Buffer.Read(p)
-}
-
-func (b *threadSafeBuffer) Write(p []byte) (n int, err error) {
-	b.Mutex.Lock()
-	defer b.Mutex.Unlock()
-	return b.Buffer.Write(p)
-}
-
-// Dump returns the previous content of the buffer and creates a new
-// empty buffer for future writes.
-func (b *threadSafeBuffer) Dump() string {
-	b.Mutex.Lock()
-	defer b.Mutex.Unlock()
-	buf := b.Buffer.String()
-	b.Buffer = bytes.Buffer{}
-	return buf
-}
-
-// newInMemoryLoggerBuffer creates a logger that writes logs into a byte buffer.
-// This byte buffer is returned and can be used to process the logs at later stage.
-func newInMemoryLoggerBuffer(config Configuration) (*zap.Logger, *threadSafeBuffer) {
-	logConfig := config.logConfig()
-	buf := &threadSafeBuffer{}
-	core := zapcore.NewCore(
-		zapcore.NewConsoleEncoder(logConfig.Config.EncoderConfig),
-		zapcore.AddSync(buf),
-		logConfig.Config.Level)
-
-	var opts []zap.Option
-	if logConfig.Config.Development {
-		opts = append(opts, zap.Development())
-	}
-	if !logConfig.Config.DisableCaller {
-		opts = append(opts, zap.AddCaller())
-	}
-	stackLevel := zap.ErrorLevel
-	if logConfig.Config.Development {
-		stackLevel = zap.WarnLevel
-	}
-	if !logConfig.Config.DisableStacktrace {
-		opts = append(opts, zap.AddStacktrace(stackLevel))
-	}
-	return zap.New(core, opts...), buf
-}
-
-func wrapLog(log string) string {
-	s := strings.Join(strings.Split(log, "\n"), "\n⏳ ")
-	return "Rewind of background logs (prefix ⏳):\n" +
-		"⏳ " + strings.TrimSuffix(s, "⏳ ")
-}
diff --git a/vendor/knative.dev/pkg/test/upgrade/logger.go b/vendor/knative.dev/pkg/test/upgrade/logger.go
deleted file mode 100644
index 903544286..000000000
--- a/vendor/knative.dev/pkg/test/upgrade/logger.go
+++ /dev/null
@@ -1,110 +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 upgrade
-
-import (
-	"errors"
-	"fmt"
-	"net/url"
-	"strings"
-	"testing"
-
-	"go.uber.org/zap"
-)
-
-const testingTScheme = "testing-t"
-
-var (
-	// ErrInvalidTestingOutputMethod when given invalid testing output method.
-	ErrInvalidTestingOutputMethod = errors.New("invalid testing output method")
-	// ErrNoTestingTRegisteredForTest when no testing.TB registered for test.
-	ErrNoTestingTRegisteredForTest = errors.New("no testing.TB registered for test")
-
-	testingTeeBees = make(map[string]testing.TB)
-)
-
-func (c Configuration) logger(tb testing.TB) (*zap.SugaredLogger, error) {
-	// TODO(mgencur): Remove when dependent repositories use LogConfig instead of Log.
-	// This is for backwards compatibility.
-	if c.Log != nil {
-		return c.Log.Sugar(), nil
-	}
-
-	testName := tb.Name()
-	testingTeeBees[testName] = tb
-	cfg := c.logConfig().Config
-	outs := []string{fmt.Sprintf("%s://Log/%s", testingTScheme, testName)}
-	if isStandardOutputOnly(cfg.OutputPaths) {
-		cfg.OutputPaths = outs
-	} else {
-		cfg.OutputPaths = append(cfg.OutputPaths, outs...)
-	}
-	errOuts := []string{fmt.Sprintf("%s://Error/%s", testingTScheme, testName)}
-	if isStandardOutputOnly(cfg.ErrorOutputPaths) {
-		cfg.ErrorOutputPaths = errOuts
-	} else {
-		cfg.ErrorOutputPaths = append(cfg.ErrorOutputPaths, errOuts...)
-	}
-	err := zap.RegisterSink(testingTScheme, func(url *url.URL) (zap.Sink, error) {
-		var ok bool
-		testName = strings.TrimPrefix(url.Path, "/")
-		tb, ok = testingTeeBees[testName]
-		if !ok {
-			return nil, fmt.Errorf("%w: %s", ErrNoTestingTRegisteredForTest, testName)
-		}
-		switch url.Host {
-		case "Log":
-			return sink{writer: tb.Log}, nil
-		case "Error":
-			return sink{writer: tb.Error}, nil
-		}
-		return nil, fmt.Errorf("%w: %s", ErrInvalidTestingOutputMethod, url.Host)
-	})
-	if err != nil && !strings.HasPrefix(err.Error(), "sink factory already registered") {
-		return nil, err
-	}
-	var logger *zap.Logger
-	logger, err = cfg.Build(c.logConfig().Options...)
-	if err != nil {
-		return nil, err
-	}
-	return logger.Sugar(), nil
-}
-
-type sink struct {
-	writer func(args ...interface{})
-}
-
-func (t sink) Write(bytes []byte) (n int, err error) {
-	msg := string(bytes)
-	t.writer(msg)
-	return len(bytes), nil
-}
-
-func (t sink) Sync() error {
-	return nil
-}
-
-func (t sink) Close() error {
-	return nil
-}
-
-// isStandardOutputOnly checks the output paths and returns true, if they
-// contain only standard outputs.
-func isStandardOutputOnly(paths []string) bool {
-	return len(paths) == 1 && (paths[0] == "stderr" || paths[0] == "stdout")
-}
diff --git a/vendor/knative.dev/pkg/test/upgrade/private_types.go b/vendor/knative.dev/pkg/test/upgrade/private_types.go
index d301b5d68..445f9898a 100644
--- a/vendor/knative.dev/pkg/test/upgrade/private_types.go
+++ b/vendor/knative.dev/pkg/test/upgrade/private_types.go
@@ -16,15 +16,9 @@ limitations under the License.
 
 package upgrade
 
-import (
-	"bytes"
-	"sync"
-)
-
 type suiteExecution struct {
 	suite         *enrichedSuite
 	configuration Configuration
-	failed        bool
 }
 
 type enrichedSuite struct {
@@ -41,7 +35,7 @@ type enrichedTests struct {
 
 type stoppableOperation struct {
 	BackgroundOperation
-	stop chan StopEvent
+	stop chan struct{}
 }
 
 type operationGroup struct {
@@ -63,10 +57,3 @@ type simpleBackgroundOperation struct {
 	setup   func(c Context)
 	handler func(bc BackgroundContext)
 }
-
-// threadSafeBuffer avoids race conditions on bytes.Buffer.
-// See: https://stackoverflow.com/a/36226525/844449
-type threadSafeBuffer struct {
-	bytes.Buffer
-	sync.Mutex
-}
diff --git a/vendor/knative.dev/pkg/test/upgrade/steps.go b/vendor/knative.dev/pkg/test/upgrade/steps.go
index e4c270c0a..d5e1a71a9 100644
--- a/vendor/knative.dev/pkg/test/upgrade/steps.go
+++ b/vendor/knative.dev/pkg/test/upgrade/steps.go
@@ -22,8 +22,8 @@ import (
 
 const skippingOperationTemplate = `Skipping "%s" as previous operation have failed`
 
-func (se *suiteExecution) installingBase(num int) {
-	se.processOperationGroup(operationGroup{
+func (se *suiteExecution) installingBase(t *testing.T, num int) {
+	se.processOperationGroup(t, operationGroup{
 		num:                   num,
 		operations:            se.suite.installations.Base,
 		groupName:             "InstallingBase",
@@ -33,8 +33,8 @@ func (se *suiteExecution) installingBase(num int) {
 	})
 }
 
-func (se *suiteExecution) preUpgradeTests(num int) {
-	se.processOperationGroup(operationGroup{
+func (se *suiteExecution) preUpgradeTests(t *testing.T, num int) {
+	se.processOperationGroup(t, operationGroup{
 		num:                   num,
 		operations:            se.suite.tests.preUpgrade,
 		groupName:             "PreUpgradeTests",
@@ -45,93 +45,43 @@ func (se *suiteExecution) preUpgradeTests(num int) {
 	})
 }
 
-func (se *suiteExecution) startContinualTests(num int) {
+func (se *suiteExecution) runContinualTests(t *testing.T, num int, stopCh <-chan struct{}) {
+	l := se.configuration.logger(t)
 	operations := se.suite.tests.continual
 	groupTemplate := "%d) 🔄 Starting continual tests. " +
 		"%d tests are registered."
 	elementTemplate := `%d.%d) Starting continual tests of "%s".`
 	numOps := len(operations)
-	se.configuration.T.Run("ContinualTests", func(t *testing.T) {
-		l, err := se.configuration.logger(t)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if numOps > 0 {
-			l.Infof(groupTemplate, num, numOps)
-			for i := range operations {
-				operation := operations[i]
-				l.Infof(elementTemplate, num, i+1, operation.Name())
-				if se.failed {
-					l.Debugf(skippingOperationTemplate, operation.Name())
-					return
-				}
+	if numOps > 0 {
+		l.Infof(groupTemplate, num, numOps)
+		for i := range operations {
+			operation := operations[i]
+			l.Debugf(elementTemplate, num, i+1, operation.Name())
+			t.Run(operation.Name(), func(t *testing.T) {
+				l := se.configuration.logger(t)
 				setup := operation.Setup()
-
-				logger, buffer := newInMemoryLoggerBuffer(se.configuration)
-				t.Run("Setup"+operation.Name(), func(t *testing.T) {
-					l, err = se.configuration.logger(t)
-					if err != nil {
-						t.Fatal(err)
-					}
-					setup(Context{T: t, Log: logger.Sugar()})
-				})
-
-				handler := operation.Handler()
-				go func() {
-					handler(BackgroundContext{
-						Log:       logger.Sugar(),
-						Stop:      operation.stop,
-						logBuffer: buffer,
-					})
-				}()
-				se.failed = se.failed || t.Failed()
-				if se.failed {
-					// need to dump logs here, because verify will not be executed.
-					l.Error(wrapLog(buffer.Dump()))
+				setup(Context{T: t, Log: l})
+				if t.Failed() {
 					return
 				}
-			}
-
-		} else {
-			l.Infof("%d) 🔄 No continual tests registered. Skipping.", num)
-		}
-	})
-}
-
-func (se *suiteExecution) verifyContinualTests(num int) {
-	testsCount := len(se.suite.tests.continual)
-	if testsCount > 0 {
-		se.configuration.T.Run("VerifyContinualTests", func(t *testing.T) {
-			l, err := se.configuration.logger(t)
-			if err != nil {
-				t.Fatal(err)
-			}
-			l.Infof("%d) ✋ Verifying %d running continual tests.", num, testsCount)
-			for i, operation := range se.suite.tests.continual {
-				t.Run(operation.Name(), func(t *testing.T) {
-					l, err = se.configuration.logger(t)
-					if err != nil {
-						t.Fatal(err)
-					}
-					l.Infof(`%d.%d) Verifying "%s".`, num, i+1, operation.Name())
-					finished := make(chan struct{})
-					operation.stop <- StopEvent{
-						T:        t,
-						Finished: finished,
-						logger:   l,
-						name:     "Stop of " + operation.Name(),
-					}
-					<-finished
-					se.failed = se.failed || t.Failed()
-					l.Debugf(`Finished "%s"`, operation.Name())
+				t.Parallel()
+				handle := operation.Handler()
+				// Blocking operation.
+				handle(BackgroundContext{
+					T:    t,
+					Log:  l,
+					Stop: stopCh,
 				})
-			}
-		})
+				l.Debugf(`Finished "%s"`, operation.Name())
+			})
+		}
+	} else {
+		l.Infof("%d) 🔄 No continual tests registered. Skipping.", num)
 	}
 }
 
-func (se *suiteExecution) upgradeWith(num int) {
-	se.processOperationGroup(operationGroup{
+func (se *suiteExecution) upgradeWith(t *testing.T, num int) {
+	se.processOperationGroup(t, operationGroup{
 		num:                   num,
 		operations:            se.suite.installations.UpgradeWith,
 		groupName:             "UpgradeWith",
@@ -141,8 +91,8 @@ func (se *suiteExecution) upgradeWith(num int) {
 	})
 }
 
-func (se *suiteExecution) postUpgradeTests(num int) {
-	se.processOperationGroup(operationGroup{
+func (se *suiteExecution) postUpgradeTests(t *testing.T, num int) {
+	se.processOperationGroup(t, operationGroup{
 		num:                   num,
 		operations:            se.suite.tests.postUpgrade,
 		groupName:             "PostUpgradeTests",
@@ -153,8 +103,8 @@ func (se *suiteExecution) postUpgradeTests(num int) {
 	})
 }
 
-func (se *suiteExecution) downgradeWith(num int) {
-	se.processOperationGroup(operationGroup{
+func (se *suiteExecution) downgradeWith(t *testing.T, num int) {
+	se.processOperationGroup(t, operationGroup{
 		num:                   num,
 		operations:            se.suite.installations.DowngradeWith,
 		groupName:             "DowngradeWith",
@@ -164,8 +114,8 @@ func (se *suiteExecution) downgradeWith(num int) {
 	})
 }
 
-func (se *suiteExecution) postDowngradeTests(num int) {
-	se.processOperationGroup(operationGroup{
+func (se *suiteExecution) postDowngradeTests(t *testing.T, num int) {
+	se.processOperationGroup(t, operationGroup{
 		num:                   num,
 		operations:            se.suite.tests.postDowngrade,
 		groupName:             "PostDowngradeTests",
diff --git a/vendor/knative.dev/pkg/test/upgrade/suite_execution.go b/vendor/knative.dev/pkg/test/upgrade/suite_execution.go
index 9f1783e68..cca5d23f7 100644
--- a/vendor/knative.dev/pkg/test/upgrade/suite_execution.go
+++ b/vendor/knative.dev/pkg/test/upgrade/suite_execution.go
@@ -20,30 +20,22 @@ import (
 	"testing"
 )
 
-func (se *suiteExecution) processOperationGroup(op operationGroup) {
-	se.configuration.T.Run(op.groupName, func(t *testing.T) {
-		l, err := se.configuration.logger(t)
-		if err != nil {
-			t.Fatal(err)
-		}
+func (se *suiteExecution) processOperationGroup(t *testing.T, op operationGroup) {
+	t.Run(op.groupName, func(t *testing.T) {
+		l := se.configuration.logger(t)
 		if len(op.operations) > 0 {
 			l.Infof(op.groupTemplate, op.num, len(op.operations))
 			for i, operation := range op.operations {
 				l.Infof(op.elementTemplate, op.num, i+1, operation.Name())
-				if se.failed {
+				if t.Failed() {
 					l.Debugf(skippingOperationTemplate, operation.Name())
 					return
 				}
 				handler := operation.Handler()
 				t.Run(operation.Name(), func(t *testing.T) {
-					l, err = se.configuration.logger(t)
-					if err != nil {
-						t.Fatal(err)
-					}
 					handler(Context{T: t, Log: l})
 				})
-				se.failed = se.failed || t.Failed()
-				if se.failed {
+				if t.Failed() {
 					return
 				}
 			}
@@ -55,38 +47,50 @@ func (se *suiteExecution) processOperationGroup(op operationGroup) {
 
 func (se *suiteExecution) execute() {
 	idx := 1
-	operations := []func(num int){
+	stopCh := make(chan struct{})
+	t := se.configuration.T
+	operations := []func(t *testing.T, num int){
 		se.installingBase,
 		se.preUpgradeTests,
 	}
 	for _, operation := range operations {
-		operation(idx)
+		operation(t, idx)
 		idx++
-		if se.failed {
+		if t.Failed() {
 			return
 		}
 	}
 
-	se.startContinualTests(idx)
-	idx++
-	if se.failed {
-		return
-	}
-	defer func() {
-		se.verifyContinualTests(idx)
-	}()
+	t.Run("Run", func(t *testing.T) {
+		// Calls t.Parallel() after doing setup phase. The second part runs in parallel
+		// with Steps below.
+		se.runContinualTests(t, idx, stopCh)
 
-	operations = []func(num int){
-		se.upgradeWith,
-		se.postUpgradeTests,
-		se.downgradeWith,
-		se.postDowngradeTests,
-	}
-	for _, operation := range operations {
-		operation(idx)
 		idx++
-		if se.failed {
+		// At this point only the setup phase of continual tests was done. We want
+		// to quit early in the event of failures.
+		if t.Failed() {
+			close(stopCh)
 			return
 		}
-	}
+
+		operations = []func(t *testing.T, num int){
+			se.upgradeWith,
+			se.postUpgradeTests,
+			se.downgradeWith,
+			se.postDowngradeTests,
+		}
+		t.Run("Steps", func(t *testing.T) {
+			defer close(stopCh)
+			// The rest of this test group will run in parallel with individual continual tests.
+			t.Parallel()
+			for _, operation := range operations {
+				operation(t, idx)
+				idx++
+				if t.Failed() {
+					return
+				}
+			}
+		})
+	})
 }
diff --git a/vendor/knative.dev/pkg/test/upgrade/types.go b/vendor/knative.dev/pkg/test/upgrade/types.go
index e46a516a4..2c93e922d 100644
--- a/vendor/knative.dev/pkg/test/upgrade/types.go
+++ b/vendor/knative.dev/pkg/test/upgrade/types.go
@@ -81,35 +81,22 @@ type Context struct {
 }
 
 // BackgroundContext is a upgrade test execution context that will be passed
-// down to each handler of BackgroundOperation. It contains a StopEvent channel
-// which end user should use to obtain a testing.T for error reporting. Until
-// StopEvent is sent user may use zap.SugaredLogger to log state of execution if
-// necessary. The logs are stored in a threadSafeBuffer and flushed to the test
-// output when the test fails.
+// down to each handler of BackgroundOperation. It contains a stop event channel.
+// Until stop event is sent user may use zap.SugaredLogger to log state of execution if
+// necessary.
 type BackgroundContext struct {
-	Log       *zap.SugaredLogger
-	Stop      <-chan StopEvent
-	logBuffer *threadSafeBuffer
-}
-
-// StopEvent represents an event that is to be received by background operation
-// to indicate that is should stop it's operations and validate results using
-// passed T. User should use Finished channel to signalize upgrade suite that
-// all stop & verify operations are finished and it is safe to end tests.
-type StopEvent struct {
-	T        *testing.T
-	Finished chan<- struct{}
-	name     string
-	logger   *zap.SugaredLogger
+	T    *testing.T
+	Log  *zap.SugaredLogger
+	Stop <-chan struct{}
 }
 
 // WaitForStopEventConfiguration holds a values to be used be WaitForStopEvent
-// function. OnStop will be called when StopEvent is sent. OnWait will be
+// function. OnStop will be called when a stop event is sent. OnWait will be
 // invoked in a loop while waiting, and each wait act is driven by WaitTime
 // amount.
 type WaitForStopEventConfiguration struct {
 	Name     string
-	OnStop   func(event StopEvent)
+	OnStop   func()
 	OnWait   func(bc BackgroundContext, self WaitForStopEventConfiguration)
 	WaitTime time.Duration
 }
@@ -117,24 +104,14 @@ type WaitForStopEventConfiguration struct {
 // Configuration holds required and optional configuration to run upgrade tests.
 type Configuration struct {
 	T *testing.T
-	// TODO(mgencur): Remove when dependent repositories migrate to LogConfig.
-	// Keep this for backwards compatibility.
-	Log *zap.Logger
 	LogConfig
 }
 
-func (c Configuration) logConfig() LogConfig {
-	if len(c.LogConfig.Config.OutputPaths) == 0 {
-		c.LogConfig.Config = zap.NewDevelopmentConfig()
-	}
-	return c.LogConfig
-}
-
 // LogConfig holds the logger configuration. It allows for passing just the
-// logger configuration and also a custom function for building the resulting
-// logger.
+// logger configuration options.
 type LogConfig struct {
 	// Config from which the zap.Logger be created.
+	// Deprecated: This config doesn't have effect. Use Options instead.
 	Config zap.Config
 	// Options holds options for the zap.Logger.
 	Options []zap.Option
diff --git a/vendor/knative.dev/reconciler-test/pkg/environment/execution.go b/vendor/knative.dev/reconciler-test/pkg/environment/execution.go
index dea615af1..2235a35a1 100644
--- a/vendor/knative.dev/reconciler-test/pkg/environment/execution.go
+++ b/vendor/knative.dev/reconciler-test/pkg/environment/execution.go
@@ -2,21 +2,19 @@ package environment
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	"knative.dev/reconciler-test/pkg/feature"
 )
 
 func categorizeSteps(steps []feature.Step) map[feature.Timing][]feature.Step {
-	res := make(map[feature.Timing][]feature.Step, 4)
+	stepsByTiming := make(map[feature.Timing][]feature.Step, 4)
 
-	res[feature.Setup] = filterStepTimings(steps, feature.Setup)
-	res[feature.Requirement] = filterStepTimings(steps, feature.Requirement)
-	res[feature.Assert] = filterStepTimings(steps, feature.Assert)
-	res[feature.Teardown] = filterStepTimings(steps, feature.Teardown)
+	for _, timing := range feature.Timings() {
+		stepsByTiming[timing] = filterStepTimings(steps, timing)
+	}
 
-	return res
+	return stepsByTiming
 }
 
 func filterStepTimings(steps []feature.Step, timing feature.Timing) []feature.Step {
@@ -29,56 +27,42 @@ func filterStepTimings(steps []feature.Step, timing feature.Timing) []feature.St
 	return res
 }
 
-// executeWithSkippingT executes the step in a sub test wrapping t in order to never fail the subtest
-// This blocks until the test completes.
-func (mr *MagicEnvironment) executeWithSkippingT(ctx context.Context, originalT *testing.T, f *feature.Feature, s *feature.Step) feature.T {
-	originalT.Helper()
-	return mr.executeStep(ctx, originalT, f, s, createSkippingT)
+func (mr *MagicEnvironment) executeOptional(ctx context.Context, t *testing.T, f *feature.Feature, s *feature.Step) {
+	t.Helper()
+	mr.executeStep(ctx, t, f, s, createSkippingT)
 }
 
-// executeWithoutWrappingT executes the step in a sub test without wrapping t.
-// This blocks until the test completes.
-func (mr *MagicEnvironment) executeWithoutWrappingT(ctx context.Context, originalT *testing.T, f *feature.Feature, s *feature.Step) feature.T {
-	originalT.Helper()
-	return mr.executeStep(ctx, originalT, f, s, func(t *testing.T) feature.T {
-		return t
-	})
+// execute executes the step in a sub test without wrapping t.
+func (mr *MagicEnvironment) execute(ctx context.Context, t *testing.T, f *feature.Feature, s *feature.Step) {
+	t.Helper()
+	mr.executeStep(ctx, t, f, s, func(t *testing.T) feature.T { return t })
 }
 
-func (mr *MagicEnvironment) executeStep(ctx context.Context, originalT *testing.T, f *feature.Feature, s *feature.Step, tDecorator func(t *testing.T) feature.T) feature.T {
-	originalT.Helper()
-
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
+func (mr *MagicEnvironment) executeStep(ctx context.Context, t *testing.T, f *feature.Feature, s *feature.Step, tDecorator func(t *testing.T) feature.T) {
+	t.Helper()
 
-	var internalT feature.T
-	originalT.Run(f.Name+"/"+s.T.String()+"/"+s.TestName(), func(st *testing.T) {
-		st.Helper()
-		internalT = tDecorator(st)
+	t.Run(s.Name, func(t *testing.T) {
+		t.Parallel()
+		t.Helper()
+		ft := tDecorator(t)
+		t.Cleanup(func() {
+			mr.milestones.StepFinished(f.Name, s, ft)
+		})
 
 		// Create a cancel tied to this step
 		internalCtx, internalCancelFn := context.WithCancel(ctx)
 
 		defer func() {
 			if r := recover(); r != nil {
-				internalT.Errorf("Panic happened: '%v'", r)
+				ft.Errorf("Panic happened: '%v'", r)
 			}
 			// Close the context as soon as possible (this defer is invoked before the Cleanup)
 			internalCancelFn()
 		}()
 
-		mr.milestones.StepStarted(f.Name, s, internalT)
-		internalT.Cleanup(func() {
-			mr.milestones.StepFinished(f.Name, s, internalT)
-			wg.Done() // Make sure wg.Done() is invoked at the very end of the execution
-		})
+		mr.milestones.StepStarted(f.Name, s, ft)
 
 		// Perform step.
-		s.Fn(internalCtx, internalT)
+		s.Fn(internalCtx, ft)
 	})
-
-	// Wait for the test to execute before spawning the next one
-	wg.Wait()
-
-	return internalT
 }
diff --git a/vendor/knative.dev/reconciler-test/pkg/environment/logging.go b/vendor/knative.dev/reconciler-test/pkg/environment/logging.go
index bf0f35e22..a85e492be 100644
--- a/vendor/knative.dev/reconciler-test/pkg/environment/logging.go
+++ b/vendor/knative.dev/reconciler-test/pkg/environment/logging.go
@@ -20,12 +20,16 @@ import (
 	"context"
 
 	"go.uber.org/zap/zaptest"
+
 	"knative.dev/reconciler-test/pkg/feature"
 	testlog "knative.dev/reconciler-test/pkg/logging"
 )
 
 // loggingSteps returns a number of steps that logs environment-managed resources.
 func (mr *MagicEnvironment) loggingSteps() []feature.Step {
+	mr.refsMu.Lock()
+	defer mr.refsMu.Unlock()
+
 	return []feature.Step{{
 		Name: "Log references",
 		S:    feature.Any,
diff --git a/vendor/knative.dev/reconciler-test/pkg/environment/magic.go b/vendor/knative.dev/reconciler-test/pkg/environment/magic.go
index a469c9c04..69526dd6e 100644
--- a/vendor/knative.dev/reconciler-test/pkg/environment/magic.go
+++ b/vendor/knative.dev/reconciler-test/pkg/environment/magic.go
@@ -28,6 +28,7 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"knative.dev/pkg/logging"
+
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/milestone"
 	"knative.dev/reconciler-test/pkg/state"
@@ -69,6 +70,7 @@ type MagicEnvironment struct {
 	namespace        string
 	namespaceCreated bool
 	refs             []corev1.ObjectReference
+	refsMu           sync.Mutex
 
 	// milestones sends milestone events, if configured.
 	milestones milestone.Emitter
@@ -86,11 +88,19 @@ const (
 )
 
 func (mr *MagicEnvironment) Reference(ref ...corev1.ObjectReference) {
+	mr.refsMu.Lock()
+	defer mr.refsMu.Unlock()
+
 	mr.refs = append(mr.refs, ref...)
 }
 
 func (mr *MagicEnvironment) References() []corev1.ObjectReference {
-	return mr.refs
+	mr.refsMu.Lock()
+	defer mr.refsMu.Unlock()
+
+	r := make([]corev1.ObjectReference, len(mr.refs))
+	copy(r, mr.refs)
+	return r
 }
 
 func (mr *MagicEnvironment) Finish() {
@@ -215,7 +225,9 @@ func (mr *MagicEnvironment) TemplateConfig(base map[string]interface{}) map[stri
 	for k, v := range base {
 		cfg[k] = v
 	}
-	cfg["namespace"] = mr.namespace
+	if _, ok := cfg["namespace"]; !ok {
+		cfg["namespace"] = mr.namespace
+	}
 	return cfg
 }
 
@@ -295,74 +307,58 @@ func (mr *MagicEnvironment) Test(ctx context.Context, originalT *testing.T, f *f
 	ctx = state.ContextWith(ctx, f.State)
 	ctx = feature.ContextWith(ctx, f)
 
-	steps := categorizeSteps(f.Steps)
+	stepsByTiming := categorizeSteps(f.Steps)
 
-	skipAssertions := false
-	skipRequirements := false
-	skipReason := ""
+	mr.milestones.StepsPlanned(f.Name, stepsByTiming, originalT)
 
-	mr.milestones.StepsPlanned(f.Name, steps, originalT)
+	// skip is flag that signals whether the steps for the subsequent timings should
+	// be skipped because a step in a previous timing failed.
+	//
+	// Setup and Teardown steps are executed always
+	skip := false
 
-	for _, s := range steps[feature.Setup] {
-		s := s
+	originalT.Run(f.Name, func(t *testing.T) {
 
-		// Setup are executed always, no matter their level and state
-		internalT := mr.executeWithoutWrappingT(ctx, originalT, f, &s)
+		for _, timing := range feature.Timings() {
+			steps := feature.Steps(stepsByTiming[timing])
 
-		// Failed setup fails everything, so just run the teardown
-		if internalT.Failed() {
-			skipAssertions = true
-			skipRequirements = true // No need to test other requirements
-			break                   // No need to continue the setup
-		}
-	}
+			// Special case for teardown timing
+			if timing == feature.Teardown {
+				// Prepend logging steps to the teardown phase when a previous timing failed.
+				if skip {
+					steps = append(mr.loggingSteps(), steps...)
+				}
 
-	for _, s := range steps[feature.Requirement] {
-		s := s
+				// Teardown steps are executed always, no matter their level and state
+				skip = false
+			}
 
-		if skipRequirements {
-			break
-		}
+			originalT.Logf("Running %d steps for timing:\n%s\n\n", len(steps), steps.String())
 
-		internalT := mr.executeWithoutWrappingT(ctx, originalT, f, &s)
+			t.Run(timing.String(), func(t *testing.T) {
+				// no parallel, various timing steps run in order: setup, requirement, assert, teardown
 
-		if internalT.Failed() {
-			skipAssertions = true
-			skipRequirements = true // No need to test other requirements
-		}
-	}
+				if skip {
+					t.Skipf("Skipping steps for timing %s due to failed previous timing\n", timing.String())
+					return
+				}
 
-	for _, s := range steps[feature.Assert] {
-		s := s
-
-		if skipAssertions {
-			break
-		}
+				for _, s := range steps {
+					s := s
+					if mr.shouldFail(&s) {
+						mr.execute(ctx, t, f, &s)
+					} else {
+						mr.executeOptional(ctx, t, f, &s)
+					}
+				}
+			})
 
-		if mr.shouldFail(&s) {
-			mr.executeWithoutWrappingT(ctx, originalT, f, &s)
-		} else {
-			mr.executeWithSkippingT(ctx, originalT, f, &s)
+			if t.Failed() {
+				// skip the following timings since curring timing failed
+				skip = true
+			}
 		}
-
-		// TODO implement fail fast feature to avoid proceeding with testing if an "expected level" assert fails here
-	}
-
-	if originalT.Failed() {
-		// Prepend logging steps to the teardown phase.
-		steps[feature.Teardown] = append(mr.loggingSteps(), steps[feature.Teardown]...)
-	}
-
-	for _, s := range steps[feature.Teardown] {
-		s := s
-
-		// Teardown are executed always, no matter their level and state
-		mr.executeWithoutWrappingT(ctx, originalT, f, &s)
-	}
-
-	if skipReason != "" {
-		originalT.Skipf("Skipping feature '%s' assertions because %s", f.Name, skipReason)
-	}
+	})
 }
 
 type unknownResult struct{}
@@ -373,6 +369,7 @@ func (u unknownResult) Failed() bool {
 
 // TODO: this logic is strange and hard to follow.
 func (mr *MagicEnvironment) shouldFail(s *feature.Step) bool {
+	// if it's _not_ alpha nor must
 	return !(mr.s&s.S == 0 || mr.l&s.L == 0)
 }
 
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
index 8c1e49733..1495f62a4 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"sync"
 	"time"
 
 	conformanceevent "github.com/cloudevents/conformance/pkg/event"
@@ -33,6 +34,7 @@ import (
 
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/resources/svc"
 )
 
@@ -43,11 +45,20 @@ func NewProber() *EventProber {
 }
 
 type EventProber struct {
-	target          target
-	shortNameToName map[string]string
-	ids             []string
+	target   target
+	targetMu sync.Mutex
+
+	shortNameToName   map[string]string
+	shortNameToNameMu sync.Mutex
+
+	ids   []string
+	idsMu sync.Mutex
+
 	senderOptions   []EventsHubOption
-	receiverOptions []EventsHubOption
+	senderOptionsMu sync.Mutex
+
+	receiverOptions   []EventsHubOption
+	receiverOptionsMu sync.Mutex
 }
 
 type target struct {
@@ -64,6 +75,9 @@ type EventInfoCombined struct {
 
 // SetTargetResource configures the senders target as a GVR and name, used when sender is installed.
 func (p *EventProber) SetTargetResource(targetGVR schema.GroupVersionResource, targetName string) {
+	p.targetMu.Lock()
+	defer p.targetMu.Unlock()
+
 	p.target = target{
 		gvr:  targetGVR,
 		name: targetName,
@@ -83,15 +97,15 @@ func (p *EventProber) SetTargetKRef(ref *duckv1.KReference) error {
 		Kind:    ref.Kind,
 	}
 	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
-	p.target = target{
-		gvr:  gvr,
-		name: ref.Name,
-	}
+	p.SetTargetResource(gvr, ref.Name)
 	return nil
 }
 
 // SetTargetURI configures the senders target as a URI, used when sender is installed.
 func (p *EventProber) SetTargetURI(targetURI string) {
+	p.targetMu.Lock()
+	defer p.targetMu.Unlock()
+
 	p.target = target{
 		uri: targetURI,
 	}
@@ -99,44 +113,49 @@ func (p *EventProber) SetTargetURI(targetURI string) {
 
 // ReceiversRejectFirstN adds DropFirstN to the default config for new receivers.
 func (p *EventProber) ReceiversRejectFirstN(n uint) {
-	p.receiverOptions = append(p.receiverOptions, DropFirstN(n))
+	p.appendReceiverOptions(DropFirstN(n))
 }
 
 // ReceiversRejectResponseCode adds DropEventsResponseCode to the default config for new receivers.
 func (p *EventProber) ReceiversRejectResponseCode(code int) {
-	p.receiverOptions = append(p.receiverOptions, DropEventsResponseCode(code))
+	p.appendReceiverOptions(DropEventsResponseCode(code))
 }
 
 // ReceiversHaveResponseDelay adds ResponseWaitTime to the default config for
 // new receivers.
 func (p *EventProber) ReceiversHaveResponseDelay(delay time.Duration) {
-	p.receiverOptions = append(p.receiverOptions, ResponseWaitTime(delay))
+	p.appendReceiverOptions(ResponseWaitTime(delay))
 }
 
 // ReceiverInstall installs an eventshub receiver into the test env.
 func (p *EventProber) ReceiverInstall(prefix string, opts ...EventsHubOption) feature.StepFn {
 	name := feature.MakeRandomK8sName(prefix)
-	p.shortNameToName[prefix] = name
-	opts = append(p.receiverOptions, opts...)
+	p.setShortNameToName(prefix, name)
+	opts = append(p.getReceiverOptions(), opts...)
 	opts = append(opts, StartReceiver)
+
 	return Install(name, opts...)
 }
 
 // SenderInstall installs an eventshub sender resource into the test env.
 func (p *EventProber) SenderInstall(prefix string, opts ...EventsHubOption) feature.StepFn {
 	name := feature.MakeRandomK8sName(prefix)
-	p.shortNameToName[prefix] = name
+	p.setShortNameToName(prefix, name)
+
 	return func(ctx context.Context, t feature.T) {
-		opts := append(opts, p.senderOptions...)
-		if len(p.target.uri) > 0 {
-			opts = append(opts, StartSenderURL(p.target.uri))
-		} else if !p.target.gvr.Empty() {
-			opts = append(opts, StartSenderToResource(p.target.gvr, p.target.name))
+		opts := append(opts, p.getSenderOptions()...)
+		if len(p.getTarget().uri) > 0 {
+			opts = append(opts, StartSenderURL(p.getTarget().uri))
+		} else if !p.getTarget().gvr.Empty() {
+			k8s.IsAddressable(p.getTarget().gvr, p.getTarget().name)(ctx, t)
+			opts = append(opts, StartSenderToResource(p.getTarget().gvr, p.getTarget().name))
 		} else {
 			t.Fatal("no target is configured for event loop")
 		}
 		// Install into the env.
 		Install(name, opts...)(ctx, t)
+
+		p.SenderDone(prefix)(ctx, t)
 	}
 }
 
@@ -146,9 +165,10 @@ func (p *EventProber) SenderDone(prefix string) feature.StepFn {
 		interval, timeout := environment.PollTimingsFromContext(ctx)
 		var events []EventInfoCombined
 		err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+			expected := p.getIdsCopy()
 			events = p.SentBy(ctx, prefix)
-			t.Log(p.shortNameToName[prefix], "has sent", len(events))
-			if len(events) == len(p.ids) {
+			t.Log(p.getNameFromPrefix(prefix), "has sent", len(events), "expected", len(expected))
+			if len(events) == len(expected) {
 				return true, nil
 			}
 			return false, nil
@@ -174,13 +194,13 @@ func (p *EventProber) ReceiverDone(from, to string) feature.StepFn {
 		)
 		err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 			sent = p.SentBy(ctx, from)
-			t.Log(p.shortNameToName[from], "has sent", len(sent))
+			t.Log(p.getNameFromPrefix(from), "has sent", len(sent))
 
 			received = p.ReceivedBy(ctx, to)
-			t.Log(p.shortNameToName[to], "has received", len(received))
+			t.Log(p.getNameFromPrefix(to), "has received", len(received))
 
 			rejected = p.RejectedBy(ctx, to)
-			t.Log(p.shortNameToName[to], "has rejected", len(rejected))
+			t.Log(p.getNameFromPrefix(to), "has rejected", len(rejected))
 
 			if len(sent) == len(received)+len(rejected) {
 				return true, nil
@@ -220,7 +240,7 @@ func CorrelateSent(origin string, in []EventInfo) []EventInfoCombined {
 
 // SentBy returns events sent by the named sender.
 func (p *EventProber) SentBy(ctx context.Context, prefix string) []EventInfoCombined {
-	name := p.shortNameToName[prefix]
+	name := p.getNameFromPrefix(prefix)
 	store := StoreFromContext(ctx, name)
 
 	return CorrelateSent(name, store.Collected())
@@ -228,7 +248,7 @@ func (p *EventProber) SentBy(ctx context.Context, prefix string) []EventInfoComb
 
 // ReceivedBy returns events received by the named receiver.
 func (p *EventProber) ReceivedBy(ctx context.Context, prefix string) []EventInfo {
-	name := p.shortNameToName[prefix]
+	name := p.getNameFromPrefix(prefix)
 	store := StoreFromContext(ctx, name)
 
 	events, _, _, _ := store.Find(func(info EventInfo) error {
@@ -258,7 +278,7 @@ func (p *EventProber) RejectedBy(ctx context.Context, prefix string) []EventInfo
 
 // ReceivedOrRejectedBy returns events received or rejected by the named receiver.
 func (p *EventProber) ReceivedOrRejectedBy(ctx context.Context, prefix string) []EventInfo {
-	name := p.shortNameToName[prefix]
+	name := p.getNameFromPrefix(prefix)
 	store := StoreFromContext(ctx, name)
 
 	events, _, _, _ := store.Find(func(info EventInfo) error {
@@ -281,19 +301,19 @@ func (p *EventProber) ExpectYAMLEvents(path string) error {
 		return fmt.Errorf("failed to load events from %q", path)
 	}
 	for _, event := range events {
-		p.ids = append(p.ids, event.Attributes.ID)
+		p.appendIds(event.Attributes.ID)
 	}
 	return nil
 }
 
 // ExpectEvents registers event IDs into the prober.
 func (p *EventProber) ExpectEvents(ids []string) {
-	p.ids = append(p.ids, ids...)
+	p.appendIds(ids...)
 }
 
 // SenderEventsFromURI configures a sender to send a url/yaml based events.
 func (p *EventProber) SenderEventsFromURI(uri string) {
-	p.senderOptions = append(p.senderOptions, InputYAML(uri))
+	p.appendSenderOptions(InputYAML(uri))
 }
 
 // SenderEventsFromSVC configures a sender to send a yaml based events fetched
@@ -306,7 +326,7 @@ func (p *EventProber) SenderEventsFromSVC(svcName, path string) feature.StepFn {
 			t.Error(err)
 		}
 		u.Path = path
-		p.senderOptions = append(p.senderOptions, InputYAML(u.String()))
+		p.appendSenderOptions(InputYAML(u.String()))
 	}
 }
 
@@ -318,14 +338,16 @@ func (p *EventProber) SenderFullEvents(count int) {
 	if count == 1 {
 		id := uuid.New().String()
 		event.SetID(id)
-		p.ids = []string{id}
-		p.senderOptions = append(p.senderOptions, InputEvent(event))
+		p.appendIds(id)
+		p.appendSenderOptions(InputEvent(event))
 	} else {
-		p.senderOptions = append(p.senderOptions,
-			InputEvent(event), SendMultipleEvents(count, 10*time.Millisecond), EnableIncrementalId) // TODO: make configurable.
-		p.ids = []string{}
+		p.appendSenderOptions(
+			InputEvent(event),
+			SendMultipleEvents(count, 10*time.Millisecond),
+			EnableIncrementalId,
+		)
 		for i := 1; i <= count; i++ {
-			p.ids = append(p.ids, strconv.Itoa(i))
+			p.appendIds(strconv.Itoa(i))
 		}
 	}
 }
@@ -338,31 +360,33 @@ func (p *EventProber) SenderMinEvents(count int) {
 	if count == 1 {
 		id := uuid.New().String()
 		event.SetID(id)
-		p.ids = []string{id}
-		p.senderOptions = append(p.senderOptions, InputEvent(event))
+		p.appendIds(id)
+		p.appendSenderOptions(InputEvent(event))
 	} else {
-		p.senderOptions = append(p.senderOptions, InputEvent(event), SendMultipleEvents(count, 10*time.Millisecond)) // TODO: make configurable.
-		p.ids = []string{}
+		p.appendSenderOptions(InputEvent(event), SendMultipleEvents(count, 10*time.Millisecond))
 		for i := 1; i <= count; i++ {
-			p.ids = append(p.ids, strconv.Itoa(i))
+			p.appendIds(strconv.Itoa(i))
 		}
 	}
 }
 
 // AsKReference returns the short-named component as a KReference.
 func (p *EventProber) AsKReference(prefix string) *duckv1.KReference {
-	return svc.AsKReference(p.shortNameToName[prefix])
+	return svc.AsKReference(p.getNameFromPrefix(prefix))
 }
 
 // AssertSentAll tests that `fromPrefix` sent all known events known to the prober.
 func (p *EventProber) AssertSentAll(fromPrefix string) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
+		p.SenderDone(fromPrefix)(ctx, t)
+
 		events := p.SentBy(ctx, fromPrefix)
-		if len(p.ids) != len(events) {
+		expected := p.getIdsCopy()
+		if len(expected) != len(events) {
 			t.Errorf("expected %q to have sent %d events, actually sent %d",
-				fromPrefix, len(p.ids), len(events))
+				fromPrefix, len(expected), len(events))
 		}
-		for _, id := range p.ids {
+		for _, id := range expected {
 			found := false
 			for _, event := range events {
 				if id == event.Sent.SentId {
@@ -383,6 +407,8 @@ func (p *EventProber) AssertSentAll(fromPrefix string) feature.StepFn {
 // AssertReceivedAll tests that all events sent by `fromPrefix` were received by `toPrefix`.
 func (p *EventProber) AssertReceivedAll(fromPrefix, toPrefix string) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
+		p.ReceiverDone(fromPrefix, toPrefix)(ctx, t)
+
 		sent := p.SentBy(ctx, fromPrefix)
 		ids := make([]string, len(sent))
 		for i, s := range sent {
@@ -412,6 +438,8 @@ func (p *EventProber) AssertReceivedAll(fromPrefix, toPrefix string) feature.Ste
 // AssertReceivedAll tests that all events sent by `fromPrefix` were received by `toPrefix`.
 func (p *EventProber) AssertReceivedOrRejectedAll(fromPrefix, toPrefix string) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
+		p.ReceiverDone(fromPrefix, toPrefix)(ctx, t)
+
 		sent := p.SentBy(ctx, fromPrefix)
 		ids := make([]string, len(sent))
 		for i, s := range sent {
@@ -458,7 +486,6 @@ func combine(ei []EventInfo) []EventInfoCombined {
 // AssertReceivedNone tests that no events sent by `fromPrefix` were received by `toPrefix`.
 func (p *EventProber) AssertReceivedNone(fromPrefix, toPrefix string) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
-
 		events := p.ReceivedBy(ctx, toPrefix)
 		if len(events) > 0 {
 			t.Errorf("expected %q to not have received any events from %s, actually received %d",
@@ -466,3 +493,77 @@ func (p *EventProber) AssertReceivedNone(fromPrefix, toPrefix string) feature.St
 		}
 	}
 }
+
+func (p *EventProber) setShortNameToName(k, v string) {
+	p.shortNameToNameMu.Lock()
+	defer p.shortNameToNameMu.Unlock()
+
+	p.shortNameToName[k] = v
+}
+
+func (p *EventProber) getNameFromPrefix(prefix string) string {
+	p.shortNameToNameMu.Lock()
+	defer p.shortNameToNameMu.Unlock()
+
+	return p.shortNameToName[prefix]
+}
+
+func (p *EventProber) getSenderOptions() []EventsHubOption {
+	p.senderOptionsMu.Lock()
+	defer p.senderOptionsMu.Unlock()
+
+	return p.senderOptions
+}
+
+func (p *EventProber) getReceiverOptions() []EventsHubOption {
+	p.receiverOptionsMu.Lock()
+	defer p.receiverOptionsMu.Unlock()
+
+	return p.receiverOptions
+}
+
+func (p *EventProber) appendReceiverOptions(opt ...EventsHubOption) {
+	p.receiverOptionsMu.Lock()
+	defer p.receiverOptionsMu.Unlock()
+
+	p.receiverOptions = append(p.receiverOptions, opt...)
+}
+
+func (p *EventProber) appendSenderOptions(opt ...EventsHubOption) {
+	p.senderOptionsMu.Lock()
+	defer p.senderOptionsMu.Unlock()
+
+	p.senderOptions = append(p.senderOptions, opt...)
+}
+
+func (p *EventProber) appendIds(ids ...string) {
+	p.idsMu.Lock()
+	defer p.idsMu.Unlock()
+
+	p.ids = append(p.ids, ids...)
+}
+
+func (p *EventProber) getIdsCopy() []string {
+	p.idsMu.Lock()
+	defer p.idsMu.Unlock()
+
+	ids := make([]string, len(p.ids))
+	copy(ids, p.ids)
+	return ids
+}
+
+func (p *EventProber) getTarget() target {
+	p.targetMu.Lock()
+	defer p.targetMu.Unlock()
+
+	return p.target
+}
+
+func (p *EventProber) DumpState(ctx context.Context, t feature.T) {
+	p.shortNameToNameMu.Lock()
+	defer p.shortNameToNameMu.Unlock()
+
+	for k, v := range p.shortNameToName {
+		t.Logf("collected for %s (%s)\n", v, k, StoreFromContext(ctx, v).dumpCollected())
+	}
+}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go
index a684478cf..099b73a83 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/rbac/rbac.go
@@ -20,6 +20,8 @@ import (
 	"context"
 	"embed"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 )
@@ -31,7 +33,7 @@ var templates embed.FS
 // The resources are named according to the current namespace defined in the environment.
 func Install() feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
-		if _, err := manifest.InstallYamlFS(ctx, templates, map[string]interface{}{}); err != nil {
+		if _, err := manifest.InstallYamlFS(ctx, templates, map[string]interface{}{}); err != nil && !apierrors.IsAlreadyExists(err) {
 			t.Fatal(err)
 		}
 	}
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/sender/sender.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/sender/sender.go
index e350a5a4b..f423612b4 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/sender/sender.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/sender/sender.go
@@ -36,8 +36,10 @@ import (
 	"github.com/cloudevents/sdk-go/v2/types"
 	"github.com/kelseyhightower/envconfig"
 	"go.opencensus.io/trace"
+	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"knative.dev/pkg/logging"
+
 	"knative.dev/reconciler-test/pkg/eventshub"
 )
 
@@ -144,6 +146,7 @@ func Start(ctx context.Context, logs *eventshub.EventLogs, clientOpts ...Option)
 			}
 
 			if _, err := nethttp.DefaultClient.Do(req); err != nil {
+				logging.FromContext(ctx).Error(zap.Error(err))
 				return false, nil
 			}
 			return true, nil
diff --git a/vendor/knative.dev/reconciler-test/pkg/feature/feature.go b/vendor/knative.dev/reconciler-test/pkg/feature/feature.go
index 422bd95bb..6c073392d 100644
--- a/vendor/knative.dev/reconciler-test/pkg/feature/feature.go
+++ b/vendor/knative.dev/reconciler-test/pkg/feature/feature.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"runtime"
 	"strings"
+	"sync"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -31,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/injection/clients/dynamicclient"
+
 	"knative.dev/reconciler-test/pkg/state"
 )
 
@@ -40,14 +42,18 @@ type Feature struct {
 	Steps []Step
 	State state.Store
 	// Contains all the resources created as part of this Feature.
-	refs []corev1.ObjectReference
+	refs   []corev1.ObjectReference
+	refsMu sync.Mutex
 }
 
-func (f Feature) MarshalJSON() ([]byte, error) {
+func (f *Feature) MarshalJSON() ([]byte, error) {
 	return f.marshalJSON(true)
 }
 
-func (f Feature) marshalJSON(pretty bool) ([]byte, error) {
+func (f *Feature) marshalJSON(pretty bool) ([]byte, error) {
+	f.refsMu.Lock()
+	defer f.refsMu.Unlock()
+
 	in := struct {
 		Name  string                   `json:"name"`
 		Steps []Step                   `json:"steps"`
@@ -70,7 +76,7 @@ func (f Feature) marshalJSON(pretty bool) ([]byte, error) {
 
 // DumpWith calls the provided log function with a nicely formatted string
 // that represents the Feature.
-func (f Feature) DumpWith(log func(args ...interface{})) {
+func (f *Feature) DumpWith(log func(args ...interface{})) {
 	b, err := f.marshalJSON(false)
 	if err != nil {
 		log("Skipping feature logging due to error: ", err.Error())
@@ -118,6 +124,13 @@ type Step struct {
 	Fn   StepFn `json:"-"`
 }
 
+type Steps []Step
+
+func (ss Steps) String() string {
+	bytes, _ := json.MarshalIndent(ss, "", "  ")
+	return string(bytes)
+}
+
 // TestName returns the constructed test name based on the timing, step, state,
 // level, and the Name provided in the step.
 func (s *Step) TestName() string {
@@ -132,13 +145,21 @@ func (s *Step) TestName() string {
 // Reference adds references to keep track of for example, for cleaning things
 // after a Feature completes.
 func (f *Feature) Reference(ref ...corev1.ObjectReference) {
+	f.refsMu.Lock()
+	defer f.refsMu.Unlock()
+
 	f.refs = append(f.refs, ref...)
 }
 
 // References returns all known resources to the Feature registered via
 // `Reference`.
 func (f *Feature) References() []corev1.ObjectReference {
-	return f.refs
+	f.refsMu.Lock()
+	defer f.refsMu.Unlock()
+
+	r := make([]corev1.ObjectReference, len(f.refs))
+	copy(r, f.refs)
+	return r
 }
 
 // DeleteResources delete all known resources to the Feature registered
@@ -208,6 +229,9 @@ func (f *Feature) DeleteResources(ctx context.Context, t T) {
 		t.Fatalf("failed to wait for resources to be deleted: %v", err)
 	}
 
+	f.refsMu.Lock()
+	defer f.refsMu.Unlock()
+
 	f.refs = refFailedDeletion
 }
 
diff --git a/vendor/knative.dev/reconciler-test/pkg/k8s/job.go b/vendor/knative.dev/reconciler-test/pkg/k8s/job.go
index 9bf436086..8a635fdd3 100644
--- a/vendor/knative.dev/reconciler-test/pkg/k8s/job.go
+++ b/vendor/knative.dev/reconciler-test/pkg/k8s/job.go
@@ -38,6 +38,22 @@ import (
 // WaitUntilJobDone waits until a job has finished.
 // Timing is optional but if provided is [interval, timeout].
 func WaitUntilJobDone(ctx context.Context, t feature.T, name string, timing ...time.Duration) error {
+	return WaitForJobCondition(ctx, t, name, IsJobComplete, timing...)
+}
+
+// WaitUntilJobSucceeded waits until a job has succeeded.
+// Timing is optional but if provided is [interval, timeout].
+func WaitUntilJobSucceeded(ctx context.Context, t feature.T, name string, timing ...time.Duration) error {
+	return WaitForJobCondition(ctx, t, name, IsJobSucceeded, timing...)
+}
+
+// WaitUntilJobFailed waits until a job has failed.
+// Timing is optional but if provided is [interval, timeout].
+func WaitUntilJobFailed(ctx context.Context, t feature.T, name string, timing ...time.Duration) error {
+	return WaitForJobCondition(ctx, t, name, IsJobFailed, timing...)
+}
+
+func WaitForJobCondition(ctx context.Context, t feature.T, name string, isConditionFunc func(job *batchv1.Job) bool, timing ...time.Duration) error {
 	interval, timeout := PollTimings(ctx, timing)
 	namespace := environment.FromContext(ctx).Namespace()
 	kube := kubeclient.Get(ctx)
@@ -53,15 +69,16 @@ func WaitUntilJobDone(ctx context.Context, t feature.T, name string, timing ...t
 			}
 			return false, err
 		}
-		complete := IsJobComplete(job)
-		if !complete {
+
+		conditionIsTrue := isConditionFunc(job)
+		if !conditionIsTrue {
 			status, err := json.Marshal(job.Status)
 			if err != nil {
 				return false, err
 			}
 			t.Logf("%s/%s job status %s", namespace, name, status)
 		}
-		return complete, nil
+		return conditionIsTrue, nil
 	})
 	if err != nil {
 		return err
diff --git a/vendor/knative.dev/reconciler-test/pkg/k8s/wait.go b/vendor/knative.dev/reconciler-test/pkg/k8s/wait.go
index 5a1179505..94249a68b 100644
--- a/vendor/knative.dev/reconciler-test/pkg/k8s/wait.go
+++ b/vendor/knative.dev/reconciler-test/pkg/k8s/wait.go
@@ -37,6 +37,7 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/injection/clients/dynamicclient"
 	"knative.dev/pkg/kmeta"
+
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 )
@@ -251,6 +252,11 @@ func WaitForServiceEndpoints(ctx context.Context, t feature.T, name string, numb
 	if err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		svc, err := services.Get(ctx, name, metav1.GetOptions{})
 		if err != nil {
+			if apierrors.IsNotFound(err) {
+				t.Log("service", "namespace", ns, "name", name, err)
+				// keep polling
+				return false, nil
+			}
 			return false, err
 		}
 		ip := svc.Spec.ClusterIP
@@ -265,6 +271,11 @@ func WaitForServiceEndpoints(ctx context.Context, t feature.T, name string, numb
 	if err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		endpoint, err := endpoints.Get(ctx, name, metav1.GetOptions{})
 		if err != nil {
+			if apierrors.IsNotFound(err) {
+				t.Log("endpoint", "namespace", ns, "name", name, err)
+				// keep polling
+				return false, nil
+			}
 			return false, err
 		}
 		num := countEndpointsNum(endpoint)
@@ -374,6 +385,11 @@ func WaitForPodRunningOrFail(ctx context.Context, t feature.T, podName string) {
 	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		p, err := p.Get(ctx, podName, metav1.GetOptions{})
 		if err != nil {
+			if apierrors.IsNotFound(err) {
+				t.Log("pod", "namespace", ns, "name", podName, err)
+				// keep polling
+				return false, nil
+			}
 			return true, err
 		}
 		isRunning := podRunning(p)
diff --git a/vendor/knative.dev/reconciler-test/pkg/resources/job/job.go b/vendor/knative.dev/reconciler-test/pkg/resources/job/job.go
new file mode 100644
index 000000000..c5a60e820
--- /dev/null
+++ b/vendor/knative.dev/reconciler-test/pkg/resources/job/job.go
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2023 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 job
+
+import (
+	"context"
+	"embed"
+	"time"
+
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+//go:embed *.yaml
+var yaml embed.FS
+
+func Install(name string, image string, options ...manifest.CfgFn) feature.StepFn {
+	cfg := map[string]interface{}{
+		"name":  name,
+		"image": image,
+	}
+
+	for _, fn := range options {
+		fn(cfg)
+	}
+
+	return func(ctx context.Context, t feature.T) {
+		if err := registerImage(ctx, image); err != nil {
+			t.Fatal(err)
+		}
+		manifest.PodSecurityCfgFn(ctx, t)(cfg)
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func IsDone(name string, timing ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		if err := k8s.WaitUntilJobDone(ctx, t, name, timing...); err != nil {
+			t.Error("Job did not turn into done state", err)
+		}
+	}
+}
+
+func IsSucceeded(name string, timing ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		if err := k8s.WaitUntilJobSucceeded(ctx, t, name, timing...); err != nil {
+			t.Error("Job did not succeed", err)
+		}
+	}
+}
+
+func IsFailed(name string, timing ...time.Duration) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		if err := k8s.WaitUntilJobFailed(ctx, t, name, timing...); err != nil {
+			t.Error("Job did not fail", err)
+		}
+	}
+}
+
+func registerImage(ctx context.Context, image string) error {
+	reg := environment.RegisterPackage(image)
+	_, err := reg(ctx, environment.FromContext(ctx))
+	return err
+}
diff --git a/vendor/knative.dev/reconciler-test/pkg/resources/job/job.yaml b/vendor/knative.dev/reconciler-test/pkg/resources/job/job.yaml
new file mode 100644
index 000000000..2269966a9
--- /dev/null
+++ b/vendor/knative.dev/reconciler-test/pkg/resources/job/job.yaml
@@ -0,0 +1,80 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+  {{ if .annotations }}
+  annotations:
+    {{ range $key, $value := .annotations }}
+    {{ $key }}: "{{ $value }}"
+    {{ end }}
+  {{ end }}
+  {{ if .labels }}
+  labels:
+    {{ range $key, $value := .labels }}
+    {{ $key }}: {{ $value }}
+    {{ end }}
+  {{ end }}
+spec:
+  {{ if .backoffLimit }}
+  backoffLimit: {{ .backoffLimit }}
+  {{ end }}
+  {{ if .ttlSecondsAfterFinished }}
+  ttlSecondsAfterFinished: {{ .ttlSecondsAfterFinished }}
+  {{ end }}
+  template:
+    {{ if or .podannotations .podlabels }}
+    metadata:
+      {{ if .podannotations }}
+      annotations:
+        {{ range $key, $value := .podannotations }}
+        {{ $key }}: "{{ $value }}"
+        {{ end }}
+      {{ end }}
+      {{ if .podlabels }}
+      labels:
+        {{ range $key, $value := .podlabels }}
+        {{ $key }}: {{ $value }}
+        {{ end }}
+      {{ end }}
+    {{ end }}
+    spec:
+      {{ if .podSecurityContext }}
+      securityContext:
+        runAsNonRoot: {{ .podSecurityContext.runAsNonRoot }}
+        seccompProfile:
+          type: {{ .podSecurityContext.seccompProfile.type }}
+      {{ end }}
+      {{ if .restartPolicy }}
+      restartPolicy: {{ .restartPolicy }}
+      {{ end }}
+      containers:
+      - name: job-container
+        image: {{ .image }}
+        {{ if .envs }}
+        env:
+        {{ range $key, $value := .envs }}
+        - name: {{ printf "%q" $key }}
+          value: {{ printf "%q" $value }}
+        {{ end }}
+        {{ end }}
+        {{ if .containerSecurityContext }}
+        securityContext:
+          capabilities:
+            {{ if .containerSecurityContext.capabilities.drop }}
+            drop:
+            {{ range $_, $value := .containerSecurityContext.capabilities.drop }}
+            - {{ $value }}
+            {{ end }}
+            {{ end }}
+            {{ if .containerSecurityContext.capabilities.add }}
+            add:
+            {{ range $_, $value := .containerSecurityContext.capabilities.add }}
+            - {{ $value }}
+            {{ end }}
+            {{ end }}
+          allowPrivilegeEscalation: {{ .containerSecurityContext.allowPrivilegeEscalation }}
+        {{ end }}
+        {{ if .imagePullPolicy }}
+        imagePullPolicy: {{ .imagePullPolicy }}
+        {{ end }}
diff --git a/vendor/knative.dev/reconciler-test/pkg/resources/job/options.go b/vendor/knative.dev/reconciler-test/pkg/resources/job/options.go
new file mode 100644
index 000000000..74be32489
--- /dev/null
+++ b/vendor/knative.dev/reconciler-test/pkg/resources/job/options.go
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2023 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 job
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+func WithEnvs(envs map[string]string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if envs != nil {
+			cfg["envs"] = envs
+		}
+	}
+}
+
+func WithAnnotations(annotations map[string]interface{}) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if annotations != nil {
+			cfg["annotations"] = annotations
+		}
+	}
+}
+
+func WithLabels(labels map[string]string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if labels != nil {
+			cfg["labels"] = labels
+		}
+	}
+}
+
+func WithPodAnnotations(annotations map[string]interface{}) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if annotations != nil {
+			cfg["podannotations"] = annotations
+		}
+	}
+}
+
+func WithPodLabels(labels map[string]string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if labels != nil {
+			cfg["podlabels"] = labels
+		}
+	}
+}
+
+func WithImagePullPolicy(ipp corev1.PullPolicy) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["imagePullPolicy"] = ipp
+	}
+}
+
+func WithRestartPolicy(restartPolicy corev1.RestartPolicy) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["restartPolicy"] = restartPolicy
+	}
+}
+
+func WithBackoffLimit(backoffLimit int) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["backoffLimit"] = backoffLimit
+	}
+}
+
+func WithTTLSecondsAfterFinished(ttlSecondsAfterFinished int) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["ttlSecondsAfterFinished"] = ttlSecondsAfterFinished
+	}
+}