Repository: knative-sandbox/eventing-kafka
Pull request #574 (2021-04-27 02:41:22)

Copy OWNERS_ALIASES from knative/community; fill in OWNERS files
Convert CODEOWNERS to OWNERS files per https://github.com/knative/test-infra/issues/2751.

Do not delete CODEOWNERS until the migration is complete and Prow is switched back.


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

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

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/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"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#574](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a6741bd) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/e427c58ebba9f61db5c698f9b7079b22a4d28fde?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (e427c58) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@           Coverage Diff           @@
##             main     #574   +/-   ##
=======================================
  Coverage   71.68%   71.68%           
=======================================
  Files         137      137           
  Lines        5644     5644           
=======================================
  Hits         4046     4046           
  Misses       1371     1371           
  Partials      227      227           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [e427c58...a6741bd](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/574?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @evankanderson: PR needs rebase.

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1395511598558220288) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1395511598579191808) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1395511598558220288) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1395511598579191808) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-go-coverage/1395511598734381056) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1395511598558220288) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1395511598579191808) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-go-coverage/1395511598734381056) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-upgrade-tests/1395511598658883584) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1395511598558220288) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1395511598579191808) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-go-coverage/1395511598734381056) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-upgrade-tests/1395511598658883584) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1395511598616940544) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1395511598558220288) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1395511598579191808) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-go-coverage/1395511598734381056) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-upgrade-tests/1395511598658883584) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1395511598616940544) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-unit-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-unit-tests/1395511598705020928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1395511598537248768) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395511598637912064) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1395511598558220288) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1395511598579191808) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-go-coverage/1395511598734381056) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-upgrade-tests/1395511598658883584) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1395511598616940544) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-unit-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-unit-tests/1395511598705020928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | 61c06f25c4cdbc0e88b34282edfce0d6203580ad | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/574/pull-knative-sandbox-eventing-kafka-build-tests/1395511598679855104) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<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 -->
evankanderson: /retest

Patch:
diff --git a/OWNERS b/OWNERS
index 78fee8de6..70fed6301 100644
--- a/OWNERS
+++ b/OWNERS
@@ -1,26 +1,11 @@
 # Initial set of approvers taken from eventing WG leads, Kafka approvers and original developers of the channel added
 approvers:
-- vaikas
-- lionelvillard
-- grantr
-- matzew
-- lberk
-- slinkydeveloper
-- davyodom
-- travis-minke-sap
-- phamilton
-- aliok
+- technical-oversight-committee
+- knative-release-leads
+- eventing-writers
+- eventing-kafka-approvers
 
 reviewers:
-- vaikas
-- lionelvillard
-- grantr
-- matzew
-- lberk
-- slinkydeveloper
-- steven0711dong
-- davyodom
-- travis-minke-sap
-- phamilton
-- devguyio
-- aliok
+- eventing-writers
+- eventing-kafka-approvers
+
diff --git a/cmd/source/mtcontroller/OWNERS b/cmd/source/mtcontroller/OWNERS
new file mode 100644
index 000000000..671657b79
--- /dev/null
+++ b/cmd/source/mtcontroller/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- eventing-kafka-mtsource-approvers
+
diff --git a/cmd/source/mtreceive_adapter/OWNERS b/cmd/source/mtreceive_adapter/OWNERS
new file mode 100644
index 000000000..671657b79
--- /dev/null
+++ b/cmd/source/mtreceive_adapter/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- eventing-kafka-mtsource-approvers
+
diff --git a/config/post-install/clusterrole.yaml b/config/post-install/clusterrole.yaml
deleted file mode 100644
index a2a722c29..000000000
--- a/config/post-install/clusterrole.yaml
+++ /dev/null
@@ -1,65 +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: rbac.authorization.k8s.io/v1
-kind: ClusterRole
-metadata:
-  name: knative-kafka-eventing-post-install-job-role
-  labels:
-    eventing.knative.dev/release: devel
-rules:
-  # Storage version upgrader needs to be able to patch CRDs.
-  - apiGroups:
-      - "apiextensions.k8s.io"
-    resources:
-      - "customresourcedefinitions"
-      - "customresourcedefinitions/status"
-    verbs:
-      - "get"
-      - "list"
-      - "update"
-      - "patch"
-      - "watch"
-  # Our own resources we care about.
-  - apiGroups:
-      - "sources.knative.dev"
-    resources:
-      - "kafkasources"
-    verbs:
-      - "get"
-      - "list"
-      - "create"
-      - "update"
-      - "delete"
-      - "patch"
-      - "watch"
-  - apiGroups:
-      - "bindings.knative.dev"
-    resources:
-      - "kafkabindings"
-    verbs:
-      - "get"
-      - "list"
-      - "create"
-      - "update"
-      - "delete"
-      - "patch"
-      - "watch"
-  - apiGroups:
-      - ""
-    resources:
-      - "namespaces"
-    verbs:
-      - "get"
-      - "list"
diff --git a/config/post-install/serviceaccount.yaml b/config/post-install/serviceaccount.yaml
deleted file mode 100644
index 55723644e..000000000
--- a/config/post-install/serviceaccount.yaml
+++ /dev/null
@@ -1,38 +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: ServiceAccount
-metadata:
-  name: knative-kafka-eventing-post-install-job
-  namespace: knative-eventing
-  labels:
-    kafka.eventing.knative.dev/release: devel
-
----
-
-apiVersion: rbac.authorization.k8s.io/v1
-kind: ClusterRoleBinding
-metadata:
-  name: knative-kafka-eventing-post-install-job-role-binding
-  labels:
-    kafka.eventing.knative.dev/release: devel
-subjects:
-  - kind: ServiceAccount
-    name: knative-kafka-eventing-post-install-job
-    namespace: knative-eventing
-roleRef:
-  kind: ClusterRole
-  name: knative-kafka-eventing-post-install-job-role
-  apiGroup: rbac.authorization.k8s.io
diff --git a/config/post-install/storage-version-migrator.yaml b/config/post-install/storage-version-migrator.yaml
deleted file mode 100644
index ffaf2c177..000000000
--- a/config/post-install/storage-version-migrator.yaml
+++ /dev/null
@@ -1,46 +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
-#
-#     https://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: batch/v1
-kind: Job
-metadata:
-  # Adding a prefix to avoid naming conflict with previous version's post-install jobs,
-  # we cannot use `generateName` here as it's not supported by `kubectl apply -f`
-  #
-  # If `ttlSecondsAfterFinished` feature gate becomes generally available in the future,
-  # we can rely on that and keep using the same Job name.
-  name: v0.23-kafka-storage-version-migration
-  namespace: knative-eventing
-  labels:
-    app: "storage-version-migration"
-    kafka.eventing.knative.dev/release: devel
-spec:
-  ttlSecondsAfterFinished: 600
-  backoffLimit: 10
-  template:
-    metadata:
-      labels:
-        app: "storage-version-migration"
-        eventing.knative.dev/release: devel
-      annotations:
-        sidecar.istio.io/inject: "false"
-    spec:
-      serviceAccountName: knative-kafka-eventing-post-install-job
-      restartPolicy: OnFailure
-      containers:
-        - name: migrate
-          image: ko://knative.dev/eventing-kafka/vendor/knative.dev/pkg/apiextensions/storageversion/cmd/migrate
-          args:
-            - "kafkasources.sources.knative.dev"
-            - "kafkabindings.bindings.knative.dev"
diff --git a/config/source/multi/OWNERS b/config/source/multi/OWNERS
new file mode 100644
index 000000000..671657b79
--- /dev/null
+++ b/config/source/multi/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- eventing-kafka-mtsource-approvers
+
diff --git a/go.mod b/go.mod
index 4bd13cb25..17e67359b 100644
--- a/go.mod
+++ b/go.mod
@@ -33,10 +33,10 @@ require (
 	k8s.io/apimachinery v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
-	knative.dev/control-protocol v0.0.0-20210517064824-b879a6d906be
-	knative.dev/eventing v0.23.1-0.20210520114617-5af518d57612
+	knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78
+	knative.dev/eventing v0.23.1-0.20210524085620-e48c80d7f969
 	knative.dev/hack v0.0.0-20210428122153-93ad9129c268
 	knative.dev/networking v0.0.0-20210520062317-b2552eca09f7
 	knative.dev/pkg v0.0.0-20210520062216-e749d6a2ad0e
-	knative.dev/reconciler-test v0.0.0-20210514214942-8d70771450f1
+	knative.dev/reconciler-test v0.0.0-20210520170117-ffe49e0e092f
 )
diff --git a/go.sum b/go.sum
index 8b6af5161..757eba694 100644
--- a/go.sum
+++ b/go.sum
@@ -1309,10 +1309,10 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6 h1:+WnxoVtG8TMiudHBSEtrVL
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20210517064824-b879a6d906be h1:q4pexlMC0YaB4EhDtgGLT2LowJaNCtOycUAyv7FeyRI=
-knative.dev/control-protocol v0.0.0-20210517064824-b879a6d906be/go.mod h1:z0X6o/iPLE0cwK+YVDZSdPH/VtrJqbqSlZXBWiWEv+I=
-knative.dev/eventing v0.23.1-0.20210520114617-5af518d57612 h1:FOXGMay+XyWFmzMWlB6u+5kMeONOMOB4Mjl7yblVtJA=
-knative.dev/eventing v0.23.1-0.20210520114617-5af518d57612/go.mod h1:4WewoHOfKYH1nGMrFrZzNz4ZXnCy8WThWley770Zyjc=
+knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78 h1:OzXUGWQFLxO+O00rIUmB5Ir7ZiHg5AVlBxgnVdYzves=
+knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78/go.mod h1:z0X6o/iPLE0cwK+YVDZSdPH/VtrJqbqSlZXBWiWEv+I=
+knative.dev/eventing v0.23.1-0.20210524085620-e48c80d7f969 h1:o1EN/8udcyE1zh1ni1GZsyhFHJfY2sOLgh/wmMSLv8k=
+knative.dev/eventing v0.23.1-0.20210524085620-e48c80d7f969/go.mod h1:4WewoHOfKYH1nGMrFrZzNz4ZXnCy8WThWley770Zyjc=
 knative.dev/hack v0.0.0-20210428122153-93ad9129c268 h1:lBIj9Epd9UQ55NEaHzAdY/UZbuaegCdGPKVC2+Z68Q0=
 knative.dev/hack v0.0.0-20210428122153-93ad9129c268/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210428122153-93ad9129c268/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
@@ -1323,8 +1323,9 @@ knative.dev/pkg v0.0.0-20210518131015-67897f4ec290/go.mod h1:fIl4l4OmZodkElyaHoT
 knative.dev/pkg v0.0.0-20210520062216-e749d6a2ad0e h1:BTHdk8J4PQ1UQIZlpBQaN4sw+vYqap4P97OWP/qytbo=
 knative.dev/pkg v0.0.0-20210520062216-e749d6a2ad0e/go.mod h1:fIl4l4OmZodkElyaHoT0LCF5wT+3+P/kinawQ4XlLtE=
 knative.dev/reconciler-test v0.0.0-20210512062647-d5adf72546be/go.mod h1:ZHvzjwe5aUnWDx+u5y9EPaZ++5PM9Yapp/IvBDiS+Yk=
-knative.dev/reconciler-test v0.0.0-20210514214942-8d70771450f1 h1:Xf/JUzEjkkmW5ppczck49Dpi828AuIdqzZFO5DwqdHs=
 knative.dev/reconciler-test v0.0.0-20210514214942-8d70771450f1/go.mod h1:ZHvzjwe5aUnWDx+u5y9EPaZ++5PM9Yapp/IvBDiS+Yk=
+knative.dev/reconciler-test v0.0.0-20210520170117-ffe49e0e092f h1:SR7fMoSw9/bxznFti0VzgkMSZM+6jqziSjXkC2MbLT0=
+knative.dev/reconciler-test v0.0.0-20210520170117-ffe49e0e092f/go.mod h1:ZHvzjwe5aUnWDx+u5y9EPaZ++5PM9Yapp/IvBDiS+Yk=
 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/OWNERS b/hack/OWNERS
new file mode 100644
index 000000000..225c5fa28
--- /dev/null
+++ b/hack/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- productivity-wg-leads
+
diff --git a/pkg/common/consumer/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
index 06b792d46..1ff21beba 100644
--- a/pkg/common/consumer/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -39,6 +39,7 @@ type customConsumerGroup struct {
 	cancel              func()
 	handlerErrorChannel chan error
 	sarama.ConsumerGroup
+	releasedCh chan bool
 }
 
 // Merge handler errors chan and consumer group error chan
@@ -48,6 +49,10 @@ func (c *customConsumerGroup) Errors() <-chan error {
 
 func (c *customConsumerGroup) Close() error {
 	c.cancel()
+
+	// Wait for graceful session claims release
+	<-c.releasedCh
+
 	return c.ConsumerGroup.Close()
 }
 
@@ -60,10 +65,14 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
 	}
 
 	errorCh := make(chan error, 10)
+	releasedCh := make(chan bool)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	go func() {
-		defer close(errorCh)
+		defer func() {
+			close(errorCh)
+			releasedCh <- true
+		}()
 		for {
 			consumerHandler := NewConsumerHandler(logger, handler, errorCh, options...)
 
@@ -83,7 +92,7 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
 		}
 	}()
 
-	return &customConsumerGroup{cancel, errorCh, consumerGroup}, err
+	return &customConsumerGroup{cancel, errorCh, consumerGroup, releasedCh}, err
 }
 
 func NewConsumerGroupFactory(addrs []string, config *sarama.Config) KafkaConsumerGroupFactory {
diff --git a/pkg/common/scheduler/OWNERS b/pkg/common/scheduler/OWNERS
new file mode 100644
index 000000000..671657b79
--- /dev/null
+++ b/pkg/common/scheduler/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- eventing-kafka-mtsource-approvers
+
diff --git a/pkg/common/scheduler/statefulset/scheduler.go b/pkg/common/scheduler/statefulset/scheduler.go
index 4ff729986..6e1126b3f 100644
--- a/pkg/common/scheduler/statefulset/scheduler.go
+++ b/pkg/common/scheduler/statefulset/scheduler.go
@@ -21,6 +21,7 @@ import (
 	"errors"
 	"math"
 	"sort"
+	"strconv"
 	"sync"
 	"time"
 
@@ -122,6 +123,19 @@ func NewStatefulSetScheduler(ctx context.Context,
 }
 
 func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
+	placements, err := s.scheduleVPod(vpod)
+	if placements == nil {
+		return placements, err
+	}
+	sort.SliceStable(placements, func(i int, j int) bool {
+		podIOrdinal, _ := strconv.Atoi(placements[i].PodName[len(placements[i].PodName)-1:])
+		podJOrdinal, _ := strconv.Atoi(placements[j].PodName[len(placements[j].PodName)-1:])
+		return podIOrdinal < podJOrdinal
+	})
+	return placements, err
+}
+
+func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -206,7 +220,7 @@ func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Pla
 
 func (s *StatefulSetScheduler) removeReplicas(diff int32, placements []duckv1alpha1.Placement) []duckv1alpha1.Placement {
 	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
-	for i := 0; i < len(placements); i++ {
+	for i := len(placements) - 1; i > -1; i-- {
 		if diff >= placements[i].VReplicas {
 			// remove the entire placement
 			diff -= placements[i].VReplicas
diff --git a/pkg/source/OWNERS b/pkg/source/OWNERS
new file mode 100644
index 000000000..671657b79
--- /dev/null
+++ b/pkg/source/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- eventing-kafka-mtsource-approvers
+
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 67e39c4ed..d67d1c87d 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -55,6 +55,11 @@ func NewEnvConfig() adapter.EnvConfigAccessor {
 	return new(AdapterConfig)
 }
 
+type cancelContext struct {
+	fn      context.CancelFunc
+	stopped chan bool
+}
+
 type Adapter struct {
 	config      *AdapterConfig
 	logger      *zap.SugaredLogger
@@ -64,7 +69,7 @@ type Adapter struct {
 	memLimit    int64
 
 	sourcesMu sync.RWMutex
-	sources   map[string]context.CancelFunc
+	sources   map[string]cancelContext
 }
 
 var _ adapter.Adapter = (*Adapter)(nil)
@@ -86,7 +91,7 @@ func newAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClie
 		kubeClient:  kubeclient.Get(ctx),
 		memLimit:    ml.Value(),
 		sourcesMu:   sync.RWMutex{},
-		sources:     make(map[string]context.CancelFunc),
+		sources:     make(map[string]cancelContext),
 	}
 }
 
@@ -162,7 +167,13 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 	if ok {
 		// TODO: do not stop if the only thing that changes is the number of vreplicas
 		logger.Info("stopping adapter")
-		cancel()
+		cancel.fn()
+
+		// Wait for the adapter to stop
+		<-cancel.stopped
+
+		// Nothing to stop anymore
+		delete(a.sources, key)
 	}
 
 	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
@@ -274,14 +285,21 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 	}
 
 	ctx, cancelFn := context.WithCancel(ctx)
+
+	cancel = cancelContext{
+		fn:      cancelFn,
+		stopped: make(chan bool),
+	}
+
 	go func(ctx context.Context) {
 		err := adapter.Start(ctx)
 		if err != nil {
 			a.logger.Errorw("adapter failed to start", zap.Error(err))
 		}
+		cancel.stopped <- true
 	}(ctx)
 
-	a.sources[key] = cancelFn
+	a.sources[key] = cancel
 	a.logger.Infow("source added", "name", obj.Name)
 	return nil
 }
@@ -296,14 +314,15 @@ func (a *Adapter) Remove(obj *v1beta1.KafkaSource) {
 	cancel, ok := a.sources[key]
 
 	if !ok {
-		a.logger.Infow("source not found", "name", obj.Name)
+		a.logger.Infow("source was not running. removed.", "name", obj.Name)
 		return
 	}
 
-	cancel()
+	cancel.fn()
+	<-cancel.stopped
 
 	delete(a.sources, key)
-	a.logger.Infow("source removed", "name", obj.Name, "count", len(a.sources))
+	a.logger.Infow("source removed", "name", obj.Name, "remaining", len(a.sources))
 }
 
 // ResolveSecret resolves the secret reference
diff --git a/vendor/knative.dev/eventing/pkg/adapter/v2/cloudevents.go b/vendor/knative.dev/eventing/pkg/adapter/v2/cloudevents.go
index e9febad24..e1c07471f 100644
--- a/vendor/knative.dev/eventing/pkg/adapter/v2/cloudevents.go
+++ b/vendor/knative.dev/eventing/pkg/adapter/v2/cloudevents.go
@@ -181,8 +181,9 @@ func (c *client) reportMetrics(ctx context.Context, event cloudevents.Event, res
 			var res *http.Result
 			if !cloudevents.ResultAs(retryResult, &res) {
 				c.reportError(reportArgs, result)
+			} else {
+				c.reporter.ReportRetryEventCount(reportArgs, res.StatusCode)
 			}
-			c.reporter.ReportRetryEventCount(reportArgs, res.StatusCode)
 		}
 	}
 }
diff --git a/config/post-install/placeholder.go b/vendor/knative.dev/eventing/test/upgrade/prober/errors.go
similarity index 82%
rename from config/post-install/placeholder.go
rename to vendor/knative.dev/eventing/test/upgrade/prober/errors.go
index d5a6b2f6c..e70da17bd 100644
--- a/config/post-install/placeholder.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/errors.go
@@ -1,9 +1,12 @@
 /*
 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.
@@ -11,6 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package pre_install is a placeholder that allows us to pull in config files
-// via go mod vendor.
-package post_install
+package prober
+
+func (p *prober) ensureNoError(err error) {
+	if err != nil {
+		p.client.T.Fatal(err)
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 7dd2e349e..b31a41ced 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1097,7 +1097,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/control-protocol v0.0.0-20210517064824-b879a6d906be
+# knative.dev/control-protocol v0.0.0-20210520180517-3214e528bf78
 ## explicit
 knative.dev/control-protocol/pkg
 knative.dev/control-protocol/pkg/certificates
@@ -1105,7 +1105,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.23.1-0.20210520114617-5af518d57612
+# knative.dev/eventing v0.23.1-0.20210524085620-e48c80d7f969
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1211,6 +1211,7 @@ knative.dev/eventing/test/test_images/wathola-forwarder
 knative.dev/eventing/test/test_images/wathola-receiver
 knative.dev/eventing/test/test_images/wathola-sender
 knative.dev/eventing/test/upgrade/prober
+knative.dev/eventing/test/upgrade/prober/sut
 knative.dev/eventing/test/upgrade/prober/wathola/client
 knative.dev/eventing/test/upgrade/prober/wathola/config
 knative.dev/eventing/test/upgrade/prober/wathola/event
@@ -1329,7 +1330,7 @@ knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/conversion
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20210514214942-8d70771450f1
+# knative.dev/reconciler-test v0.0.0-20210520170117-ffe49e0e092f
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/pkg/common/scheduler/statefulset/scheduler_test.go b/pkg/common/scheduler/statefulset/scheduler_test.go
index 0d68a1afd..3df6250db 100644
--- a/pkg/common/scheduler/statefulset/scheduler_test.go
+++ b/pkg/common/scheduler/statefulset/scheduler_test.go
@@ -135,8 +135,8 @@ func TestStatefulsetScheduler(t *testing.T) {
 				{PodName: "statefulset-name-1", VReplicas: 10},
 			},
 			expected: []duckv1alpha1.Placement{
-				{PodName: "statefulset-name-0", VReplicas: 5},
-				{PodName: "statefulset-name-1", VReplicas: 10},
+				{PodName: "statefulset-name-0", VReplicas: 10},
+				{PodName: "statefulset-name-1", VReplicas: 5},
 			},
 		},
 		{
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index 676950a2d..f29e6025f 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -40,7 +40,7 @@ import (
 
 var (
 	runningAdapterChan  = make(chan *sampleAdapter)
-	stoppingAdapterChan = make(chan *sampleAdapter)
+	stoppingAdapterChan = make(chan *sampleAdapter, 1)
 )
 
 const (
@@ -54,18 +54,18 @@ func TestUpdateRemoveSources(t *testing.T) {
 	env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
 	ceClient := adaptertest.NewTestClient()
 
-	adapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
+	mtadapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
 
-	adapterStopped := make(chan bool)
+	mtadapterStopped := make(chan bool)
 	go func() {
-		err := adapter.Start(ctx)
+		err := mtadapter.Start(ctx)
 		if err != nil {
 			t.Error("Unexpected error ", err)
 		}
-		adapterStopped <- true
+		mtadapterStopped <- true
 	}()
 
-	err := adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
+	err := mtadapter.Update(ctx, &sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name",
 			Namespace: "test-ns",
@@ -83,11 +83,20 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Fatalf("Unexpected error %v", err)
 	}
 
-	if _, ok := adapter.sources["test-ns/test-name"]; !ok {
+	select {
+	case a := <-runningAdapterChan:
+		if !a.running {
+			t.Error("Expected adapter to be running")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("sub-adapter failed to start after 100 ms")
+	}
+
+	if _, ok := mtadapter.sources["test-ns/test-name"]; !ok {
 		t.Error(`Expected adapter to contain "test-ns/test-name"`)
 	}
 
-	err = adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
+	err = mtadapter.Update(ctx, &sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name-badBroker",
 			Namespace: "test-ns",
@@ -109,7 +118,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Fatalf("Unexpected error %v", err)
 	}
 
-	if _, ok := adapter.sources["test-ns/test-name-badBroker"]; !ok {
+	if _, ok := mtadapter.sources["test-ns/test-name-badBroker"]; !ok {
 		t.Error(`Expected adapter to contain "test-ns/test-name-badBroker"`)
 	}
 
@@ -122,7 +131,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Error("sub-adapter failed to start after 100 ms")
 	}
 
-	adapter.Remove(&sourcesv1beta1.KafkaSource{
+	mtadapter.Remove(&sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name",
 			Namespace: "test-ns",
@@ -131,11 +140,20 @@ func TestUpdateRemoveSources(t *testing.T) {
 		Status: sourcesv1beta1.KafkaSourceStatus{},
 	})
 
-	if _, ok := adapter.sources["test-ns/test-name"]; ok {
+	select {
+	case a := <-stoppingAdapterChan:
+		if a.running {
+			t.Error("Expected adapter to not be running")
+		}
+	case <-time.After(100 * time.Millisecond):
+		t.Error("sub-adapter failed to stop after 100 ms")
+	}
+
+	if _, ok := mtadapter.sources["test-ns/test-name"]; ok {
 		t.Error(`Expected adapter to not contain "test-ns/test-name"`)
 	}
 
-	adapter.Remove(&sourcesv1beta1.KafkaSource{
+	mtadapter.Remove(&sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name-badBroker",
 			Namespace: "test-ns",
@@ -144,10 +162,6 @@ func TestUpdateRemoveSources(t *testing.T) {
 		Status: sourcesv1beta1.KafkaSourceStatus{},
 	})
 
-	if _, ok := adapter.sources["test-ns/test-name-badBroker"]; ok {
-		t.Error(`Expected adapter to not contain "test-ns/test-name-badBroker"`)
-	}
-
 	select {
 	case a := <-stoppingAdapterChan:
 		if a.running {
@@ -157,6 +171,10 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Error("sub-adapter failed to stop after 100 ms")
 	}
 
+	if _, ok := mtadapter.sources["test-ns/test-name-badBroker"]; ok {
+		t.Error(`Expected adapter to not contain "test-ns/test-name-badBroker"`)
+	}
+
 	// Make sure the adapter is still running
 	select {
 	case <-ctx.Done():
@@ -167,7 +185,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 	cancelAdapter()
 
 	select {
-	case <-adapterStopped:
+	case <-mtadapterStopped:
 	case <-time.After(2 * time.Second):
 		t.Error("adapter failed to stop after 2 seconds")
 	}
@@ -384,5 +402,6 @@ func (d *sampleAdapter) Start(ctx context.Context) error {
 
 	d.running = false
 	stoppingAdapterChan <- d
+
 	return nil
 }
diff --git a/test/OWNERS b/test/OWNERS
new file mode 100644
index 000000000..225c5fa28
--- /dev/null
+++ b/test/OWNERS
@@ -0,0 +1,3 @@
+approvers:
+- productivity-wg-leads
+
diff --git a/test/test_images/wathola-kafka-sender/main.go b/test/test_images/wathola-kafka-sender/main.go
new file mode 100644
index 000000000..9f645098c
--- /dev/null
+++ b/test/test_images/wathola-kafka-sender/main.go
@@ -0,0 +1,35 @@
+/*
+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 (
+	"go.uber.org/zap"
+	"knative.dev/eventing-kafka/test/upgrade/continual"
+	"knative.dev/eventing/test/upgrade/prober/wathola/sender"
+	"knative.dev/pkg/signals"
+)
+
+func main() {
+	ctx := signals.NewContext()
+	log, err := zap.NewDevelopment()
+	if err != nil {
+		panic(err)
+	}
+	es := continual.CreateKafkaSender(ctx, log.Sugar())
+	sender.RegisterEventSender(es)
+	sender.New().SendContinually()
+}
diff --git a/test/test_images/wathola-kafka-sender/pod.yaml b/test/test_images/wathola-kafka-sender/pod.yaml
new file mode 100644
index 000000000..5125c4c61
--- /dev/null
+++ b/test/test_images/wathola-kafka-sender/pod.yaml
@@ -0,0 +1,10 @@
+# This file is required to trigger ko resolve in upload-test-images.sh
+
+apiVersion: v1
+kind: Pod
+metadata:
+  name: wathola-kafka-sender
+spec:
+  containers:
+  - name: wathola-kafka-sender
+    image: ko://knative.dev/eventing-kafka/test/test_images/wathola-kafka-sender
diff --git a/test/upgrade/config.toml b/test/upgrade/config.toml
deleted file mode 100644
index 7b6514a2e..000000000
--- a/test/upgrade/config.toml
+++ /dev/null
@@ -1,11 +0,0 @@
-# logLevel = 5 # DEBUG(5)
-[sender]
-address = '{{- .BrokerURL -}}'
-interval = {{ .Config.Interval.Nanoseconds }}
-
-[forwarder]
-target = 'http://wathola-receiver.{{- .Config.Namespace -}}.svc.cluster.local'
-
-[receiver]
-  [receiver.errors]
-  unavailablePeriodToReport = 35_000_000_000 # 35sec in nsec.
diff --git a/test/upgrade/continual.go b/test/upgrade/continual.go
index 1f0616874..a76f5b8b4 100644
--- a/test/upgrade/continual.go
+++ b/test/upgrade/continual.go
@@ -17,181 +17,25 @@ limitations under the License.
 package upgrade
 
 import (
-	"context"
-	"fmt"
-	"time"
-
-	"github.com/kelseyhightower/envconfig"
-	"github.com/stretchr/testify/assert"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/resources"
-	"knative.dev/eventing/test/upgrade/prober"
+	"knative.dev/eventing-kafka/test/upgrade/continual"
 	pkgupgrade "knative.dev/pkg/test/upgrade"
 )
 
-var (
-	defaultCustomizeConfig = func(config *prober.Config) {
-		config.FailOnErrors = true
-		config.Interval = 2 * time.Millisecond
-		config.BrokerOpts = append(config.BrokerOpts,
-			resources.WithDeliveryForBroker(&eventingduckv1.DeliverySpec{
-				Retry:         &retryCount,
-				BackoffPolicy: &backoffPolicy,
-				BackoffDelay:  &backoffDelay,
-			}))
-	}
-	retryCount    = int32(12)
-	backoffPolicy = eventingduckv1.BackoffPolicyExponential
-	backoffDelay  = "PT1S"
-)
-
-// ChannelContinualTest tests channel operation in continual manner during the
-// whole upgrade and downgrade process asserting that all event are propagated
-// well.
-func ChannelContinualTest(opts ContinualTestOptions) pkgupgrade.BackgroundOperation {
-	opts = fillInDefaults(opts)
-	ctx := context.Background()
-	var client *testlib.Client
-	var probe prober.Prober
-	setup := func(c pkgupgrade.Context) {
-		// setup
-		client = testlib.Setup(c.T, false)
-		opts.SetKafkaChannelAsDefault(KafkaChannelAsDefaultOptions{
-			UpgradeCtx:         c,
-			Ctx:                ctx,
-			Client:             client,
-			ReplicationOptions: *opts.ReplicationOptions,
-			RetryOptions:       *opts.RetryOptions,
-		})
-		config := prober.NewConfig(client.Namespace)
-		// TODO: knative/eventing#5176 - this is cumbersome
-		config.ConfigTemplate = "../../../../../../test/upgrade/config.toml"
-		opts.ConfigCustomize(config)
-		// envconfig.Process invocation is repeated from within prober.NewConfig to
-		// make sure every knob is configurable, but using defaults from Eventing
-		// Kafka instead of Core. The prefix is also changed.
-		err := envconfig.Process("eventing_kafka_source_upgrade_tests", config)
-		assert.NoError(c.T, err)
-
-		probe = prober.RunEventProber(ctx, c.Log, client, config)
-	}
-	verify := func(c pkgupgrade.Context) {
-		// verify
-		if client != nil {
-			defer testlib.TearDown(client)
-		}
-		if probe != nil {
-			prober.AssertEventProber(ctx, c.T, probe)
-		}
+// ChannelContinualTests returns background operations to test channel
+// functionality in continual manner during the whole upgrade and downgrade
+// process asserting that all events are propagated well.
+func ChannelContinualTests(opts continual.ChannelTestOptions) []pkgupgrade.BackgroundOperation {
+	return []pkgupgrade.BackgroundOperation{
+		continual.ChannelTest(opts),
+		continual.BrokerBackedByChannelTest(opts),
 	}
-	return pkgupgrade.NewBackgroundVerification(
-		"ChannelContinualTest", setup, verify)
 }
 
-// SourceContinualTest tests source operation in continual manner during the
-// whole upgrade and downgrade process asserting that all event are propagated
+// SourceContinualTests tests source operation in continual manner during the
+// whole upgrade and downgrade process asserting that all events are propagated
 // well.
-func SourceContinualTest(_ ContinualTestOptions) pkgupgrade.BackgroundOperation {
-	setup := func(c pkgupgrade.Context) {
-		// TODO: not yet implemented
-		c.Log.Warn("TODO: not yet implemented")
-	}
-	verify := func(c pkgupgrade.Context) {
-		// TODO: not yet implemented
-		c.Log.Warn("TODO: not yet implemented")
-	}
-	return pkgupgrade.NewBackgroundVerification(
-		"SourceContinualTest", setup, verify)
-}
-
-// ReplicationOptions hold options for replication.
-type ReplicationOptions struct {
-	NumPartitions     int
-	ReplicationFactor int
-}
-
-// RetryOptions holds options for retries.
-type RetryOptions struct {
-	RetryCount    int
-	BackoffPolicy eventingduckv1.BackoffPolicyType
-	BackoffDelay  string
-}
-
-// ContinualTestOptions holds options for EventingKafka continual tests.
-type ContinualTestOptions struct {
-	ConfigCustomize          func(config *prober.Config)
-	SetKafkaChannelAsDefault func(ctx KafkaChannelAsDefaultOptions)
-	*ReplicationOptions
-	*RetryOptions
-}
-
-// KafkaChannelAsDefaultOptions holds a options to run SetKafkaChannelAsDefault func.
-type KafkaChannelAsDefaultOptions struct {
-	UpgradeCtx pkgupgrade.Context
-	Ctx        context.Context
-	*testlib.Client
-	ReplicationOptions
-	RetryOptions
-}
-
-func fillInDefaults(opts ContinualTestOptions) ContinualTestOptions {
-	o := opts
-	if opts.ConfigCustomize == nil {
-		o.ConfigCustomize = defaultCustomizeConfig
-	}
-	if opts.SetKafkaChannelAsDefault == nil {
-		o.SetKafkaChannelAsDefault = configureKafkaChannelAsDefault
-	}
-	if opts.RetryOptions == nil {
-		o.RetryOptions = &RetryOptions{
-			RetryCount:    int(retryCount),
-			BackoffPolicy: backoffPolicy,
-			BackoffDelay:  backoffDelay,
-		}
+func SourceContinualTests(opts continual.SourceTestOptions) []pkgupgrade.BackgroundOperation {
+	return []pkgupgrade.BackgroundOperation{
+		continual.SourceTest(opts),
 	}
-	if opts.ReplicationOptions == nil {
-		o.ReplicationOptions = &ReplicationOptions{
-			NumPartitions:     6,
-			ReplicationFactor: 3,
-		}
-	}
-	return o
-}
-
-func configureKafkaChannelAsDefault(opts KafkaChannelAsDefaultOptions) {
-	c := opts.UpgradeCtx
-	systemNs := "knative-eventing"
-	configmaps := opts.Client.Kube.CoreV1().ConfigMaps(systemNs)
-	cm := &corev1.ConfigMap{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: systemNs,
-			Name:      "config-br-default-channel",
-		},
-		Data: map[string]string{
-			"channelTemplateSpec": fmt.Sprintf(`apiVersion: %s
-kind: %s
-spec:
-  numPartitions: %d
-  replicationFactor: %d
-  delivery:
-    retry: %d
-    backoffPolicy: %s
-    backoffDelay: %s`,
-				defaultChannelType.APIVersion, defaultChannelType.Kind,
-				opts.NumPartitions, opts.ReplicationFactor,
-				opts.RetryCount, opts.BackoffPolicy, opts.BackoffDelay,
-			),
-		},
-	}
-	cm, err := configmaps.Update(opts.Ctx, cm, metav1.UpdateOptions{})
-
-	if !assert.NoError(c.T, err) {
-		c.T.FailNow()
-	}
-
-	c.Log.Info("Updated config-br-default-channel in ns knative-eventing"+
-		" to eq: ", cm.Data)
 }
diff --git a/test/upgrade/continual/channel-config.toml b/test/upgrade/continual/channel-config.toml
new file mode 100644
index 000000000..99f04dcf1
--- /dev/null
+++ b/test/upgrade/continual/channel-config.toml
@@ -0,0 +1,7 @@
+# logLevel = 'DEBUG'
+[sender]
+address = '{{- .Endpoint -}}'
+interval = {{ .Config.Interval.Nanoseconds }}
+
+[forwarder]
+target = 'http://wathola-receiver.{{- .Namespace -}}.svc.cluster.local'
diff --git a/test/upgrade/continual/channel.go b/test/upgrade/continual/channel.go
new file mode 100644
index 000000000..c082763a2
--- /dev/null
+++ b/test/upgrade/continual/channel.go
@@ -0,0 +1,275 @@
+/*
+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 continual
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	reconcilertesting "knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/testing"
+	testlib "knative.dev/eventing-kafka/test/lib"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/eventing/test/lib/duck"
+	"knative.dev/eventing/test/lib/resources"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
+)
+
+const (
+	channelConfigTemplatePath = "test/upgrade/continual/channel-config.toml"
+	defaultRetryCount         = 12
+	defaultBackoffPolicy      = eventingduckv1.BackoffPolicyExponential
+	defaultBackoffDelay       = "PT1S"
+)
+
+var (
+	defaultChannelType = metav1.TypeMeta{
+		APIVersion: "messaging.knative.dev/v1beta1",
+		Kind:       "KafkaChannel",
+	}
+)
+
+// ChannelTestOptions holds test options for KafkaChannel tests.
+type ChannelTestOptions struct {
+	*TestOptions
+	*ReplicationOptions
+	*RetryOptions
+	*metav1.TypeMeta
+}
+
+// ChannelTest tests channel operation in continual manner during the
+// whole upgrade and downgrade process asserting that all event are propagated
+// well.
+func ChannelTest(opts ChannelTestOptions) pkgupgrade.BackgroundOperation {
+	opts = opts.withDefaults()
+	return continualVerification(
+		"ChannelContinualTest",
+		opts.TestOptions,
+		channelSut(opts),
+		channelConfigTemplatePath,
+	)
+}
+
+// BrokerBackedByChannelTest tests a broker backed by KafkaChannel operation in
+// continual manner during the whole upgrade and downgrade process asserting
+// that all event are propagated well.
+func BrokerBackedByChannelTest(opts ChannelTestOptions) pkgupgrade.BackgroundOperation {
+	opts = opts.withDefaults()
+	return continualVerification(
+		"BrokerBackedByChannelContinualTest",
+		opts.TestOptions,
+		brokerBackedByChannelSut(opts),
+		channelConfigTemplatePath,
+	)
+}
+
+func (o ChannelTestOptions) withDefaults() ChannelTestOptions {
+	cto := o
+	if cto.TestOptions == nil {
+		cto.TestOptions = &TestOptions{}
+	}
+	if cto.TypeMeta == nil {
+		cto.TypeMeta = &defaultChannelType
+	}
+	if cto.RetryOptions == nil {
+		cto.RetryOptions = defaultRetryOptions()
+	}
+	if cto.ReplicationOptions == nil {
+		cto.ReplicationOptions = defaultReplicationOptions()
+	}
+	return cto
+}
+
+func channelSut(opts ChannelTestOptions) sut.SystemUnderTest {
+	return &kafkaChannelSut{
+		channelTypeMeta:    opts.TypeMeta,
+		ReplicationOptions: opts.ReplicationOptions,
+		RetryOptions:       opts.RetryOptions,
+	}
+}
+
+func defaultReplicationOptions() *ReplicationOptions {
+	return &ReplicationOptions{
+		NumPartitions:     6,
+		ReplicationFactor: 3,
+	}
+}
+
+func defaultRetryOptions() *RetryOptions {
+	return &RetryOptions{
+		RetryCount:    defaultRetryCount,
+		BackoffPolicy: defaultBackoffPolicy,
+		BackoffDelay:  defaultBackoffDelay,
+	}
+}
+
+func brokerBackedByChannelSut(opts ChannelTestOptions) sut.SystemUnderTest {
+	return &brokerBackedByKafkaChannelSut{
+		channelTypeMeta:    opts.TypeMeta,
+		ReplicationOptions: opts.ReplicationOptions,
+		RetryOptions:       opts.RetryOptions,
+	}
+}
+
+type kafkaChannelSut struct {
+	channelTypeMeta *metav1.TypeMeta
+	*ReplicationOptions
+	*RetryOptions
+}
+
+func (k kafkaChannelSut) Deploy(ctx sut.Context, destination duckv1.Destination) interface{} {
+	c := ctx.Client
+	name := "sut"
+	ch := reconcilertesting.NewKafkaChannel(
+		name,
+		c.Namespace,
+		k.RetryOptions.channelOption(),
+		k.ReplicationOptions.channelOption(),
+	)
+	testlib.CreateKafkaChannelV1Beta1OrFail(c, ch)
+	metaResource := resources.NewMetaResource(name, c.Namespace, k.channelTypeMeta)
+	if err := duck.WaitForResourceReady(c.Dynamic, metaResource); err != nil {
+		c.T.Fatal(err)
+	}
+
+	var sutUrl *apis.URL
+	if u, err := c.GetAddressableURI(name, k.channelTypeMeta); err != nil {
+		c.T.Fatal(err)
+	} else {
+		sutUrl, err = apis.ParseURL(u)
+		if err != nil {
+			c.T.Fatal(err)
+		}
+	}
+
+	c.CreateSubscriptionOrFail(
+		name, name, k.channelTypeMeta,
+		withDestinationForSubscription(&destination),
+		k.RetryOptions.subscriptionOption(),
+	)
+
+	return sutUrl
+}
+
+func withDestinationForSubscription(destination *duckv1.Destination) resources.SubscriptionOption {
+	return func(subscription *messagingv1.Subscription) {
+		subscription.Spec.Subscriber = destination
+	}
+}
+
+func (ro RetryOptions) subscriptionOption() resources.SubscriptionOption {
+	return func(subscription *messagingv1.Subscription) {
+		ensureSubscriptionHasDelivery(subscription)
+		r := int32(ro.RetryCount)
+		subscription.Spec.Delivery.Retry = &r
+		subscription.Spec.Delivery.BackoffPolicy = &ro.BackoffPolicy
+		subscription.Spec.Delivery.BackoffDelay = &ro.BackoffDelay
+	}
+}
+
+func (ro RetryOptions) channelOption() reconcilertesting.KafkaChannelOption {
+	return func(channel *v1beta1.KafkaChannel) {
+		ensureChannelHasDelivery(channel)
+		r := int32(ro.RetryCount)
+		channel.Spec.Delivery.Retry = &r
+		channel.Spec.Delivery.BackoffPolicy = &ro.BackoffPolicy
+		channel.Spec.Delivery.BackoffDelay = &ro.BackoffDelay
+	}
+}
+
+func (ro ReplicationOptions) channelOption() reconcilertesting.KafkaChannelOption {
+	return func(channel *v1beta1.KafkaChannel) {
+		channel.Spec.ReplicationFactor = int16(ro.ReplicationFactor)
+		channel.Spec.NumPartitions = int32(ro.NumPartitions)
+	}
+}
+
+func ensureChannelHasDelivery(channel *v1beta1.KafkaChannel) {
+	if channel.Spec.Delivery == nil {
+		channel.Spec.Delivery = &eventingduckv1.DeliverySpec{}
+	}
+}
+
+func ensureSubscriptionHasDelivery(subscription *messagingv1.Subscription) {
+	if subscription.Spec.Delivery == nil {
+		subscription.Spec.Delivery = &eventingduckv1.DeliverySpec{}
+	}
+}
+
+type brokerBackedByKafkaChannelSut struct {
+	*ReplicationOptions
+	*RetryOptions
+	channelTypeMeta *metav1.TypeMeta
+	defaultSut      sut.SystemUnderTest
+}
+
+func (b *brokerBackedByKafkaChannelSut) Deploy(
+	ctx sut.Context,
+	destination duckv1.Destination,
+) interface{} {
+	b.setKafkaAsDefaultForBroker(ctx)
+
+	b.defaultSut = sut.NewDefault()
+	return b.defaultSut.Deploy(ctx, destination)
+}
+
+func (b *brokerBackedByKafkaChannelSut) Teardown(ctx sut.Context) {
+	if b.defaultSut == nil {
+		ctx.T.Fatal("default SUT isn't set!?!")
+	}
+	if tr, ok := b.defaultSut.(sut.HasTeardown); ok {
+		tr.Teardown(ctx)
+	}
+}
+
+func (b *brokerBackedByKafkaChannelSut) setKafkaAsDefaultForBroker(ctx sut.Context) {
+	systemNs := "knative-eventing"
+	configmaps := ctx.Client.Kube.CoreV1().ConfigMaps(systemNs)
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: systemNs,
+			Name:      "config-br-default-channel",
+		},
+		Data: map[string]string{
+			"channelTemplateSpec": fmt.Sprintf(`apiVersion: %s
+kind: %s
+spec:
+  numPartitions: %d
+  replicationFactor: %d
+  delivery:
+    retry: %d
+    backoffPolicy: %s
+    backoffDelay: %s`,
+				b.channelTypeMeta.APIVersion, b.channelTypeMeta.Kind,
+				b.NumPartitions, b.ReplicationFactor,
+				b.RetryCount, b.BackoffPolicy, b.BackoffDelay,
+			),
+		},
+	}
+	cm, err := configmaps.Update(ctx.Ctx, cm, metav1.UpdateOptions{})
+	if err != nil {
+		ctx.T.Fatal(err)
+	}
+
+	ctx.Log.Info("Updated config-br-default-channel in ns knative-eventing"+
+		" to eq: ", cm.Data)
+}
diff --git a/test/upgrade/continual/kafkasender.go b/test/upgrade/continual/kafkasender.go
new file mode 100644
index 000000000..dbf9b7725
--- /dev/null
+++ b/test/upgrade/continual/kafkasender.go
@@ -0,0 +1,133 @@
+/*
+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 continual
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/Shopify/sarama"
+	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/binding"
+	"go.uber.org/zap"
+	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing/test/upgrade/prober/wathola/sender"
+)
+
+var (
+	// ErrIllegalEndpointFormat if given endpoint structure is illegal and can't
+	// be used.
+	ErrIllegalEndpointFormat = errors.New(
+		"illegal format for Kafka topic endpoint")
+
+	// ErrCantConnectToKafka if connection to kafka can't be established.
+	ErrCantConnectToKafka = errors.New(
+		"unable to connect to Kafka bootstrap servers")
+
+	// ErrCantSend if event can't be sent to given Kafka topic
+	ErrCantSend = errors.New("can't send event to kafka topic")
+)
+
+// CreateKafkaSender will create a wathola sender that sends events to Kafka
+// topic directly.
+func CreateKafkaSender(ctx context.Context, log *zap.SugaredLogger) sender.EventSender {
+	return &kafkaSender{
+		ctx: ctx,
+		log: log,
+	}
+}
+
+func (k *kafkaSender) Supports(endpoint interface{}) bool {
+	switch endpoint.(type) {
+	case map[string]interface{}:
+		_, err := castAsTopicEndpoint(endpoint)
+		return err == nil
+	default:
+		return false
+	}
+}
+
+func (k *kafkaSender) SendEvent(ce cloudevents.Event, rawEndpoint interface{}) error {
+	endpoint, err := castAsTopicEndpoint(rawEndpoint)
+	if err != nil {
+		// this should never happen, as Supports func should be called first.
+		return err
+	}
+	conf, err := client.NewConfigBuilder().
+		WithClientId("continualtests-kafka-sender").
+		WithDefaults().
+		Build(k.ctx)
+	if err != nil {
+		return err
+	}
+	producer, err := sarama.NewSyncProducer(endpoint.bootstrapServersSlice(), conf)
+	if err != nil {
+		return fmt.Errorf("%w: %v", ErrCantConnectToKafka, err)
+	}
+	message := binding.ToMessage(&ce)
+	kafkaProducerMessage := &sarama.ProducerMessage{
+		Topic: endpoint.TopicName,
+	}
+	transformers := make([]binding.Transformer, 0)
+	err = protocolkafka.WriteProducerMessage(k.ctx, message, kafkaProducerMessage, transformers...)
+	if err != nil {
+		return fmt.Errorf("%w: %v", ErrCantSend, err)
+	}
+	part, offset, err := producer.SendMessage(kafkaProducerMessage)
+	if err != nil {
+		return fmt.Errorf("%w: %v", ErrCantSend, err)
+	}
+	k.log.Infof("Event %s has been send to kafka topic %s (partition: %d, offset: %d)",
+		ce.ID(), endpoint.TopicName, part, offset)
+	return nil
+}
+
+type kafkaSender struct {
+	log *zap.SugaredLogger
+	ctx context.Context
+}
+
+type kafkaTopicEndpoint struct {
+	BootstrapServers string
+	TopicName        string
+}
+
+func (e kafkaTopicEndpoint) bootstrapServersSlice() []string {
+	return strings.Split(e.BootstrapServers, ",")
+}
+
+func castAsTopicEndpoint(endpoint interface{}) (kafkaTopicEndpoint, error) {
+	m, ok := endpoint.(map[string]interface{})
+	if !ok {
+		return kafkaTopicEndpoint{}, ErrIllegalEndpointFormat
+	}
+	servers := m["bootstrapServers"]
+	if servers == "" {
+		return kafkaTopicEndpoint{}, ErrIllegalEndpointFormat
+	}
+	topic := m["topicName"]
+	if topic == "" {
+		return kafkaTopicEndpoint{}, ErrIllegalEndpointFormat
+	}
+	return kafkaTopicEndpoint{
+		BootstrapServers: fmt.Sprintf("%v", servers),
+		TopicName:        fmt.Sprintf("%v", topic),
+	}, nil
+}
diff --git a/test/upgrade/continual/source-config.toml b/test/upgrade/continual/source-config.toml
new file mode 100644
index 000000000..0276bcd10
--- /dev/null
+++ b/test/upgrade/continual/source-config.toml
@@ -0,0 +1,9 @@
+# logLevel = 'DEBUG'
+[sender]
+interval = {{ .Config.Interval.Nanoseconds }}
+  [sender.address]
+    bootstrapServers = '{{- .Endpoint.BootstrapServers -}}'
+    topicName = '{{- .Endpoint.TopicName -}}'
+
+[forwarder]
+target = 'http://wathola-receiver.{{- .Namespace -}}.svc.cluster.local'
diff --git a/test/upgrade/continual/source.go b/test/upgrade/continual/source.go
new file mode 100644
index 000000000..ff771ee99
--- /dev/null
+++ b/test/upgrade/continual/source.go
@@ -0,0 +1,139 @@
+/*
+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 continual
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/google/uuid"
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/eventing-kafka/test/e2e/helpers"
+	contribtestlib "knative.dev/eventing-kafka/test/lib"
+	contribresources "knative.dev/eventing-kafka/test/lib/resources"
+	"knative.dev/eventing/test/upgrade/prober"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	pkgTest "knative.dev/pkg/test"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
+)
+
+const (
+	defaultKafkaBootstrapPort    = 9092
+	defaultKafkaClusterName      = "my-cluster"
+	defaultKafkaClusterNamespace = "kafka"
+	sourceConfigTemplatePath     = "test/upgrade/continual/source-config.toml"
+)
+
+// SourceTestOptions holds test options for KafkaSource tests.
+type SourceTestOptions struct {
+	*TestOptions
+	*KafkaCluster
+}
+
+// SourceTest tests source operation in continual manner during the
+// whole upgrade and downgrade process asserting that all event are propagated
+// well.
+func SourceTest(opts SourceTestOptions) pkgupgrade.BackgroundOperation {
+	opts = opts.withDefaults()
+	return continualVerification(
+		"SourceContinualTests",
+		opts.TestOptions,
+		&kafkaSourceSut{KafkaCluster: opts.KafkaCluster},
+		sourceConfigTemplatePath,
+	)
+}
+
+func (o SourceTestOptions) withDefaults() SourceTestOptions {
+	sto := o
+	if sto.TestOptions == nil {
+		sto.TestOptions = &TestOptions{}
+	}
+	if sto.KafkaCluster == nil {
+		sto.KafkaCluster = &KafkaCluster{}
+	}
+	c := sto.KafkaCluster.withDefaults()
+	sto.KafkaCluster = &c
+	sto.Configurators = append([]prober.Configurator{
+		func(config *prober.Config) error {
+			config.Wathola.ImageResolver = kafkaSourceSenderImageResolver
+			return nil
+		},
+	}, sto.Configurators...)
+	return sto
+}
+
+func (c KafkaCluster) withDefaults() KafkaCluster {
+	kc := c
+	if kc.Name == "" {
+		kc.Name = defaultKafkaClusterName
+	}
+	if kc.Namespace == "" {
+		kc.Namespace = defaultKafkaClusterNamespace
+	}
+	if len(kc.BootstrapServers) == 0 {
+		kc.BootstrapServers = []string{
+			fmt.Sprintf("%s-kafka-bootstrap.%s.svc:%d",
+				kc.Name,
+				kc.Namespace,
+				defaultKafkaBootstrapPort,
+			),
+		}
+	}
+	return kc
+}
+
+func kafkaSourceSenderImageResolver(component string) string {
+	if component == "wathola-sender" {
+		// replacing the original image with modified one from this repo
+		component = "wathola-kafka-sender"
+	}
+	return pkgTest.ImagePath(component)
+}
+
+type kafkaSourceSut struct {
+	*KafkaCluster
+}
+
+func (k kafkaSourceSut) Deploy(ctx sut.Context, destination duckv1.Destination) interface{} {
+	topicName := uuid.NewString()
+	c := k.KafkaCluster
+	helpers.MustCreateTopic(ctx.Client, c.Name, c.Namespace,
+		topicName, 6)
+	contribtestlib.CreateKafkaSourceV1Beta1OrFail(ctx.Client, contribresources.KafkaSourceV1Beta1(
+		c.serversLine(),
+		topicName,
+		toObjectReference(destination),
+	))
+	return kafkaTopicEndpoint{
+		BootstrapServers: k.KafkaCluster.serversLine(),
+		TopicName:        topicName,
+	}
+}
+
+func (c KafkaCluster) serversLine() string {
+	return strings.Join(c.BootstrapServers, ",")
+}
+
+func toObjectReference(destination duckv1.Destination) *corev1.ObjectReference {
+	return &corev1.ObjectReference{
+		APIVersion: destination.Ref.APIVersion,
+		Kind:       destination.Ref.Kind,
+		Namespace:  destination.Ref.Namespace,
+		Name:       destination.Ref.Name,
+	}
+}
diff --git a/test/upgrade/continual/types.go b/test/upgrade/continual/types.go
new file mode 100644
index 000000000..09fbb8b9f
--- /dev/null
+++ b/test/upgrade/continual/types.go
@@ -0,0 +1,49 @@
+/*
+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 continual
+
+import (
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/test/upgrade/prober"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+)
+
+// ReplicationOptions hold options for replication.
+type ReplicationOptions struct {
+	NumPartitions     int
+	ReplicationFactor int
+}
+
+// RetryOptions holds options for retries.
+type RetryOptions struct {
+	RetryCount    int
+	BackoffPolicy eventingduckv1.BackoffPolicyType
+	BackoffDelay  string
+}
+
+// KafkaCluster represents Kafka cluster endpoint.
+type KafkaCluster struct {
+	BootstrapServers []string
+	Name             string
+	Namespace        string
+}
+
+// TestOptions holds options for EventingKafka continual tests.
+type TestOptions struct {
+	prober.ContinualVerificationOptions
+	SUTs map[string]sut.SystemUnderTest
+}
diff --git a/test/upgrade/continual/verification.go b/test/upgrade/continual/verification.go
new file mode 100644
index 000000000..914ab4126
--- /dev/null
+++ b/test/upgrade/continual/verification.go
@@ -0,0 +1,84 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package continual
+
+import (
+	"fmt"
+
+	"github.com/kelseyhightower/envconfig"
+	"knative.dev/eventing/test/upgrade/prober"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
+)
+
+func continualVerification(
+	testName string,
+	opts *TestOptions,
+	defaultSut sut.SystemUnderTest,
+	configTemplate string,
+) pkgupgrade.BackgroundOperation {
+	return prober.NewContinualVerification(
+		testName,
+		verificationOptions(
+			opts,
+			resolveSut(testName, opts, defaultSut),
+			configTemplate,
+		),
+	)
+}
+
+func resolveSut(
+	testname string,
+	opts *TestOptions,
+	defaultSut sut.SystemUnderTest,
+) sut.SystemUnderTest {
+	var resolved sut.SystemUnderTest
+	if opts.SUTs != nil {
+		resolved = opts.SUTs[testname]
+	}
+	if resolved == nil {
+		resolved = defaultSut
+	}
+	return resolved
+}
+
+func verificationOptions(
+	opts *TestOptions,
+	theSut sut.SystemUnderTest,
+	configTemplate string,
+) prober.ContinualVerificationOptions {
+	return prober.ContinualVerificationOptions{
+		Configurators: append(
+			opts.Configurators,
+			configurator(theSut, configTemplate),
+		),
+		ClientOptions: opts.ClientOptions,
+	}
+}
+
+func configurator(theSut sut.SystemUnderTest, configTemplate string) prober.Configurator {
+	return func(config *prober.Config) error {
+		config.SystemUnderTest = theSut
+		// TODO: knative/eventing#5176 - this is cumbersome
+		config.ConfigTemplate = fmt.Sprintf("../../../../../../%s",
+			configTemplate)
+		// envconfig.Process invocation is repeated from within prober.NewConfig to
+		// make sure every knob is configurable, but using defaults from Eventing
+		// Kafka instead of Core. The prefix is also changed.
+		return envconfig.Process("eventing_kafka_upgrade_tests", config)
+	}
+}
diff --git a/test/upgrade/suite.go b/test/upgrade/suite.go
index 779595760..fd32b0655 100644
--- a/test/upgrade/suite.go
+++ b/test/upgrade/suite.go
@@ -17,6 +17,7 @@ limitations under the License.
 package upgrade
 
 import (
+	"knative.dev/eventing-kafka/test/upgrade/continual"
 	"knative.dev/eventing-kafka/test/upgrade/installation"
 	pkgupgrade "knative.dev/pkg/test/upgrade"
 )
@@ -37,10 +38,10 @@ func Suite() pkgupgrade.Suite {
 				ChannelPostDowngradeTest(),
 				SourcePostDowngradeTest(),
 			},
-			Continual: []pkgupgrade.BackgroundOperation{
-				ChannelContinualTest(ContinualTestOptions{}),
-				SourceContinualTest(ContinualTestOptions{}),
-			},
+			Continual: merge(
+				ChannelContinualTests(continual.ChannelTestOptions{}),
+				SourceContinualTests(continual.SourceTestOptions{}),
+			),
 		},
 		Installations: pkgupgrade.Installations{
 			Base: []pkgupgrade.Operation{
@@ -56,3 +57,15 @@ func Suite() pkgupgrade.Suite {
 		},
 	}
 }
+
+func merge(slices ...[]pkgupgrade.BackgroundOperation) []pkgupgrade.BackgroundOperation {
+	l := 0
+	for _, slice := range slices {
+		l += len(slice)
+	}
+	result := make([]pkgupgrade.BackgroundOperation, 0, l)
+	for _, slice := range slices {
+		result = append(result, slice...)
+	}
+	return result
+}
diff --git a/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go b/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go
index 3948862e0..e4e9b9f19 100644
--- a/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go
+++ b/vendor/knative.dev/eventing/pkg/adapter/v2/test/test_client.go
@@ -86,6 +86,10 @@ func (c *TestCloudEventsClient) Send(ctx context.Context, out event.Event) proto
 		return errors.New("totally not an http result")
 	} else if eventData.Type == "unit.sendFail" {
 		return http.NewResult(400, "%w", protocol.ResultNACK)
+	} else if eventData.Type == "unit.nonHttpRetry" {
+		var attempts []protocol.Result
+		attempts = append(attempts, errors.New("totally not an http result"))
+		return http.NewRetriesResult(http.NewResult(200, "%w", protocol.ResultACK), 1, time.Now(), attempts)
 	}
 	return http.NewResult(200, "%w", protocol.ResultACK)
 }
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go
index 440590760..6a92ab290 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go
@@ -32,6 +32,7 @@ import (
 )
 
 // EventTransformationForSubscriptionTestHelper is the helper function for channel_event_transformation_test
+// Deprecated, use reconciler-test based tests.
 func EventTransformationForSubscriptionTestHelper(
 	ctx context.Context,
 	t *testing.T,
diff --git a/vendor/knative.dev/eventing/test/lib/resources/eventing.go b/vendor/knative.dev/eventing/test/lib/resources/eventing.go
index cf7667330..d4720d2b8 100644
--- a/vendor/knative.dev/eventing/test/lib/resources/eventing.go
+++ b/vendor/knative.dev/eventing/test/lib/resources/eventing.go
@@ -242,22 +242,31 @@ func WithDependencyAnnotationTrigger(dependencyAnnotation string) TriggerOption
 
 // WithSubscriberServiceRefForTrigger returns an option that adds a Subscriber Knative Service Ref for the given v1 Trigger.
 func WithSubscriberServiceRefForTrigger(name string) TriggerOption {
-	return func(t *eventingv1.Trigger) {
-		if name != "" {
-			t.Spec.Subscriber = duckv1.Destination{
-				Ref: KnativeRefForService(name, t.Namespace),
-			}
+	return WithSubscriberDestination(func(t *eventingv1.Trigger) duckv1.Destination {
+		return duckv1.Destination{
+			Ref: KnativeRefForService(name, t.Namespace),
 		}
-	}
+	})
 }
 
 // WithSubscriberURIForTrigger returns an option that adds a Subscriber URI for the given v1 Trigger.
 func WithSubscriberURIForTrigger(uri string) TriggerOption {
-	apisURI, _ := apis.ParseURL(uri)
-	return func(t *eventingv1.Trigger) {
-		t.Spec.Subscriber = duckv1.Destination{
+	return WithSubscriberDestination(func(t *eventingv1.Trigger) duckv1.Destination {
+		apisURI, _ := apis.ParseURL(uri)
+		return duckv1.Destination{
 			URI: apisURI,
 		}
+	})
+}
+
+// WithSubscriberDestination returns an option that adds a Subscriber for given
+// duckv1.Destination.
+func WithSubscriberDestination(destFactory func(t *eventingv1.Trigger) duckv1.Destination) TriggerOption {
+	return func(t *eventingv1.Trigger) {
+		dest := destFactory(t)
+		if dest.Ref != nil || dest.URI != nil {
+			t.Spec.Subscriber = dest
+		}
 	}
 }
 
diff --git a/vendor/knative.dev/eventing/test/lib/test_runner.go b/vendor/knative.dev/eventing/test/lib/test_runner.go
index 8e17f4af6..4066da298 100644
--- a/vendor/knative.dev/eventing/test/lib/test_runner.go
+++ b/vendor/knative.dev/eventing/test/lib/test_runner.go
@@ -214,7 +214,7 @@ func TearDown(client *Client) {
 
 	// If the test is run by CI, export the pod logs in the namespace to the artifacts directory,
 	// which will then be uploaded to GCS after the test job finishes.
-	if prow.IsCI() {
+	if prow.IsCI() && client.T.Failed() {
 		dir := filepath.Join(prow.GetLocalArtifactsDir(), podLogsDir)
 		client.T.Logf("Export logs in %q to %q", client.Namespace, dir)
 		if err := client.ExportLogs(dir); err != nil {
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/config.toml b/vendor/knative.dev/eventing/test/upgrade/prober/config.toml
index ca5c1c7c9..20afa87d6 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/config.toml
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/config.toml
@@ -1,6 +1,6 @@
 # logLevel = 'DEBUG'
 [sender]
-address = '{{- .BrokerURL -}}'
+address = '{{- .Endpoint -}}'
 interval = {{ .Config.Interval.Nanoseconds }}
 [forwarder]
-target = 'http://wathola-receiver.{{- .Config.Namespace -}}.svc.cluster.local'
+target = 'http://wathola-receiver.{{- .Namespace -}}.svc.cluster.local'
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
index a195f7d4f..ba1057afd 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
@@ -20,68 +20,77 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
+	"os"
 	"path"
 	"runtime"
+	"strings"
 	"text/template"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
 	"github.com/wavesoftware/go-ensure"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/duck"
 	"knative.dev/eventing/test/lib/resources"
-	"knative.dev/pkg/apis"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	pkgTest "knative.dev/pkg/test"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
 )
 
 const (
-	defaultConfigName          = "wathola-config"
-	defaultConfigHomedirPath   = ".config/wathola"
-	defaultHomedir             = "/home/nonroot"
-	defaultConfigFilename      = "config.toml"
-	defaultWatholaEventsPrefix = "com.github.cardil.wathola"
-	defaultBrokerName          = "default"
-	defaultHealthEndpoint      = "/healthz"
-	defaultFinishedSleep       = 5 * time.Second
+	defaultConfigName        = "wathola-config"
+	defaultConfigHomedirPath = ".config/wathola"
+	defaultHomedir           = "/home/nonroot"
+	defaultConfigFilename    = "config.toml"
+	defaultHealthEndpoint    = "/healthz"
+	defaultFinishedSleep     = 5 * time.Second
 
 	Silence DuplicateAction = "silence"
 	Warn    DuplicateAction = "warn"
 	Error   DuplicateAction = "error"
+
+	prefix = "eventing_upgrade_tests"
+
+	// TODO(ksuszyns): Remove this val in next release
+	// Deprecated: use 'eventing_upgrade_tests' prefix instead
+	deprecatedPrefix = "e2e_upgrade_tests"
 )
 
 // DuplicateAction is the action to take in case of duplicated events
 type DuplicateAction string
 
-var eventTypes = []string{"step", "finished"}
-
 // Config represents a configuration for prober.
 type Config struct {
 	Wathola
-	Namespace     string
 	Interval      time.Duration
 	FinishedSleep time.Duration
 	Serving       ServingConfig
 	FailOnErrors  bool
 	OnDuplicate   DuplicateAction
-	BrokerOpts    []resources.BrokerOption
+	Ctx           context.Context
+	// BrokerOpts holds opts for broker.
+	// TODO(ksuszyns): Remove this opt in next release
+	// Deprecated: use Wathola.SystemUnderTest instead.
+	BrokerOpts []resources.BrokerOption
+	// Namespace holds namespace in which test is about to be executed.
+	// TODO(ksuszyns): Remove this opt in next release
+	// Deprecated: namespace is about to be taken from testlib.Client created by
+	// NewRunner
+	Namespace string
 }
 
 // Wathola represents options related strictly to wathola testing tool.
 type Wathola struct {
-	ConfigMap
+	ConfigToml
 	ImageResolver
-	EventsTypePrefix string
-	HealthEndpoint   string
-	BrokerName       string
+	SystemUnderTest sut.SystemUnderTest
+	HealthEndpoint  string
 }
 
 // ImageResolver will resolve the container image for given component.
 type ImageResolver func(component string) string
 
-// ConfigMap represents options of wathola config toml file.
-type ConfigMap struct {
+// ConfigToml represents options of wathola config toml file.
+type ConfigToml struct {
 	// ConfigTemplate is a template file that will be compiled to the configmap
 	ConfigTemplate   string
 	ConfigMapName    string
@@ -95,123 +104,130 @@ type ServingConfig struct {
 	ScaleToZero bool
 }
 
+// NewConfigOrFail will create a prober.Config or fail trying.
+func NewConfigOrFail(c pkgupgrade.Context) *Config {
+	errSink := func(err error) {
+		c.T.Fatal(err)
+	}
+	warnf := c.Log.Warnf
+	return newConfig(errSink, warnf)
+}
+
 // NewConfig creates a new configuration object with default values filled in.
 // Values can be influenced by kelseyhightower/envconfig with
-// `e2e_upgrade_tests` prefix.
-func NewConfig(namespace string) *Config {
+// `eventing_upgrade_tests` prefix.
+// TODO(ksuszyns): Remove this func in next release
+// Deprecated: use NewContinualVerification or NewConfigOrFail
+func NewConfig(namespace ...string) *Config {
+	errSink := func(err error) {
+		ensure.NoError(err)
+	}
+	warnf := func(template string, args ...interface{}) {
+		_, err := fmt.Fprintf(os.Stderr, template, args)
+		ensure.NoError(err)
+	}
+	config := newConfig(errSink, warnf)
+	if len(namespace) > 0 {
+		config.Namespace = strings.Join(namespace, ",")
+	}
+	return config
+}
+
+func newConfig(
+	errSink func(err error), warnf func(template string, args ...interface{}),
+) *Config {
 	config := &Config{
-		Namespace:     "",
 		Interval:      Interval,
 		FinishedSleep: defaultFinishedSleep,
 		FailOnErrors:  true,
 		OnDuplicate:   Warn,
-		BrokerOpts:    make([]resources.BrokerOption, 0),
+		Ctx:           context.Background(),
 		Serving: ServingConfig{
 			Use:         false,
 			ScaleToZero: true,
 		},
 		Wathola: Wathola{
 			ImageResolver: pkgTest.ImagePath,
-			ConfigMap: ConfigMap{
+			ConfigToml: ConfigToml{
 				ConfigTemplate:   defaultConfigFilename,
 				ConfigMapName:    defaultConfigName,
 				ConfigMountPoint: fmt.Sprintf("%s/%s", defaultHomedir, defaultConfigHomedirPath),
 				ConfigFilename:   defaultConfigFilename,
 			},
-			EventsTypePrefix: defaultWatholaEventsPrefix,
-			HealthEndpoint:   defaultHealthEndpoint,
-			BrokerName:       defaultBrokerName,
+			HealthEndpoint:  defaultHealthEndpoint,
+			SystemUnderTest: sut.NewDefault(),
 		},
 	}
 
-	// FIXME: remove while fixing https://github.com/knative/eventing/issues/2665
-	config.FailOnErrors = false
+	if err := envconfig.Process(prefix, config); err != nil {
+		errSink(err)
+	}
 
-	err := envconfig.Process("e2e_upgrade_tests", config)
-	ensure.NoError(err)
-	config.Namespace = namespace
+	// TODO(ksuszyns): Remove this block in next release
+	for _, enventry := range os.Environ() {
+		if strings.HasPrefix(strings.ToLower(enventry), deprecatedPrefix) {
+			warnf(
+				"DEPRECATED: using deprecated '%s' prefix. Use '%s' instead.",
+				deprecatedPrefix, prefix)
+			if err := envconfig.Process(deprecatedPrefix, config); err != nil {
+				errSink(err)
+			}
+			break
+		}
+	}
 	return config
 }
 
 func (p *prober) deployConfiguration() {
-	p.deployBroker()
-	p.deployConfigMap()
-	p.deployTriggers()
-}
-
-func (p *prober) deployBroker() {
-	p.client.CreateBrokerOrFail(p.config.BrokerName, p.config.BrokerOpts...)
-}
-
-func (p *prober) fetchBrokerURL() (*apis.URL, error) {
-	namespace := p.config.Namespace
-	p.log.Debugf("Fetching %s broker URL for ns %s",
-		p.config.BrokerName, namespace)
-	meta := resources.NewMetaResource(
-		p.config.BrokerName, p.config.Namespace, testlib.BrokerTypeMeta,
-	)
-	err := duck.WaitForResourceReady(p.client.Dynamic, meta)
-	if err != nil {
-		return nil, err
+	sc := sut.Context{
+		Ctx:    p.config.Ctx,
+		Log:    p.log,
+		Client: p.client,
 	}
-	broker, err := p.client.Eventing.EventingV1().Brokers(namespace).Get(
-		context.Background(), p.config.BrokerName, metav1.GetOptions{},
-	)
-	if err != nil {
-		return nil, err
+	ref := resources.KnativeRefForService(receiverName, p.client.Namespace)
+	if p.config.Serving.Use {
+		ref = resources.KnativeRefForKservice(forwarderName, p.client.Namespace)
 	}
-	url := broker.Status.Address.URL
-	p.log.Debugf("%s broker URL for ns %s is %v",
-		p.config.BrokerName, namespace, url)
-	return url, nil
+	dest := duckv1.Destination{Ref: ref}
+	s := p.config.SystemUnderTest
+	endpoint := s.Deploy(sc, dest)
+	p.client.Cleanup(func() {
+		if tr, ok := s.(sut.HasTeardown); ok {
+			tr.Teardown(sc)
+		}
+	})
+	p.deployConfigToml(endpoint)
 }
 
-func (p *prober) deployConfigMap() {
+func (p *prober) deployConfigToml(endpoint interface{}) {
 	name := p.config.ConfigMapName
-	p.log.Infof("Deploying config map: \"%s/%s\"", p.config.Namespace, name)
-	brokerURL, err := p.fetchBrokerURL()
-	ensure.NoError(err)
-	configData := p.compileTemplate(p.config.ConfigTemplate, brokerURL)
-	p.client.CreateConfigMapOrFail(name, p.config.Namespace, map[string]string{
+	p.log.Infof("Deploying config map: \"%s/%s\"", p.client.Namespace, name)
+	configData := p.compileTemplate(p.config.ConfigTemplate, endpoint)
+	p.client.CreateConfigMapOrFail(name, p.client.Namespace, map[string]string{
 		p.config.ConfigFilename: configData,
 	})
 }
 
-func (p *prober) deployTriggers() {
-	for _, eventType := range eventTypes {
-		name := fmt.Sprintf("wathola-trigger-%v", eventType)
-		fullType := fmt.Sprintf("%v.%v", p.config.EventsTypePrefix, eventType)
-		subscriberOption := resources.WithSubscriberServiceRefForTrigger(receiverName)
-		if p.config.Serving.Use {
-			subscriberOption = resources.WithSubscriberKServiceRefForTrigger(forwarderName)
-		}
-		p.client.CreateTriggerOrFail(name,
-			resources.WithBroker(p.config.BrokerName),
-			resources.WithAttributesTriggerFilter(
-				eventingv1.TriggerAnyFilter,
-				fullType,
-				map[string]interface{}{},
-			),
-			subscriberOption,
-		)
-	}
-}
-
-func (p *prober) compileTemplate(templateName string, brokerURL fmt.Stringer) string {
+func (p *prober) compileTemplate(templateName string, endpoint interface{}) string {
 	_, filename, _, _ := runtime.Caller(0)
 	templateFilepath := path.Join(path.Dir(filename), templateName)
 	templateBytes, err := ioutil.ReadFile(templateFilepath)
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	tmpl, err := template.New(templateName).Parse(string(templateBytes))
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	var buff bytes.Buffer
 	data := struct {
 		*Config
+		Namespace string
+		// Deprecated: use Endpoint
 		BrokerURL string
+		Endpoint  interface{}
 	}{
 		p.config,
-		brokerURL.String(),
+		p.client.Namespace,
+		fmt.Sprintf("%v", endpoint),
+		endpoint,
 	}
-	ensure.NoError(tmpl.Execute(&buff, data))
+	p.ensureNoError(tmpl.Execute(&buff, data))
 	return buff.String()
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/continual.go b/vendor/knative.dev/eventing/test/upgrade/prober/continual.go
new file mode 100644
index 000000000..237fdb582
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/continual.go
@@ -0,0 +1,59 @@
+/*
+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 prober
+
+import (
+	testlib "knative.dev/eventing/test/lib"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
+)
+
+// Configurator will customize default config.
+type Configurator func(config *Config) error
+
+// ContinualVerificationOptions holds options for NewContinualVerification func.
+type ContinualVerificationOptions struct {
+	Configurators []Configurator
+	ClientOptions []testlib.SetupClientOption
+}
+
+// NewContinualVerification will create a new continual verification operation
+// that will verify that SUT is propagating events well through the test
+// process. It's a general pattern that can be used to validate upgrade or
+// performance testing.
+func NewContinualVerification(
+	name string, opts ContinualVerificationOptions,
+) pkgupgrade.BackgroundOperation {
+	var config *Config
+	var runner Runner
+	setup := func(c pkgupgrade.Context) {
+		config = NewConfigOrFail(c)
+		if len(opts.Configurators) > 0 {
+			for _, fn := range opts.Configurators {
+				err := fn(config)
+				if err != nil {
+					c.T.Fatal(err)
+				}
+			}
+		}
+		runner = NewRunner(config, opts.ClientOptions...)
+		runner.Setup(c)
+	}
+	verify := func(c pkgupgrade.Context) {
+		runner.Verify(c)
+	}
+	return pkgupgrade.NewBackgroundVerification(name, setup, verify)
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go b/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go
index 3406254ca..14965fafb 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/forwarder.go
@@ -16,10 +16,8 @@
 package prober
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/wavesoftware/go-ensure"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	testlib "knative.dev/eventing/test/lib"
@@ -31,12 +29,13 @@ var (
 	forwarderName = "wathola-forwarder"
 )
 
-func (p *prober) deployForwarder(ctx context.Context) {
+func (p *prober) deployForwarder() {
 	p.log.Infof("Deploy forwarder knative service: %v", forwarderName)
 	serving := p.client.Dynamic.Resource(resources.KServicesGVR).Namespace(p.client.Namespace)
 	service := p.forwarderKService(forwarderName, p.client.Namespace)
-	_, err := serving.Create(context.Background(), service, metav1.CreateOptions{})
-	ensure.NoError(err)
+	if _, err := serving.Create(p.config.Ctx, service, metav1.CreateOptions{}); err != nil {
+		p.client.T.Fatal(err)
+	}
 
 	sc := p.servingClient()
 	testlib.WaitFor(fmt.Sprintf("forwarder ksvc be ready: %v", forwarderName), func() error {
@@ -45,7 +44,7 @@ func (p *prober) deployForwarder(ctx context.Context) {
 
 	if p.config.Serving.ScaleToZero {
 		testlib.WaitFor(fmt.Sprintf("forwarder scales to zero: %v", forwarderName), func() error {
-			return duck.WaitForKServiceScales(ctx, sc, forwarderName, p.client.Namespace, func(scale int) bool {
+			return duck.WaitForKServiceScales(p.config.Ctx, sc, forwarderName, p.client.Namespace, func(scale int) bool {
 				return scale == 0
 			})
 		})
@@ -55,8 +54,8 @@ func (p *prober) deployForwarder(ctx context.Context) {
 func (p *prober) removeForwarder() {
 	p.log.Infof("Remove forwarder knative service: %v", forwarderName)
 	serving := p.client.Dynamic.Resource(resources.KServicesGVR).Namespace(p.client.Namespace)
-	err := serving.Delete(context.Background(), forwarderName, metav1.DeleteOptions{})
-	ensure.NoError(err)
+	err := serving.Delete(p.config.Ctx, forwarderName, metav1.DeleteOptions{})
+	p.ensureNoError(err)
 }
 
 func (p *prober) forwarderKService(name, namespace string) *unstructured.Unstructured {
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
index bc6623bf5..201dc6862 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
@@ -17,63 +17,144 @@ package prober
 
 import (
 	"context"
+	"reflect"
 	"testing"
 	"time"
 
-	"github.com/wavesoftware/go-ensure"
 	"go.uber.org/zap"
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/resources"
+	"knative.dev/eventing/test/upgrade/prober/sut"
+	pkgupgrade "knative.dev/pkg/test/upgrade"
 )
 
 var (
-	// FIXME: Interval is set to 200 msec, as lower values will result in errors: knative/eventing#2357
-	// Interval = 10 * time.Millisecond
-	Interval = 200 * time.Millisecond
+	// Interval is used to send events in specific rate.
+	Interval = 10 * time.Millisecond
 )
 
-// Prober is the interface for a prober, which checks the result of the probes when stopped.
+// Prober is the interface for a prober, which checks the result of the probes
+// when stopped.
+// TODO(ksuszyns): Remove this interface in next release
+// Deprecated: use Runner instead, create it with NewRunner func.
 type Prober interface {
 	// Verify will verify prober state after finished has been send
-	Verify(ctx context.Context) ([]error, int, error)
+	Verify() ([]error, int)
 
 	// Finish send finished event
-	Finish(ctx context.Context)
+	Finish()
 
 	// ReportErrors will reports found errors in proper way
-	ReportErrors(t *testing.T, errors []error)
+	ReportErrors(errors []error)
+}
+
+// Runner will run continual verification with provided configuration.
+type Runner interface {
+	// Setup will start a continual prober in background.
+	Setup(ctx pkgupgrade.Context)
 
-	// deploy a prober to a cluster
-	deploy(ctx context.Context)
-	// remove a prober from cluster
-	remove()
+	// Verify will verify that all sent events propagated at least once.
+	Verify(ctx pkgupgrade.Context)
 }
 
-// RunEventProber starts a single Prober of the given domain.
-func RunEventProber(ctx context.Context, log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
-	pm := newProber(log, client, config)
-	pm.deploy(ctx)
-	return pm
+// NewRunner will create a runner compatible with NewContinualVerification
+// func.
+func NewRunner(config *Config, options ...testlib.SetupClientOption) Runner {
+	return &probeRunner{
+		prober:  &prober{config: config},
+		options: options,
+	}
 }
 
-// AssertEventProber will send finish event and then verify if all events propagated well
-func AssertEventProber(ctx context.Context, t *testing.T, prober Prober) {
-	prober.Finish(ctx)
-	defer prober.remove()
+type probeRunner struct {
+	*prober
+	options []testlib.SetupClientOption
+}
 
-	waitAfterFinished(prober)
+func (p *probeRunner) Setup(ctx pkgupgrade.Context) {
+	p.validate(ctx)
+	p.log = ctx.Log
+	p.client = testlib.Setup(ctx.T, false, p.options...)
+	p.deploy()
+}
 
-	eventErrs, eventCount, err := prober.Verify(ctx)
-	if err != nil {
-		t.Fatal("fetch error:", err)
+func (p *probeRunner) Verify(ctx pkgupgrade.Context) {
+	if p.client == nil {
+		ctx.T.Fatal("prober isn't initiated (client is nil)")
+		return
 	}
-	if len(eventErrs) == 0 {
-		t.Logf("All %d events propagated well", eventCount)
+	// use T from new test
+	p.client.T = ctx.T
+	t := ctx.T
+	defer testlib.TearDown(p.client)
+	defer p.remove()
+	p.Finish()
+	waitAfterFinished(p.prober)
+
+	errors, events := p.prober.Verify()
+	if len(errors) == 0 {
+		t.Logf("All %d events propagated well", events)
 	} else {
 		t.Logf("There were %d events propagated, but %d errors occurred. "+
-			"Listing them below.", eventCount, len(eventErrs))
+			"Listing them below.", events, len(errors))
+	}
+
+	p.ReportErrors(errors)
+}
+
+func (p *probeRunner) validate(ctx pkgupgrade.Context) {
+	if p.config.Namespace != "" {
+		ctx.Log.Warnf(
+			"DEPRECATED: namespace set in Config: %s. Ignoring it.",
+			p.client.Namespace)
+	}
+	if len(p.config.BrokerOpts) > 0 {
+		ctx.Log.Warn(
+			"DEPRECATED: BrokerOpts set in Config. Use custom SystemUnderTest")
+		if reflect.ValueOf(p.config.Wathola.SystemUnderTest) == reflect.ValueOf(sut.NewDefault) {
+			bt := sut.NewDefault().(*sut.BrokerAndTriggers)
+			bt.Opts = append(bt.Opts, p.config.BrokerOpts...)
+			p.config.Wathola.SystemUnderTest = bt
+		} else {
+			ctx.T.Fatal("Can't use given BrokerOpts, as custom SUT is used as " +
+				"well. Drop using BrokerOpts in favor of custom SUT.")
+		}
 	}
-	prober.ReportErrors(t, eventErrs)
+}
+
+// RunEventProber starts a single Prober of the given domain.
+// TODO(ksuszyns): Remove this func in next release
+// Deprecated: use NewRunner func instead.
+func RunEventProber(ctx context.Context, log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
+	log.Warn("prober.RunEventProber is deprecated. Use NewRunner instead.")
+	config.Ctx = ctx
+	p := &prober{
+		log:    log,
+		client: client,
+		config: config,
+	}
+	p.deploy()
+	return p
+}
+
+// AssertEventProber will send finish event and then verify if all events
+// propagated well.
+// TODO(ksuszyns): Remove this func in next release
+// Deprecated: use NewRunner func instead.
+func AssertEventProber(ctx context.Context, t *testing.T, probe Prober) {
+	t.Log("WARN: prober.AssertEventProber is deprecated. " +
+		"Use NewRunner instead.")
+	p := probe.(*prober)
+	p.client.T = t
+	p.config.Ctx = ctx
+	pr := &probeRunner{
+		prober:  p,
+		options: nil,
+	}
+	pr.Verify(pkgupgrade.Context{
+		T:   t,
+		Log: p.log,
+	})
 }
 
 type prober struct {
@@ -89,7 +170,8 @@ func (p *prober) servingClient() resources.ServingClient {
 	}
 }
 
-func (p *prober) ReportErrors(t *testing.T, errors []error) {
+func (p *prober) ReportErrors(errors []error) {
+	t := p.client.T
 	for _, err := range errors {
 		if p.config.FailOnErrors {
 			t.Error(err)
@@ -105,17 +187,17 @@ func (p *prober) ReportErrors(t *testing.T, errors []error) {
 	}
 }
 
-func (p *prober) deploy(ctx context.Context) {
+func (p *prober) deploy() {
 	p.log.Infof("Using namespace for probe testing: %v", p.client.Namespace)
 	p.deployConfiguration()
-	p.deployReceiver(ctx)
+	p.deployReceiver()
 	if p.config.Serving.Use {
-		p.deployForwarder(ctx)
+		p.deployForwarder()
 	}
-	p.client.WaitForAllTestResourcesReadyOrFail(ctx)
+	p.client.WaitForAllTestResourcesReadyOrFail(p.config.Ctx)
 
-	p.deploySender(ctx)
-	ensure.NoError(testlib.AwaitForAll(p.log))
+	p.deploySender()
+	p.ensureNoError(testlib.AwaitForAll(p.log))
 	// allow sender to send at least some events, 2 sec wait
 	time.Sleep(2 * time.Second)
 	p.log.Infof("Prober is now sending events with interval of %v in "+
@@ -126,20 +208,10 @@ func (p *prober) remove() {
 	if p.config.Serving.Use {
 		p.removeForwarder()
 	}
-	ensure.NoError(p.client.Tracker.Clean(true))
-}
-
-func newProber(log *zap.SugaredLogger, client *testlib.Client, config *Config) Prober {
-	return &prober{
-		log:    log,
-		client: client,
-		config: config,
-	}
+	p.ensureNoError(p.client.Tracker.Clean(true))
 }
 
-func waitAfterFinished(p Prober) {
-	s := p.(*prober)
-	cfg := s.config
-	s.log.Infof("Waiting %v after sender finished...", cfg.FinishedSleep)
-	time.Sleep(cfg.FinishedSleep)
+func waitAfterFinished(p *prober) {
+	p.log.Infof("Waiting %v after sender finished...", p.config.FinishedSleep)
+	time.Sleep(p.config.FinishedSleep)
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go b/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go
index f10145f8d..d55379881 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/receiver.go
@@ -16,10 +16,8 @@
 package prober
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/wavesoftware/go-ensure"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -30,36 +28,32 @@ import (
 )
 
 var (
-	receiverName           = "wathola-receiver"
-	receiverNodePort int32 = -1
+	receiverName = "wathola-receiver"
 )
 
-func (p *prober) deployReceiver(ctx context.Context) {
-	p.deployReceiverDeployment(ctx)
-	p.deployReceiverService(ctx)
+func (p *prober) deployReceiver() {
+	p.deployReceiverDeployment()
+	p.deployReceiverService()
 }
 
-func (p *prober) deployReceiverDeployment(ctx context.Context) {
+func (p *prober) deployReceiverDeployment() {
 	p.log.Info("Deploy of receiver deployment: ", receiverName)
 	deployment := p.createReceiverDeployment()
-	_, err := p.client.Kube.AppsV1().
-		Deployments(deployment.Namespace).
-		Create(ctx, deployment, metav1.CreateOptions{})
-	ensure.NoError(err)
+	p.client.CreateDeploymentOrFail(deployment)
 
 	testlib.WaitFor(fmt.Sprint("receiver deployment be ready: ", receiverName), func() error {
 		return pkgTest.WaitForDeploymentScale(
-			ctx, p.client.Kube, receiverName, p.client.Namespace, 1,
+			p.config.Ctx, p.client.Kube, receiverName, p.client.Namespace, 1,
 		)
 	})
 }
 
-func (p *prober) deployReceiverService(ctx context.Context) {
+func (p *prober) deployReceiverService() {
 	p.log.Infof("Deploy of receiver service: %v", receiverName)
 	service := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      receiverName,
-			Namespace: p.config.Namespace,
+			Namespace: p.client.Namespace,
 		},
 		Spec: corev1.ServiceSpec{
 			Ports: []corev1.ServicePort{
@@ -76,22 +70,10 @@ func (p *prober) deployReceiverService(ctx context.Context) {
 			Selector: map[string]string{
 				"app": receiverName,
 			},
-			Type: corev1.ServiceTypeNodePort,
+			Type: corev1.ServiceTypeClusterIP,
 		},
 	}
-	created, err := p.client.Kube.CoreV1().Services(p.config.Namespace).
-		Create(ctx, service, metav1.CreateOptions{})
-	ensure.NoError(err)
-	for _, portSpec := range created.Spec.Ports {
-		if portSpec.Port == 80 {
-			receiverNodePort = portSpec.NodePort
-		}
-	}
-	if receiverNodePort == -1 {
-		panic(fmt.Errorf("couldn't find a node port for service: %v", receiverName))
-	} else {
-		p.log.Debugf("Node port for service: %v is %v", receiverName, receiverNodePort)
-	}
+	p.client.CreateServiceOrFail(service)
 }
 
 func (p *prober) createReceiverDeployment() *appsv1.Deployment {
@@ -99,7 +81,7 @@ func (p *prober) createReceiverDeployment() *appsv1.Deployment {
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      receiverName,
-			Namespace: p.config.Namespace,
+			Namespace: p.client.Namespace,
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: &replicas,
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sender.go b/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
index bde438fab..5f3bdd4ec 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sender.go
@@ -16,10 +16,8 @@
 package prober
 
 import (
-	"context"
 	"fmt"
 
-	"github.com/wavesoftware/go-ensure"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -29,13 +27,13 @@ import (
 
 var senderName = "wathola-sender"
 
-func (p *prober) deploySender(ctx context.Context) {
+func (p *prober) deploySender() {
 	p.log.Info("Deploy sender deployment: ", senderName)
 	var replicas int32 = 1
 	deployment := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      senderName,
-			Namespace: p.config.Namespace,
+			Namespace: p.client.Namespace,
 		},
 		Spec: appsv1.DeploymentSpec{
 			Replicas: &replicas,
@@ -77,21 +75,21 @@ func (p *prober) deploySender(ctx context.Context) {
 
 	_, err := p.client.Kube.AppsV1().
 		Deployments(p.client.Namespace).
-		Create(ctx, deployment, metav1.CreateOptions{})
-	ensure.NoError(err)
+		Create(p.config.Ctx, deployment, metav1.CreateOptions{})
+	p.ensureNoError(err)
 
 	testlib.WaitFor(fmt.Sprint("sender deployment be ready: ", senderName), func() error {
 		return pkgTest.WaitForDeploymentScale(
-			ctx, p.client.Kube, senderName, p.client.Namespace, int(replicas),
+			p.config.Ctx, p.client.Kube, senderName, p.client.Namespace, int(replicas),
 		)
 	})
 }
 
-func (p *prober) removeSender(ctx context.Context) {
+func (p *prober) removeSender() {
 	p.log.Info("Remove of sender deployment: ", senderName)
 
 	err := p.client.Kube.AppsV1().
 		Deployments(p.client.Namespace).
-		Delete(ctx, senderName, metav1.DeleteOptions{})
-	ensure.NoError(err)
+		Delete(p.config.Ctx, senderName, metav1.DeleteOptions{})
+	p.ensureNoError(err)
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sut/broker.go b/vendor/knative.dev/eventing/test/upgrade/prober/sut/broker.go
new file mode 100644
index 000000000..a468a3b4a
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sut/broker.go
@@ -0,0 +1,141 @@
+/*
+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 sut
+
+import (
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/duck"
+	"knative.dev/eventing/test/lib/resources"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var (
+	retryCount    = int32(12)
+	backoffPolicy = eventingduckv1.BackoffPolicyExponential
+	backoffDelay  = "PT1S"
+)
+
+// BrokerAndTriggers will deploy a default broker and 2 triggers to route two types of
+// events to receiver.
+type BrokerAndTriggers struct {
+	Broker
+	Triggers
+}
+
+// Broker will hold settings for broker itself
+type Broker struct {
+	Name string
+	Opts []resources.BrokerOption
+}
+
+// Triggers will hold settings for triggers
+type Triggers struct {
+	Types []string
+}
+
+// NewBrokerAndTriggers will create default configuration for BrokerAndTriggers
+// based SUT.
+func NewBrokerAndTriggers() SystemUnderTest {
+	return &BrokerAndTriggers{
+		Broker: Broker{
+			Name: "sut",
+			Opts: []resources.BrokerOption{
+				resources.WithDeliveryForBroker(
+					&eventingduckv1.DeliverySpec{
+						Retry:         &retryCount,
+						BackoffPolicy: &backoffPolicy,
+						BackoffDelay:  &backoffDelay,
+					}),
+			},
+		},
+		Triggers: Triggers{
+			Types: eventTypes,
+		},
+	}
+}
+
+func (b *BrokerAndTriggers) Deploy(ctx Context, dest duckv1.Destination) interface{} {
+	b.deployBroker(ctx)
+	url := b.fetchURL(ctx)
+	b.deployTriggers(ctx, dest)
+	return url
+}
+
+func (b *BrokerAndTriggers) deployBroker(ctx Context) {
+	ctx.Client.CreateBrokerOrFail(b.Name, b.Broker.Opts...)
+}
+
+func (b *BrokerAndTriggers) fetchURL(ctx Context) *apis.URL {
+	namespace := ctx.Client.Namespace
+	ctx.Log.Debugf("Fetching \"%s\" broker URL for ns %s",
+		b.Name, namespace)
+	meta := resources.NewMetaResource(
+		b.Name, namespace, testlib.BrokerTypeMeta,
+	)
+	err := duck.WaitForResourceReady(ctx.Client.Dynamic, meta)
+	if err != nil {
+		ctx.T.Fatal(err)
+	}
+	broker, err := ctx.Client.Eventing.EventingV1().Brokers(namespace).Get(
+		ctx.Ctx, b.Name, metav1.GetOptions{},
+	)
+	if err != nil {
+		ctx.T.Fatal(err)
+	}
+	url := broker.Status.Address.URL
+	ctx.Log.Debugf("\"%s\" broker URL for ns %s is %v",
+		b.Name, namespace, url)
+	return url
+}
+
+func (b *BrokerAndTriggers) deployTriggers(ctx Context, dest duckv1.Destination) {
+	triggers := make([]*eventingv1.Trigger, 0, len(b.Triggers.Types))
+	for _, eventType := range b.Triggers.Types {
+		name := fmt.Sprintf("%s-%s", b.Name, eventType)
+		subscriberOption := resources.WithSubscriberDestination(func(t *eventingv1.Trigger) duckv1.Destination {
+			return dest
+		})
+		ctx.Log.Debugf("Creating trigger \"%s\" for type %s to route to %#v",
+			name, eventType, dest)
+		trgr := ctx.Client.CreateTriggerOrFail(
+			name,
+			resources.WithBroker(b.Name),
+			resources.WithAttributesTriggerFilter(
+				eventingv1.TriggerAnyFilter,
+				eventType,
+				map[string]interface{}{},
+			),
+			subscriberOption,
+		)
+		triggers = append(triggers, trgr)
+	}
+	for _, trgr := range triggers {
+		meta := resources.NewMetaResource(
+			trgr.Name, trgr.Namespace, testlib.TriggerTypeMeta,
+		)
+		err := duck.WaitForResourceReady(ctx.Client.Dynamic, meta)
+		if err != nil {
+			ctx.T.Fatal(err)
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sut/default.go b/vendor/knative.dev/eventing/test/upgrade/prober/sut/default.go
new file mode 100644
index 000000000..1f11edbdd
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sut/default.go
@@ -0,0 +1,22 @@
+/*
+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 sut
+
+// NewDefault will return a default SUT for this repo.
+func NewDefault() SystemUnderTest {
+	return NewBrokerAndTriggers()
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/sut/types.go b/vendor/knative.dev/eventing/test/upgrade/prober/sut/types.go
new file mode 100644
index 000000000..8f1341019
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/sut/types.go
@@ -0,0 +1,54 @@
+/*
+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 sut
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	testlib "knative.dev/eventing/test/lib"
+	watholaevent "knative.dev/eventing/test/upgrade/prober/wathola/event"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var eventTypes = []string{
+	watholaevent.Step{}.Type(),
+	watholaevent.Finished{}.Type(),
+}
+
+// SystemUnderTest (SUT) represents a system that we'd like to test with
+// continual prober.
+type SystemUnderTest interface {
+	// Deploy is responsible for deploying SUT and returning a URL to feed
+	// events into.
+	Deploy(ctx Context, destination duckv1.Destination) interface{}
+}
+
+// HasTeardown indicates that SystemUnderTest supports custom teardown that
+// exceeds regular teardown via usage of testlib.Tracker.
+type HasTeardown interface {
+	// Teardown will remove all deployed SUT resources.
+	Teardown(ctx Context)
+}
+
+// Context represents a context of system under test that we'd
+// like to deploy and teardown.
+type Context struct {
+	Ctx context.Context
+	Log *zap.SugaredLogger
+	*testlib.Client
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
index 1687753a0..90ce927a6 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/verify.go
@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/wavesoftware/go-ensure"
 	"go.uber.org/zap"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -42,8 +41,8 @@ const (
 )
 
 // Verify will verify prober state after finished has been sent.
-func (p *prober) Verify(ctx context.Context) (eventErrs []error, eventsSent int, fetchErr error) {
-	report := p.fetchReport(ctx)
+func (p *prober) Verify() (eventErrs []error, eventsSent int) {
+	report := p.fetchReport()
 	availRate := 0.0
 	if report.TotalRequests != 0 {
 		availRate = float64(report.EventsSent*100) / float64(report.TotalRequests)
@@ -53,7 +52,7 @@ func (p *prober) Verify(ctx context.Context) (eventErrs []error, eventsSent int,
 	p.log.Infof("Availability: %.3f%%, Requests sent: %d.",
 		availRate, report.TotalRequests)
 	if report.State == "active" {
-		return nil, report.EventsSent, errors.New("report fetched too early, receiver is in active state")
+		p.client.T.Fatal("report fetched too early, receiver is in active state")
 	}
 	for _, t := range report.Thrown.Missing {
 		eventErrs = append(eventErrs, errors.New(t))
@@ -71,16 +70,16 @@ func (p *prober) Verify(ctx context.Context) (eventErrs []error, eventsSent int,
 			eventErrs = append(eventErrs, errors.New(t))
 		}
 	}
-	return eventErrs, report.EventsSent, nil
+	return eventErrs, report.EventsSent
 }
 
 // Finish terminates sender which sends finished event.
-func (p *prober) Finish(ctx context.Context) {
-	p.removeSender(ctx)
+func (p *prober) Finish() {
+	p.removeSender()
 }
 
-func (p *prober) fetchReport(ctx context.Context) *receiver.Report {
-	exec := p.fetchExecution(ctx)
+func (p *prober) fetchReport() *receiver.Report {
+	exec := p.fetchExecution()
 	replayLogs(p.log, exec)
 	return exec.Report
 }
@@ -100,14 +99,14 @@ func replayLogs(log *zap.SugaredLogger, exec *fetcher.Execution) {
 	}
 }
 
-func (p *prober) fetchExecution(ctx context.Context) *fetcher.Execution {
-	ns := p.config.Namespace
-	job := p.deployFetcher(ctx)
-	defer p.deleteFetcher(ctx)
-	pod, err := p.findSucceededPod(ctx, job)
-	ensure.NoError(err)
-	bytes, err := pkgTest.PodLogs(ctx, p.client.Kube, pod.Name, fetcherName, ns)
-	ensure.NoError(err)
+func (p *prober) fetchExecution() *fetcher.Execution {
+	ns := p.client.Namespace
+	job := p.deployFetcher()
+	defer p.deleteFetcher()
+	pod, err := p.findSucceededPod(job)
+	p.ensureNoError(err)
+	bytes, err := pkgTest.PodLogs(p.config.Ctx, p.client.Kube, pod.Name, fetcherName, ns)
+	p.ensureNoError(err)
 	ex := &fetcher.Execution{
 		Logs: []fetcher.LogEntry{},
 		Report: &receiver.Report{
@@ -123,13 +122,13 @@ func (p *prober) fetchExecution(ctx context.Context) *fetcher.Execution {
 		},
 	}
 	err = json.Unmarshal(bytes, ex)
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	return ex
 }
 
-func (p *prober) deployFetcher(ctx context.Context) *batchv1.Job {
+func (p *prober) deployFetcher() *batchv1.Job {
 	p.log.Info("Deploying fetcher job: ", fetcherName)
-	jobs := p.client.Kube.BatchV1().Jobs(p.config.Namespace)
+	jobs := p.client.Kube.BatchV1().Jobs(p.client.Namespace)
 	var replicas int32 = 1
 	fetcherJob := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
@@ -170,28 +169,28 @@ func (p *prober) deployFetcher(ctx context.Context) *batchv1.Job {
 			},
 		},
 	}
-	created, err := jobs.Create(ctx, fetcherJob, metav1.CreateOptions{})
-	ensure.NoError(err)
+	created, err := jobs.Create(p.config.Ctx, fetcherJob, metav1.CreateOptions{})
+	p.ensureNoError(err)
 	p.log.Info("Waiting for fetcher job to succeed: ", fetcherName)
-	err = waitForJobToComplete(ctx, p.client.Kube, fetcherName, p.config.Namespace)
-	ensure.NoError(err)
+	err = waitForJobToComplete(p.config.Ctx, p.client.Kube, fetcherName, p.client.Namespace)
+	p.ensureNoError(err)
 
 	return created
 }
 
-func (p *prober) deleteFetcher(ctx context.Context) {
-	ns := p.config.Namespace
+func (p *prober) deleteFetcher() {
+	ns := p.client.Namespace
 	jobs := p.client.Kube.BatchV1().Jobs(ns)
-	err := jobs.Delete(ctx, fetcherName, metav1.DeleteOptions{})
-	ensure.NoError(err)
+	err := jobs.Delete(p.config.Ctx, fetcherName, metav1.DeleteOptions{})
+	p.ensureNoError(err)
 }
 
-func (p *prober) findSucceededPod(ctx context.Context, job *batchv1.Job) (*corev1.Pod, error) {
+func (p *prober) findSucceededPod(job *batchv1.Job) (*corev1.Pod, error) {
 	pods := p.client.Kube.CoreV1().Pods(job.Namespace)
-	podList, err := pods.List(ctx, metav1.ListOptions{
+	podList, err := pods.List(p.config.Ctx, metav1.ListOptions{
 		LabelSelector: fmt.Sprint("job-name=", job.Name),
 	})
-	ensure.NoError(err)
+	p.ensureNoError(err)
 	for _, pod := range podList.Items {
 		if pod.Status.Phase == corev1.PodSucceeded {
 			return &pod, nil
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go
index e25f0a92d..77f300a61 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/defaults.go
@@ -62,7 +62,7 @@ func defaultValues() *Config {
 				Duration: time.Second,
 			},
 			Errors: ReceiverErrorConfig{
-				UnavailablePeriodToReport: 5 * time.Second,
+				UnavailablePeriodToReport: time.Minute,
 			},
 		},
 		Forwarder: ForwarderConfig{
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go
index f6559dab5..2a7fe264b 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/event/types.go
@@ -23,9 +23,9 @@ import (
 
 const (
 	// StepType is a string type representation of step event
-	StepType = "com.github.cardil.wathola.step"
+	StepType = "dev.knative.eventing.wathola.step"
 	// FinishedType os a string type representation of finished event
-	FinishedType = "com.github.cardil.wathola.finished"
+	FinishedType = "dev.knative.eventing.wathola.finished"
 )
 
 // Step is a event call at each step of verification