Repository: knative-sandbox/eventing-kafka
Pull request #514 (2021-04-07 14:53:12)

[consolidated] Rolling deployment dispatcher on configmap change
Instead of doing more complex things (making the pods commit suicide, on-the-fly reloading, shaving horde of yaks etc) within the dispatcher to reload the configuration, just watch the configmap in the controller and do a rolling update on the dispatcher deployment.

Hashes the configmap, stores it on the dispatcher deployment as an annotation. When configmap is changed (watching it in the controller), update it on the deployment annotation. That will make Kubernetes to do a rolling deployment.

I tested my changes with https://github.com/aliok/kafka-ch-reload-tests. There are 2 pods:
- sender sends messages for 2 mins
- receiver receives messages for 3 mins

With a message frequency of 10 ms, there were only 4 times the sender was unable to send messages to the channel during the rolling deployment.

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Rolling deployment dispatcher on configmap change
- Alternative approach to https://github.com/knative-sandbox/eventing-kafka/pull/499, as discussed in that ticket

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- 🧽  Consolidated Kafka Channel dispatcher is now reconfigured when `config-kafka` configmap is updated

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

<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"]} -->
knative-prow-robot: @aliok: 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>
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka%2Fpull%2F514) for more info**.

<!-- need_author_consent -->
knative-prow-robot: @aliok: 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 | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-build-tests/1379809499245711360) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-build-tests/1379809499245711360) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1379809499107299328) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-build-tests/1379809499245711360) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1379809499107299328) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1379809499216351232) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-build-tests/1379809499245711360) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1379809499107299328) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1379809499216351232) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-unit-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-unit-tests/1379809499266682880) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-build-tests/1379809499245711360) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1379809499107299328) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1379809499216351232) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-unit-tests | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-unit-tests/1379809499266682880) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-go-coverage/1379809499321208832) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<details>

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

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

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aliok: /ok-to-test
knative-prow-robot: @aliok: 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-sasl | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1379809499166019584) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1379809499191185408) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1379809499107299328) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1379809499216351232) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | c7077aae9466169305a84b8fb4e38e310f29abe6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1379809499132465152) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-unit-tests | 40e08a5263008068f973059c66b5818224d05789 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-unit-tests/1379812627919671296) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-go-coverage/1379812627944837120/artifacts/line-cov.html#file1) | 55.3% | 54.5% | -0.8

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 40e08a5263008068f973059c66b5818224d05789 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-unit-tests/1379812627919671296) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 40e08a5263008068f973059c66b5818224d05789 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-upgrade-tests/1379812627869339648) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/514?src=pr&el=h1) Report
> Merging [#514](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/514?src=pr&el=desc) (41cb8cd) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/6b91f179e5ec2ff3690df26cc3aef2b4a7c860ae?el=desc) (6b91f17) will **decrease** coverage by `0.11%`.
> The diff coverage is `31.25%`.

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

```diff
@@            Coverage Diff             @@
##             main     #514      +/-   ##
==========================================
- Coverage   72.93%   72.82%   -0.12%     
==========================================
  Files         146      146              
  Lines        5813     5829      +16     
==========================================
+ Hits         4240     4245       +5     
- Misses       1327     1336       +9     
- Partials      246      248       +2     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/514?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...consolidated/reconciler/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/514/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9rYWZrYWNoYW5uZWwuZ28=) | `46.51% <15.38%> (-1.41%)` | :arrow_down: |
| [...ated/reconciler/controller/resources/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/514/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9yZXNvdXJjZXMvZGlzcGF0Y2hlci5nbw==) | `100.00% <100.00%> (ø)` | |

------

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-go-coverage/1380092689613066240/artifacts/line-cov.html#file0) | 55.3% | 54.2% | -1.1

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 41cb8cd89b7ea4da6c998722680dfa5494511ac5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1380092689533374464) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aliok: /test pull-knative-sandbox-eventing-kafka-integration-test-mt-source
aliok: The same test I did on https://github.com/knative-sandbox/eventing-kafka/pull/499 reports following with this PR:

```
sender
-------
In 120 seconds, tried to send 11815 messages
Success:11811
Errors:4

receiver
--------
Total received message count: 11935
124 messages received twice
```

This is closest to 0 downtime w/o adding messages into a queue or something.
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 41cb8cd89b7ea4da6c998722680dfa5494511ac5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1380102155477716992) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-go-coverage/1380113568388616192/artifacts/line-cov.html#file0) | 55.3% | 54.2% | -1.1

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 01e5c01d94f7c980cd5d706497a79cc90770d0e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1380113565926559744) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
matzew: when I change the name of my TLS secret on the configmap. does it than also go ahead and look for the new Secret, to start working with those values? 
aliok: > when I change the name of my TLS secret on the configmap. does it than also go ahead and look for the new Secret, to start working with those values?

It will.

Watching the **content** of the TLS secret is not done yet. https://github.com/knative-sandbox/eventing-kafka/issues/363 --> covers that
aliok: I don't understand why pull-knative-sandbox-eventing-kafka-integration-test-mt-source fails :(
Created a test PR to see if it fails on main too : https://github.com/knative-sandbox/eventing-kafka/pull/519
matzew: would the control-protocol help here too?
aliok: > would the control-protocol help here too?

It might, but not all config can be reloaded on-the-fly. So, gotta kill the pod.
lionelvillard: /test pull-knative-sandbox-eventing-kafka-integration-test-mt-source
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 01e5c01d94f7c980cd5d706497a79cc90770d0e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1380142919733743616) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
slinkydeveloper: > would the control-protocol help here too?

It will

> It might, but not all config can be reloaded on-the-fly. So, gotta kill the pod.

With the encrypted version of the protocol, in theory we can also send secrets through it. 
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-go-coverage/1380433680874344448/artifacts/line-cov.html#file0) | 55.3% | 54.2% | -1.1

devguyio: oh, no one assigned the PR lol 
/assign
aliok: > I left a minor comment, however I think maybe it's good to add test cases for the kafkachannel_test that verify the dispatcher is updated when config is changed. I know I came late to the review party but if this is not super urgent, I'm hoping we increase our testing coverage specially around the reconciliation loop logic validation. What do you think?

I looked around about how to add proper e2e tests for this but couldn't find anything. About, unit tests for checking if the hash is changed --> yes, I forgot that actually.
aliok: Added unit tests
@devguyio mind having a look again?
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/514/pull-knative-sandbox-eventing-kafka-go-coverage/1380455248312668160/artifacts/line-cov.html#file0) | 55.3% | 55.3% | -0.0


Patch:
diff --git a/config/channel/consolidated/deployments/dispatcher.yaml b/config/channel/consolidated/deployments/dispatcher.yaml
index c168904a5..90c462cfc 100644
--- a/config/channel/consolidated/deployments/dispatcher.yaml
+++ b/config/channel/consolidated/deployments/dispatcher.yaml
@@ -29,6 +29,10 @@ spec:
       messaging.knative.dev/role: dispatcher
   template:
     metadata:
+      annotations:
+        # This annotation is used by the controller to track updates
+        # to config-kafka and apply them in the dispatcher deployment
+        kafka.eventing.knative.dev/configmap-hash: ''
       labels:
         # Do not change. Used by the controller for probing.
         messaging.knative.dev/channel: kafka-channel
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index efb6c036b..16639c42d 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"hash/crc32"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -116,10 +117,11 @@ type Reconciler struct {
 	dispatcherImage          string
 	dispatcherServiceAccount string
 
-	kafkaConfig      *utils.KafkaConfig
-	kafkaAuthConfig  *client.KafkaAuthConfig
-	kafkaConfigError error
-	kafkaClientSet   kafkaclientset.Interface
+	kafkaConfig        *utils.KafkaConfig
+	kafkaConfigMapHash string
+	kafkaAuthConfig    *client.KafkaAuthConfig
+	kafkaConfigError   error
+	kafkaClientSet     kafkaclientset.Interface
 	// Using a shared kafkaClusterAdmin does not work currently because of an issue with
 	// Shopify/sarama, see https://github.com/Shopify/sarama/issues/1162.
 	kafkaClusterAdmin    sarama.ClusterAdmin
@@ -310,6 +312,7 @@ func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope string, disp
 		Image:               r.dispatcherImage,
 		Replicas:            1,
 		ServiceAccount:      r.dispatcherServiceAccount,
+		ConfigMapHash:       r.kafkaConfigMapHash,
 	}
 
 	expected := resources.MakeDispatcher(args)
@@ -350,6 +353,8 @@ func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope string, disp
 			return nil, fmt.Errorf("container %s does not exist in expected dispatcher deployment. Cannot check if the deployment needs an update", resources.DispatcherContainerName)
 		}
 
+		expectedConfigMapHash := r.kafkaConfigMapHash
+
 		needsUpdate := false
 
 		if existing.Image != expectedContainer.Image {
@@ -364,6 +369,20 @@ func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope string, disp
 			needsUpdate = true
 		}
 
+		if d.Spec.Template.Annotations == nil {
+			logging.FromContext(ctx).Infof("Configmap hash is not set. Updating the dispatcher deployment.")
+			d.Spec.Template.Annotations = map[string]string{
+				resources.ConfigMapHashAnnotationKey: expectedConfigMapHash,
+			}
+			needsUpdate = true
+		}
+
+		if d.Spec.Template.Annotations[resources.ConfigMapHashAnnotationKey] != expectedConfigMapHash {
+			logging.FromContext(ctx).Infof("Configmap hash is changed. Updating the dispatcher deployment.")
+			d.Spec.Template.Annotations[resources.ConfigMapHashAnnotationKey] = expectedConfigMapHash
+			needsUpdate = true
+		}
+
 		if needsUpdate {
 			d, err := r.KubeClientSet.AppsV1().Deployments(dispatcherNamespace).Update(ctx, d, metav1.UpdateOptions{})
 			if err == nil {
@@ -563,6 +582,7 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	// Eventually the previous config should be snapshotted to delete Kafka topics
 	r.kafkaConfig = kafkaConfig
 	r.kafkaConfigError = err
+	r.kafkaConfigMapHash = configmapDataCheckSum(configMap)
 
 	if err != nil {
 		logger.Errorw("Error creating AdminClient", zap.Error(err))
@@ -592,3 +612,12 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, kc *v1beta1.KafkaChannel)
 	}
 	return newReconciledNormal(kc.Namespace, kc.Name) //ok to remove finalizer
 }
+
+func configmapDataCheckSum(configMap *corev1.ConfigMap) string {
+	if configMap == nil || configMap.Data == nil {
+		return ""
+	}
+	configMapDataStr := fmt.Sprintf("%v", configMap.Data)
+	checksum := fmt.Sprintf("%08x", crc32.ChecksumIEEE([]byte(configMapDataStr)))
+	return checksum
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
index ca06a04e8..bf885c869 100644
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
+++ b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
@@ -24,7 +24,10 @@ import (
 	"knative.dev/pkg/system"
 )
 
-const DispatcherContainerName = "dispatcher"
+const (
+	DispatcherContainerName    = "dispatcher"
+	ConfigMapHashAnnotationKey = "kafka.eventing.knative.dev/configmap-hash"
+)
 
 var (
 	dispatcherName   = "kafka-ch-dispatcher"
@@ -40,6 +43,7 @@ type DispatcherArgs struct {
 	Image               string
 	Replicas            int32
 	ServiceAccount      string
+	ConfigMapHash       string
 }
 
 // MakeDispatcher generates the dispatcher deployment for the KafKa channel
@@ -63,6 +67,9 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: dispatcherLabels,
+					Annotations: map[string]string{
+						ConfigMapHashAnnotationKey: args.ConfigMapHash,
+					},
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName: args.ServiceAccount,
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index fa6bfc13b..f205e6fcb 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -134,7 +134,7 @@ func (a *Adapter) Start(ctx context.Context) (err error) {
 		options...,
 	)
 	if err != nil {
-		panic(err)
+		return fmt.Errorf("failed to start consumer group: %w", err)
 	}
 	defer func() {
 		err := group.Close()


Test patch:
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
index a86f1da50..7a77831a9 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -52,6 +53,7 @@ import (
 const (
 	testNS                       = "test-namespace"
 	testDispatcherserviceAccount = "kafka-ch-dispatcher"
+	testConfigMapHash            = "deadbeef"
 	kcName                       = "test-kc"
 	testDispatcherImage          = "test-image"
 	channelServiceAddress        = "test-kc-kn-channel.test-namespace.svc.cluster.local"
@@ -331,6 +333,7 @@ func TestAllCases(t *testing.T) {
 			systemNamespace:          testNS,
 			dispatcherImage:          testDispatcherImage,
 			dispatcherServiceAccount: testDispatcherserviceAccount,
+			kafkaConfigMapHash:       testConfigMapHash,
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
@@ -395,6 +398,7 @@ func TestTopicExists(t *testing.T) {
 			systemNamespace:          testNS,
 			dispatcherImage:          testDispatcherImage,
 			dispatcherServiceAccount: testDispatcherserviceAccount,
+			kafkaConfigMapHash:       testConfigMapHash,
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
@@ -471,6 +475,7 @@ func TestDeploymentUpdatedOnImageChange(t *testing.T) {
 			systemNamespace:          testNS,
 			dispatcherImage:          testDispatcherImage,
 			dispatcherServiceAccount: testDispatcherserviceAccount,
+			kafkaConfigMapHash:       testConfigMapHash,
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
@@ -547,6 +552,7 @@ func TestDeploymentZeroReplicas(t *testing.T) {
 			systemNamespace:          testNS,
 			dispatcherImage:          testDispatcherImage,
 			dispatcherServiceAccount: testDispatcherserviceAccount,
+			kafkaConfigMapHash:       testConfigMapHash,
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
@@ -620,6 +626,84 @@ func TestDeploymentMoreThanOneReplicas(t *testing.T) {
 			systemNamespace:          testNS,
 			dispatcherImage:          testDispatcherImage,
 			dispatcherServiceAccount: testDispatcherserviceAccount,
+			kafkaConfigMapHash:       testConfigMapHash,
+			kafkaConfig: &KafkaConfig{
+				Brokers: []string{brokerName},
+			},
+			kafkachannelLister: listers.GetKafkaChannelLister(),
+			// TODO fix
+			kafkachannelInformer: nil,
+			deploymentLister:     listers.GetDeploymentLister(),
+			serviceLister:        listers.GetServiceLister(),
+			endpointsLister:      listers.GetEndpointsLister(),
+			kafkaClusterAdmin: &mockClusterAdmin{
+				mockCreateTopicFunc: func(topic string, detail *sarama.TopicDetail, validateOnly bool) error {
+					errMsg := sarama.ErrTopicAlreadyExists.Error()
+					return &sarama.TopicError{
+						Err:    sarama.ErrTopicAlreadyExists,
+						ErrMsg: &errMsg,
+					}
+				},
+			},
+			kafkaClientSet:    fakekafkaclient.Get(ctx),
+			KubeClientSet:     kubeclient.Get(ctx),
+			EventingClientSet: eventingClient.Get(ctx),
+			statusManager: &fakeStatusManager{
+				FakeIsReady: func(ctx context.Context, channel v1beta1.KafkaChannel,
+					spec eventingduckv1.SubscriberSpec) (bool, error) {
+					return true, nil
+				},
+			},
+		}
+		return kafkachannel.NewReconciler(ctx, logging.FromContext(ctx), r.kafkaClientSet, listers.GetKafkaChannelLister(), controller.GetEventRecorder(ctx), r)
+	}, zap.L()))
+}
+
+func TestDeploymentUpdatedOnConfigMapHashChange(t *testing.T) {
+	kcKey := testNS + "/" + kcName
+	row := TableRow{
+		Name: "ConfigMap hash changed, dispatcher updated",
+		Key:  kcKey,
+		Objects: []runtime.Object{
+			makeDeploymentWithConfigMapHash("toBeUpdated"),
+			makeService(),
+			makeReadyEndpoints(),
+			reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithKafkaFinalizer(finalizerName)),
+		},
+		WantErr: false,
+		WantCreates: []runtime.Object{
+			makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
+		},
+		WantUpdates: []clientgotesting.UpdateActionImpl{{
+			Object: makeDeploymentWithConfigMapHash(testConfigMapHash),
+		}},
+		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+			Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithInitKafkaChannelConditions,
+				reconcilertesting.WithKafkaFinalizer(finalizerName),
+				reconcilertesting.WithKafkaChannelConfigReady(),
+				reconcilertesting.WithKafkaChannelTopicReady(),
+				//				reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
+				reconcilertesting.WithKafkaChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelEndpointsReady(),
+				reconcilertesting.WithKafkaChannelChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
+			),
+		}},
+		WantEvents: []string{
+			Eventf(corev1.EventTypeNormal, dispatcherDeploymentUpdated, "Dispatcher deployment updated"),
+			Eventf(corev1.EventTypeNormal, "KafkaChannelReconciled", `KafkaChannel reconciled: "test-namespace/test-kc"`),
+		},
+	}
+
+	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+
+		r := &Reconciler{
+			systemNamespace:          testNS,
+			dispatcherImage:          testDispatcherImage,
+			dispatcherServiceAccount: testDispatcherserviceAccount,
+			kafkaConfigMapHash:       testConfigMapHash,
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
@@ -750,15 +834,24 @@ func (ca *mockClusterAdmin) DeleteConsumerGroup(group string) error {
 
 var _ sarama.ClusterAdmin = (*mockClusterAdmin)(nil)
 
-func makeDeploymentWithImageAndReplicas(image string, replicas int32) *appsv1.Deployment {
+func makeDeploymentWithParams(image string, replicas int32, configMapHash string) *appsv1.Deployment {
 	return resources.MakeDispatcher(resources.DispatcherArgs{
 		DispatcherNamespace: testNS,
 		Image:               image,
 		Replicas:            replicas,
 		ServiceAccount:      testDispatcherserviceAccount,
+		ConfigMapHash:       configMapHash,
 	})
 }
 
+func makeDeploymentWithImageAndReplicas(image string, replicas int32) *appsv1.Deployment {
+	return makeDeploymentWithParams(image, replicas, testConfigMapHash)
+}
+
+func makeDeploymentWithConfigMapHash(configMapHash string) *appsv1.Deployment {
+	return makeDeploymentWithParams(testDispatcherImage, 1, configMapHash)
+}
+
 func makeDeployment() *appsv1.Deployment {
 	return makeDeploymentWithImageAndReplicas(testDispatcherImage, 1)
 }
@@ -884,3 +977,36 @@ func makePatch(namespace, name, patch string) clientgotesting.PatchActionImpl {
 		Patch: []byte(patch),
 	}
 }
+
+func TestConfigmapDataCheckSum(t *testing.T) {
+	cases := []struct {
+		name      string
+		configmap *corev1.ConfigMap
+		expected  string
+	}{{
+		name:      "nil configmap",
+		configmap: nil,
+		expected:  "",
+	}, {
+		name: "nil configmap data",
+		configmap: &corev1.ConfigMap{
+			Data: nil,
+		},
+		expected: "",
+	}, {
+		name: "with configmap data",
+		configmap: &corev1.ConfigMap{
+			Data: map[string]string{"foo": "bar"},
+		},
+		expected: "f39c9878", // precomputed manually
+	}}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			retrieved := configmapDataCheckSum(tc.configmap)
+			if diff := cmp.Diff(tc.expected, retrieved); diff != "" {
+				t.Errorf("unexpected Config (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
index 8dfd667ed..8d5c170ef 100644
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
+++ b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
@@ -41,6 +41,7 @@ func TestNewDispatcher(t *testing.T) {
 		Image:               imageName,
 		Replicas:            1,
 		ServiceAccount:      serviceAccount,
+		ConfigMapHash:       testConfigMapHash,
 	}
 
 	replicas := int32(1)
@@ -61,6 +62,9 @@ func TestNewDispatcher(t *testing.T) {
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: dispatcherLabels,
+					Annotations: map[string]string{
+						ConfigMapHashAnnotationKey: testConfigMapHash,
+					},
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName: serviceAccount,
@@ -124,6 +128,7 @@ func TestNewNamespaceDispatcher(t *testing.T) {
 		Image:               imageName,
 		Replicas:            1,
 		ServiceAccount:      serviceAccount,
+		ConfigMapHash:       testConfigMapHash,
 	}
 
 	replicas := int32(1)
@@ -144,6 +149,9 @@ func TestNewNamespaceDispatcher(t *testing.T) {
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: dispatcherLabels,
+					Annotations: map[string]string{
+						ConfigMapHashAnnotationKey: testConfigMapHash,
+					},
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName: serviceAccount,
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/service_test.go b/pkg/channel/consolidated/reconciler/controller/resources/service_test.go
index 0d848c115..940329a16 100644
--- a/pkg/channel/consolidated/reconciler/controller/resources/service_test.go
+++ b/pkg/channel/consolidated/reconciler/controller/resources/service_test.go
@@ -33,6 +33,7 @@ const (
 	testNS             = "my-test-ns"
 	testDispatcherNS   = "dispatcher-namespace"
 	testDispatcherName = "dispatcher-name"
+	testConfigMapHash  = "deadbeef"
 )
 
 func TestMakeChannelServiceAddress(t *testing.T) {
diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
index 5b81670ca..2dfc7b382 100644
--- a/pkg/source/adapter/adapter_test.go
+++ b/pkg/source/adapter/adapter_test.go
@@ -28,7 +28,6 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/cloudevents/sdk-go/v2/types"
-	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"knative.dev/eventing/pkg/adapter/v2"
 	"knative.dev/pkg/source"
@@ -441,9 +440,9 @@ func TestAdapter_Start(t *testing.T) { // just increase code coverage
 	_ = os.Setenv("KAFKA_BOOTSTRAP_SERVERS", "my-cluster-kafka-bootstrap.my-kafka-namespace:9092")
 
 	a := NewAdapter(ctx, NewEnvConfig(), nil, nil)
-	require.Panics(t, func() {
-		_ = a.Start(ctx)
-	})
-
+	err := a.Start(ctx)
+	if err == nil {
+		t.Errorf("expected error, but got nil")
+	}
 	cancel()
 }
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index 492653c5a..9a7cd3f9e 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -33,6 +33,7 @@ import (
 	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
 	"knative.dev/eventing/pkg/kncloudevents"
 
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 )
 
@@ -81,6 +82,28 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Error(`Expected adapter to contain "test-ns/test-name"`)
 	}
 
+	adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-name-badBroker",
+			Namespace: "test-ns",
+		},
+		Spec: sourcesv1beta1.KafkaSourceSpec{
+			KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+				BootstrapServers: []string{"my-cluster-kafka-bootstrap-badBroker.kafka.svc:9092"},
+			},
+		},
+		Status: sourcesv1beta1.KafkaSourceStatus{
+			Placeable: duckv1alpha1.Placeable{
+				Placement: []duckv1alpha1.Placement{
+					{PodName: podName, VReplicas: int32(1)},
+				}},
+		},
+	})
+
+	if _, ok := adapter.sources["test-ns/test-name-badBroker"]; !ok {
+		t.Error(`Expected adapter to contain "test-ns/test-name-badBroker"`)
+	}
+
 	select {
 	case a := <-runningAdapterChan:
 		if !a.running {
@@ -103,6 +126,19 @@ func TestUpdateRemoveSources(t *testing.T) {
 		t.Error(`Expected adapter to not contain "test-ns/test-name"`)
 	}
 
+	adapter.Remove(&sourcesv1beta1.KafkaSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-name-badBroker",
+			Namespace: "test-ns",
+		},
+		Spec:   sourcesv1beta1.KafkaSourceSpec{},
+		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 {
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 9c1663613..b8fe36d03 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -315,12 +315,40 @@ function install_consolidated_channel_crds {
   wait_until_pods_running "${SYSTEM_NAMESPACE}"
 }
 
+function install_released_consolidated_source {
+  install_consolidated_sources_crds latest-release
+}
+
+function install_head_consolidated_source {
+  install_consolidated_sources_crds HEAD
+}
+
 function install_consolidated_sources_crds() {
-  echo "Installing consolidated Kafka Source CRD"
-  rm -rf "${KAFKA_SOURCE_CRD_CONFIG_DIR}" && mkdir -p "${KAFKA_SOURCE_CRD_CONFIG_DIR}"
-  cp "${KAFKA_SOURCE_TEMPLATE_DIR}/"*yaml "${KAFKA_SOURCE_CRD_CONFIG_DIR}"
-  sed -i "s/namespace: knative-eventing/namespace: ${SYSTEM_NAMESPACE}/g" "${KAFKA_SOURCE_CRD_CONFIG_DIR}/"*yaml
-  ko apply -f "${KAFKA_SOURCE_CRD_CONFIG_DIR}" || return 1
+  local source url ver release_yaml
+  source="${1:-HEAD}"
+  if [[ "${source}" == 'HEAD' ]]; then
+    echo "Installing consolidated Kafka Source CRD (from HEAD)"
+    rm -rf "${KAFKA_SOURCE_CRD_CONFIG_DIR}" && mkdir -p "${KAFKA_SOURCE_CRD_CONFIG_DIR}"
+    cp "${KAFKA_SOURCE_TEMPLATE_DIR}/"*yaml "${KAFKA_SOURCE_CRD_CONFIG_DIR}"
+    sed -i "s/namespace: knative-eventing/namespace: ${SYSTEM_NAMESPACE}/g" "${KAFKA_SOURCE_CRD_CONFIG_DIR}/"*yaml
+    ko apply -f "${KAFKA_SOURCE_CRD_CONFIG_DIR}" || return $?
+  elif [[ "${source}" == 'latest-release' ]]; then
+    ver="${LATEST_RELEASE_VERSION}"
+    echo "Installing consolidated Kafka Source CRD (from latest release: ${ver})"
+    # Download the latest release of Knative Eventing Kafka.
+    url="${EVENTING_KAFKA_REPO}/releases/download/${ver}/source.yaml"
+    release_yaml="${ARTIFACTS}/source-${ver}.yaml"
+
+    curl -Lo "${release_yaml}" "${url}"
+    sed -i "s/namespace: knative-eventing/namespace: ${SYSTEM_NAMESPACE}/g" \
+      "${release_yaml}"
+    echo "Applying: ${release_yaml}"
+    kubectl apply -f "${release_yaml}"
+  else
+    fail_test "Unsupported source of installation: ${source}"
+    return 56
+  fi
+
   run_postinstall_jobs
   wait_until_pods_running "${EVENTING_NAMESPACE}" || fail_test "Failed to install the consolidated Kafka Source CRD"
 }
diff --git a/test/e2e/kafka_source.go b/test/e2e/kafka_source.go
new file mode 100644
index 000000000..1d6d16914
--- /dev/null
+++ b/test/e2e/kafka_source.go
@@ -0,0 +1,439 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	. "github.com/cloudevents/sdk-go/v2/test"
+	cetypes "github.com/cloudevents/sdk-go/v2/types"
+	"github.com/google/uuid"
+	corev1 "k8s.io/api/core/v1"
+
+	"knative.dev/eventing/pkg/utils"
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/resources"
+
+	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/apis/sources/v1alpha1"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/eventing-kafka/test/e2e/helpers"
+	contribtestlib "knative.dev/eventing-kafka/test/lib"
+	contribresources "knative.dev/eventing-kafka/test/lib/resources"
+)
+
+const (
+	kafkaBootstrapUrlPlain = "my-cluster-kafka-bootstrap.kafka.svc:9092"
+	kafkaBootstrapUrlTLS   = "my-cluster-kafka-bootstrap.kafka.svc:9093"
+	kafkaBootstrapUrlSASL  = "my-cluster-kafka-bootstrap.kafka.svc:9094"
+	kafkaClusterName       = "my-cluster"
+	kafkaClusterNamespace  = "kafka"
+
+	kafkaSASLSecret = "strimzi-sasl-secret"
+	kafkaTLSSecret  = "strimzi-tls-secret"
+)
+
+var (
+	testMtSource = os.Getenv("TEST_MT_SOURCE")
+)
+
+// SourceTestScope returns true if we should proceed with given
+// test case.
+type SourceTestScope func(auth, testCase, version string) bool
+
+// AssureKafkaSourceIsOperational assures that KafkaSource works as intended.
+func AssureKafkaSourceIsOperational(t *testing.T, scope SourceTestScope) {
+	eventTime, _ := cetypes.ParseTime("2018-04-05T17:31:00Z")
+
+	auths := map[string]struct {
+		auth authSetup
+	}{
+		"plain": {
+			auth: authSetup{
+				bootStrapServer: kafkaBootstrapUrlPlain,
+				SASLEnabled:     false,
+				TLSEnabled:      false,
+			},
+		},
+		"s512": {
+			auth: authSetup{
+				bootStrapServer: kafkaBootstrapUrlSASL,
+				SASLEnabled:     true,
+				TLSEnabled:      false,
+			},
+		},
+		"tls": {
+			auth: authSetup{
+				bootStrapServer: kafkaBootstrapUrlTLS,
+				SASLEnabled:     false,
+				TLSEnabled:      true,
+			},
+		},
+	}
+	tests := map[string]struct {
+		messageKey     string
+		messageHeaders map[string]string
+		messagePayload string
+		matcherGen     func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher
+	}{
+		"no_event": {
+			messageKey: "0",
+			messageHeaders: map[string]string{
+				"content-type": "application/json",
+			},
+			messagePayload: `{"value":5}`,
+			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
+				return AllOf(
+					HasSource(cloudEventsSourceName),
+					HasType(cloudEventsEventType),
+					HasDataContentType("application/json"),
+					HasData([]byte(`{"value":5}`)),
+					HasExtension("key", "0"),
+				)
+			},
+		},
+		"no_event_no_content_type": {
+			messageKey:     "0",
+			messagePayload: `{"value":5}`,
+			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
+				return AllOf(
+					HasSource(cloudEventsSourceName),
+					HasType(cloudEventsEventType),
+					HasData([]byte(`{"value":5}`)),
+					HasExtension("key", "0"),
+				)
+			},
+		},
+		"no_event_content_type_or_key": {
+			messagePayload: `{"value":5}`,
+			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
+				return AllOf(
+					HasSource(cloudEventsSourceName),
+					HasType(cloudEventsEventType),
+					HasData([]byte(`{"value":5}`)),
+				)
+			},
+		},
+		"no_event_with_text_plain_body": {
+			messageKey: "0",
+			messageHeaders: map[string]string{
+				"content-type": "text/plain",
+			},
+			messagePayload: "simple 10",
+			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
+				return AllOf(
+					HasSource(cloudEventsSourceName),
+					HasType(cloudEventsEventType),
+					HasDataContentType("text/plain"),
+					HasData([]byte("simple 10")),
+					HasExtension("key", "0"),
+				)
+			},
+		},
+		"structured": {
+			messageHeaders: map[string]string{
+				"content-type": "application/cloudevents+json",
+			},
+			messagePayload: mustJsonMarshal(t, map[string]interface{}{
+				"specversion":     "1.0",
+				"type":            "com.github.pull.create",
+				"source":          "https://github.com/cloudevents/spec/pull",
+				"subject":         "123",
+				"id":              "A234-1234-1234",
+				"time":            "2018-04-05T17:31:00Z",
+				"datacontenttype": "application/json",
+				"data": map[string]string{
+					"hello": "Francesco",
+				},
+				"comexampleextension1": "value",
+				"comexampleothervalue": 5,
+			}),
+			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
+				return AllOf(
+					HasSpecVersion(cloudevents.VersionV1),
+					HasType("com.github.pull.create"),
+					HasSource("https://github.com/cloudevents/spec/pull"),
+					HasSubject("123"),
+					HasId("A234-1234-1234"),
+					HasTime(eventTime),
+					HasDataContentType("application/json"),
+					HasData([]byte(`{"hello":"Francesco"}`)),
+					HasExtension("comexampleextension1", "value"),
+					HasExtension("comexampleothervalue", "5"),
+				)
+			},
+		},
+		"binary": {
+			messageHeaders: map[string]string{
+				"ce_specversion":          "1.0",
+				"ce_type":                 "com.github.pull.create",
+				"ce_source":               "https://github.com/cloudevents/spec/pull",
+				"ce_subject":              "123",
+				"ce_id":                   "A234-1234-1234",
+				"ce_time":                 "2018-04-05T17:31:00Z",
+				"content-type":            "application/json",
+				"ce_comexampleextension1": "value",
+				"ce_comexampleothervalue": "5",
+			},
+			messagePayload: mustJsonMarshal(t, map[string]string{
+				"hello": "Francesco",
+			}),
+			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
+				return AllOf(
+					HasSpecVersion(cloudevents.VersionV1),
+					HasType("com.github.pull.create"),
+					HasSource("https://github.com/cloudevents/spec/pull"),
+					HasSubject("123"),
+					HasId("A234-1234-1234"),
+					HasTime(eventTime),
+					HasDataContentType("application/json"),
+					HasData([]byte(`{"hello":"Francesco"}`)),
+					HasExtension("comexampleextension1", "value"),
+					HasExtension("comexampleothervalue", "5"),
+				)
+			},
+		},
+	}
+	for authName, auth := range auths {
+		for testcase, test := range tests {
+			test := test
+			name := testcase + "_" + authName
+			for _, version := range []string{"v1alpha1", "v1beta1"} {
+				testName := name + "-" + version
+				if !scope(authName, testcase, version) {
+					t.Log("Skipping the test case, because it's out of configured scope: ", testName)
+					continue
+				}
+				t.Run(testName, func(t *testing.T) {
+					testKafkaSource(t, name, version, test.messageKey, test.messageHeaders, test.messagePayload, test.matcherGen, auth.auth)
+				})
+			}
+		}
+	}
+}
+
+type authSetup struct {
+	bootStrapServer string
+	SASLEnabled     bool
+	TLSEnabled      bool
+}
+
+func withAuthEnablementV1Beta1(auth authSetup) contribresources.KafkaSourceV1Beta1Option {
+	// We test with sasl512 and enable tls with it, so check tls first
+	if auth.TLSEnabled {
+		return func(ks *sourcesv1beta1.KafkaSource) {
+			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
+			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaTLSSecret,
+				},
+				Key: "ca.crt",
+			}
+			ks.Spec.KafkaAuthSpec.Net.TLS.Cert.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaTLSSecret,
+				},
+				Key: "user.crt",
+			}
+			ks.Spec.KafkaAuthSpec.Net.TLS.Key.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaTLSSecret,
+				},
+				Key: "user.key",
+			}
+		}
+	}
+	if auth.SASLEnabled {
+		return func(ks *sourcesv1beta1.KafkaSource) {
+			ks.Spec.KafkaAuthSpec.Net.SASL.Enable = true
+			ks.Spec.KafkaAuthSpec.Net.SASL.User.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "user",
+			}
+			ks.Spec.KafkaAuthSpec.Net.SASL.Password.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "password",
+			}
+			ks.Spec.KafkaAuthSpec.Net.SASL.Type.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "saslType",
+			}
+			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
+			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "ca.crt",
+			}
+		}
+	}
+	return func(ks *sourcesv1beta1.KafkaSource) {}
+}
+
+func withAuthEnablementV1Alpha1(auth authSetup) contribresources.KafkaSourceV1Alpha1Option {
+	// We test with sasl512 and enable tls with it, so check tls first
+	if auth.TLSEnabled {
+		return func(ks *sourcesv1alpha1.KafkaSource) {
+			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
+			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaTLSSecret,
+				},
+				Key: "ca.crt",
+			}
+			ks.Spec.KafkaAuthSpec.Net.TLS.Cert.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaTLSSecret,
+				},
+				Key: "user.crt",
+			}
+			ks.Spec.KafkaAuthSpec.Net.TLS.Key.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaTLSSecret,
+				},
+				Key: "user.key",
+			}
+		}
+	}
+	if auth.SASLEnabled {
+		return func(ks *sourcesv1alpha1.KafkaSource) {
+			ks.Spec.KafkaAuthSpec.Net.SASL.Enable = true
+			ks.Spec.KafkaAuthSpec.Net.SASL.User.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "user",
+			}
+			ks.Spec.KafkaAuthSpec.Net.SASL.Password.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "password",
+			}
+			ks.Spec.KafkaAuthSpec.Net.SASL.Type.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "saslType",
+			}
+			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
+			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: kafkaSASLSecret,
+				},
+				Key: "ca.crt",
+			}
+		}
+	}
+	return func(ks *sourcesv1alpha1.KafkaSource) {}
+}
+func testKafkaSource(t *testing.T, name string, version string, messageKey string, messageHeaders map[string]string, messagePayload string, matcherGen func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher, auth authSetup) {
+	name = fmt.Sprintf("%s-%s", name, version)
+
+	var (
+		kafkaTopicName     = uuid.New().String()
+		consumerGroup      = uuid.New().String()
+		recordEventPodName = "e2e-kafka-r-" + strings.ReplaceAll(name, "_", "-")
+		kafkaSourceName    = "e2e-kafka-source-" + strings.ReplaceAll(name, "_", "-")
+	)
+
+	client := testlib.Setup(t, true)
+	defer testlib.TearDown(client)
+
+	if auth.SASLEnabled {
+		_, err := utils.CopySecret(client.Kube.CoreV1(), "knative-eventing", kafkaSASLSecret, client.Namespace, "default")
+		if err != nil {
+			t.Fatalf("could not copy SASL secret(%s): %v", kafkaSASLSecret, err)
+		}
+	}
+	if auth.TLSEnabled {
+		_, err := utils.CopySecret(client.Kube.CoreV1(), "knative-eventing", kafkaTLSSecret, client.Namespace, "default")
+		if err != nil {
+			t.Fatalf("could not copy secret(%s): %v", kafkaTLSSecret, err)
+		}
+	}
+	helpers.MustCreateTopic(client, kafkaClusterName, kafkaClusterNamespace, kafkaTopicName)
+	if len(recordEventPodName) > 63 {
+		recordEventPodName = recordEventPodName[:63]
+	}
+	eventTracker, _ := recordevents.StartEventRecordOrFail(context.Background(), client, recordEventPodName)
+
+	var (
+		cloudEventsSourceName string
+		cloudEventsEventType  string
+	)
+
+	t.Logf("Creating KafkaSource %s", version)
+	switch version {
+	case "v1alpha1":
+		contribtestlib.CreateKafkaSourceV1Alpha1OrFail(client, contribresources.KafkaSourceV1Alpha1(
+			auth.bootStrapServer,
+			kafkaTopicName,
+			resources.ServiceRef(recordEventPodName),
+			contribresources.WithNameV1Alpha1(kafkaSourceName),
+			contribresources.WithConsumerGroupV1Alpha1(consumerGroup),
+			withAuthEnablementV1Alpha1(auth),
+		))
+		cloudEventsSourceName = sourcesv1alpha1.KafkaEventSource(client.Namespace, kafkaSourceName, kafkaTopicName)
+		cloudEventsEventType = sourcesv1alpha1.KafkaEventType
+	case "v1beta1":
+		contribtestlib.CreateKafkaSourceV1Beta1OrFail(client, contribresources.KafkaSourceV1Beta1(
+			auth.bootStrapServer,
+			kafkaTopicName,
+			resources.ServiceRef(recordEventPodName),
+			contribresources.WithNameV1Beta1(kafkaSourceName),
+			contribresources.WithConsumerGroupV1Beta1(consumerGroup),
+			withAuthEnablementV1Beta1(auth),
+		))
+		cloudEventsSourceName = sourcesv1beta1.KafkaEventSource(client.Namespace, kafkaSourceName, kafkaTopicName)
+		cloudEventsEventType = sourcesv1beta1.KafkaEventType
+	default:
+		t.Fatalf("Unknown KafkaSource version %s", version)
+	}
+
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
+	if testMtSource == "1" {
+		time.Sleep(20 * time.Second)
+	}
+
+	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain, kafkaTopicName, messageKey, messageHeaders, messagePayload)
+
+	eventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(cloudEventsSourceName, cloudEventsEventType)))
+}
+
+func mustJsonMarshal(t *testing.T, val interface{}) string {
+	data, err := json.Marshal(val)
+	if err != nil {
+		t.Errorf("unexpected error, %v", err)
+	}
+	return string(data)
+}
diff --git a/test/e2e/kafka_source_test.go b/test/e2e/kafka_source_test.go
index fe6b3be91..4ad836ecd 100644
--- a/test/e2e/kafka_source_test.go
+++ b/test/e2e/kafka_source_test.go
@@ -1,5 +1,5 @@
-//+build e2e mtsource
-//+build source
+// +build e2e
+// +build source mtsource
 
 /*
 Copyright 2019 The Knative Authors
@@ -21,444 +21,26 @@ package e2e
 
 import (
 	"context"
-	"encoding/json"
-	"fmt"
-	"os"
-	"strings"
 	"testing"
 	"time"
 
-	cloudevents "github.com/cloudevents/sdk-go/v2"
 	. "github.com/cloudevents/sdk-go/v2/test"
-	cetypes "github.com/cloudevents/sdk-go/v2/types"
-	"github.com/google/uuid"
-	corev1 "k8s.io/api/core/v1"
-
-	"knative.dev/eventing/pkg/utils"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/recordevents"
-	"knative.dev/eventing/test/lib/resources"
-
-	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/apis/sources/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	"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/pkg/utils"
+	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/resources"
 )
 
-const (
-	kafkaBootstrapUrlPlain   = "my-cluster-kafka-bootstrap.kafka.svc:9092"
-	kafkaBootstrapUrlTLS     = "my-cluster-kafka-bootstrap.kafka.svc:9093"
-	kafkaBootstrapUrlSASL    = "my-cluster-kafka-bootstrap.kafka.svc:9094"
-	kafkaClusterName         = "my-cluster"
-	kafkaClusterNamespace    = "kafka"
-	kafkaClusterSASLUsername = "my-sasl-user"
-	kafkaClusterTLSUsername  = "my-tls-user"
-
-	kafkaSASLSecret = "strimzi-sasl-secret"
-	kafkaTLSSecret  = "strimzi-tls-secret"
-)
-
-type authSetup struct {
-	bootStrapServer string
-	SASLEnabled     bool
-	TLSEnabled      bool
-}
-
-var (
-	test_mt_source = os.Getenv("TEST_MT_SOURCE")
-)
-
-func withAuthEnablementV1Beta1(auth authSetup) contribresources.KafkaSourceV1Beta1Option {
-	// We test with sasl512 and enable tls with it, so check tls first
-	if auth.TLSEnabled == true {
-		return func(ks *sourcesv1beta1.KafkaSource) {
-			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
-			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaTLSSecret,
-				},
-				Key: "ca.crt",
-			}
-			ks.Spec.KafkaAuthSpec.Net.TLS.Cert.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaTLSSecret,
-				},
-				Key: "user.crt",
-			}
-			ks.Spec.KafkaAuthSpec.Net.TLS.Key.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaTLSSecret,
-				},
-				Key: "user.key",
-			}
-		}
-	}
-	if auth.SASLEnabled == true {
-		return func(ks *sourcesv1beta1.KafkaSource) {
-			ks.Spec.KafkaAuthSpec.Net.SASL.Enable = true
-			ks.Spec.KafkaAuthSpec.Net.SASL.User.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "user",
-			}
-			ks.Spec.KafkaAuthSpec.Net.SASL.Password.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "password",
-			}
-			ks.Spec.KafkaAuthSpec.Net.SASL.Type.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "saslType",
-			}
-			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
-			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "ca.crt",
-			}
-		}
-	}
-	return func(ks *sourcesv1beta1.KafkaSource) {}
-}
-
-func withAuthEnablementV1Alpha1(auth authSetup) contribresources.KafkaSourceV1Alpha1Option {
-	// We test with sasl512 and enable tls with it, so check tls first
-	if auth.TLSEnabled == true {
-		return func(ks *sourcesv1alpha1.KafkaSource) {
-			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
-			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaTLSSecret,
-				},
-				Key: "ca.crt",
-			}
-			ks.Spec.KafkaAuthSpec.Net.TLS.Cert.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaTLSSecret,
-				},
-				Key: "user.crt",
-			}
-			ks.Spec.KafkaAuthSpec.Net.TLS.Key.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaTLSSecret,
-				},
-				Key: "user.key",
-			}
-		}
-	}
-	if auth.SASLEnabled == true {
-		return func(ks *sourcesv1alpha1.KafkaSource) {
-			ks.Spec.KafkaAuthSpec.Net.SASL.Enable = true
-			ks.Spec.KafkaAuthSpec.Net.SASL.User.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "user",
-			}
-			ks.Spec.KafkaAuthSpec.Net.SASL.Password.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "password",
-			}
-			ks.Spec.KafkaAuthSpec.Net.SASL.Type.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "saslType",
-			}
-			ks.Spec.KafkaAuthSpec.Net.TLS.Enable = true
-			ks.Spec.KafkaAuthSpec.Net.TLS.CACert.SecretKeyRef = &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{
-					Name: kafkaSASLSecret,
-				},
-				Key: "ca.crt",
-			}
-		}
-	}
-	return func(ks *sourcesv1alpha1.KafkaSource) {}
-}
-func testKafkaSource(t *testing.T, name string, version string, messageKey string, messageHeaders map[string]string, messagePayload string, matcherGen func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher, auth authSetup) {
-	name = fmt.Sprintf("%s-%s", name, version)
-
-	var (
-		kafkaTopicName     = uuid.New().String()
-		consumerGroup      = uuid.New().String()
-		recordEventPodName = "e2e-kafka-r-" + strings.ReplaceAll(name, "_", "-")
-		kafkaSourceName    = "e2e-kafka-source-" + strings.ReplaceAll(name, "_", "-")
-	)
-
-	client := testlib.Setup(t, true)
-	defer testlib.TearDown(client)
-
-	if auth.SASLEnabled {
-		_, err := utils.CopySecret(client.Kube.CoreV1(), "knative-eventing", kafkaSASLSecret, client.Namespace, "default")
-		if err != nil {
-			t.Fatalf("could not copy SASL secret(%s): %v", kafkaSASLSecret, err)
-		}
-	}
-	if auth.TLSEnabled {
-		_, err := utils.CopySecret(client.Kube.CoreV1(), "knative-eventing", kafkaTLSSecret, client.Namespace, "default")
-		if err != nil {
-			t.Fatalf("could not copy secret(%s): %v", kafkaTLSSecret, err)
-		}
-	}
-	helpers.MustCreateTopic(client, kafkaClusterName, kafkaClusterNamespace, kafkaTopicName)
-	if len(recordEventPodName) > 63 {
-		recordEventPodName = recordEventPodName[:63]
-	}
-	eventTracker, _ := recordevents.StartEventRecordOrFail(context.Background(), client, recordEventPodName)
-
-	var (
-		cloudEventsSourceName string
-		cloudEventsEventType  string
-	)
-
-	t.Logf("Creating KafkaSource %s", version)
-	switch version {
-	case "v1alpha1":
-		contribtestlib.CreateKafkaSourceV1Alpha1OrFail(client, contribresources.KafkaSourceV1Alpha1(
-			auth.bootStrapServer,
-			kafkaTopicName,
-			resources.ServiceRef(recordEventPodName),
-			contribresources.WithNameV1Alpha1(kafkaSourceName),
-			contribresources.WithConsumerGroupV1Alpha1(consumerGroup),
-			withAuthEnablementV1Alpha1(auth),
-		))
-		cloudEventsSourceName = sourcesv1alpha1.KafkaEventSource(client.Namespace, kafkaSourceName, kafkaTopicName)
-		cloudEventsEventType = sourcesv1alpha1.KafkaEventType
-	case "v1beta1":
-		contribtestlib.CreateKafkaSourceV1Beta1OrFail(client, contribresources.KafkaSourceV1Beta1(
-			auth.bootStrapServer,
-			kafkaTopicName,
-			resources.ServiceRef(recordEventPodName),
-			contribresources.WithNameV1Beta1(kafkaSourceName),
-			contribresources.WithConsumerGroupV1Beta1(consumerGroup),
-			withAuthEnablementV1Beta1(auth),
-		))
-		cloudEventsSourceName = sourcesv1beta1.KafkaEventSource(client.Namespace, kafkaSourceName, kafkaTopicName)
-		cloudEventsEventType = sourcesv1beta1.KafkaEventType
-	default:
-		t.Fatalf("Unknown KafkaSource version %s", version)
-	}
-
-	client.WaitForAllTestResourcesReadyOrFail(context.Background())
-
-	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
-	if test_mt_source == "1" {
-		time.Sleep(20 * time.Second)
-	}
-
-	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain, kafkaTopicName, messageKey, messageHeaders, messagePayload)
-
-	eventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(cloudEventsSourceName, cloudEventsEventType)))
-}
-
 func TestKafkaSource(t *testing.T) {
-	time, _ := cetypes.ParseTime("2018-04-05T17:31:00Z")
-
-	auths := map[string]struct {
-		auth authSetup
-	}{
-		"plain": {
-			auth: authSetup{
-				bootStrapServer: kafkaBootstrapUrlPlain,
-				SASLEnabled:     false,
-				TLSEnabled:      false,
-			},
-		},
-		"s512": {
-			auth: authSetup{
-				bootStrapServer: kafkaBootstrapUrlSASL,
-				SASLEnabled:     true,
-				TLSEnabled:      false,
-			},
-		},
-		"tls": {
-			auth: authSetup{
-				bootStrapServer: kafkaBootstrapUrlTLS,
-				SASLEnabled:     false,
-				TLSEnabled:      true,
-			},
-		},
-	}
-	tests := map[string]struct {
-		messageKey     string
-		messageHeaders map[string]string
-		messagePayload string
-		matcherGen     func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher
-	}{
-		"no_event": {
-			messageKey: "0",
-			messageHeaders: map[string]string{
-				"content-type": "application/json",
-			},
-			messagePayload: `{"value":5}`,
-			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
-				return AllOf(
-					HasSource(cloudEventsSourceName),
-					HasType(cloudEventsEventType),
-					HasDataContentType("application/json"),
-					HasData([]byte(`{"value":5}`)),
-					HasExtension("key", "0"),
-				)
-			},
-		},
-		"no_event_no_content_type": {
-			messageKey:     "0",
-			messagePayload: `{"value":5}`,
-			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
-				return AllOf(
-					HasSource(cloudEventsSourceName),
-					HasType(cloudEventsEventType),
-					HasData([]byte(`{"value":5}`)),
-					HasExtension("key", "0"),
-				)
-			},
-		},
-		"no_event_content_type_or_key": {
-			messagePayload: `{"value":5}`,
-			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
-				return AllOf(
-					HasSource(cloudEventsSourceName),
-					HasType(cloudEventsEventType),
-					HasData([]byte(`{"value":5}`)),
-				)
-			},
-		},
-		"no_event_with_text_plain_body": {
-			messageKey: "0",
-			messageHeaders: map[string]string{
-				"content-type": "text/plain",
-			},
-			messagePayload: "simple 10",
-			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
-				return AllOf(
-					HasSource(cloudEventsSourceName),
-					HasType(cloudEventsEventType),
-					HasDataContentType("text/plain"),
-					HasData([]byte("simple 10")),
-					HasExtension("key", "0"),
-				)
-			},
-		},
-		"structured": {
-			messageHeaders: map[string]string{
-				"content-type": "application/cloudevents+json",
-			},
-			messagePayload: mustJsonMarshal(t, map[string]interface{}{
-				"specversion":     "1.0",
-				"type":            "com.github.pull.create",
-				"source":          "https://github.com/cloudevents/spec/pull",
-				"subject":         "123",
-				"id":              "A234-1234-1234",
-				"time":            "2018-04-05T17:31:00Z",
-				"datacontenttype": "application/json",
-				"data": map[string]string{
-					"hello": "Francesco",
-				},
-				"comexampleextension1": "value",
-				"comexampleothervalue": 5,
-			}),
-			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
-				return AllOf(
-					HasSpecVersion(cloudevents.VersionV1),
-					HasType("com.github.pull.create"),
-					HasSource("https://github.com/cloudevents/spec/pull"),
-					HasSubject("123"),
-					HasId("A234-1234-1234"),
-					HasTime(time),
-					HasDataContentType("application/json"),
-					HasData([]byte(`{"hello":"Francesco"}`)),
-					HasExtension("comexampleextension1", "value"),
-					HasExtension("comexampleothervalue", "5"),
-				)
-			},
-		},
-		"binary": {
-			messageHeaders: map[string]string{
-				"ce_specversion":          "1.0",
-				"ce_type":                 "com.github.pull.create",
-				"ce_source":               "https://github.com/cloudevents/spec/pull",
-				"ce_subject":              "123",
-				"ce_id":                   "A234-1234-1234",
-				"ce_time":                 "2018-04-05T17:31:00Z",
-				"content-type":            "application/json",
-				"ce_comexampleextension1": "value",
-				"ce_comexampleothervalue": "5",
-			},
-			messagePayload: mustJsonMarshal(t, map[string]string{
-				"hello": "Francesco",
-			}),
-			matcherGen: func(cloudEventsSourceName, cloudEventsEventType string) EventMatcher {
-				return AllOf(
-					HasSpecVersion(cloudevents.VersionV1),
-					HasType("com.github.pull.create"),
-					HasSource("https://github.com/cloudevents/spec/pull"),
-					HasSubject("123"),
-					HasId("A234-1234-1234"),
-					HasTime(time),
-					HasDataContentType("application/json"),
-					HasData([]byte(`{"hello":"Francesco"}`)),
-					HasExtension("comexampleextension1", "value"),
-					HasExtension("comexampleothervalue", "5"),
-				)
-			},
-		},
-	}
-	for authName, auth := range auths {
-		for name, test := range tests {
-			test := test
-			name := name + "_" + authName
-			t.Run(name+"-v1alpha1", func(t *testing.T) {
-				testKafkaSource(t, name, "v1alpha1", test.messageKey, test.messageHeaders, test.messagePayload, test.matcherGen, auth.auth)
-			})
-			t.Run(name+"-v1beta1", func(t *testing.T) {
-				testKafkaSource(t, name, "v1beta1", test.messageKey, test.messageHeaders, test.messagePayload, test.matcherGen, auth.auth)
-			})
-		}
-	}
+	AssureKafkaSourceIsOperational(t, func(auth, testCase, version string) bool {
+		return true
+	})
 }
 
-func mustJsonMarshal(t *testing.T, val interface{}) string {
-	data, err := json.Marshal(val)
-	if err != nil {
-		t.Errorf("unexpected error, %v", err)
-	}
-	return string(data)
-}
-
-type message struct {
-	cloudEventType string
-	payload        []byte
-	headers        map[string]string
-	key            string
-}
-
-type updateTest struct {
-	auth      authSetup
-	topicName string
-	sinkName  string
-}
-
-var (
-	defaultKafkaSource = updateTest{
-		auth: authSetup{
-			bootStrapServer: kafkaBootstrapUrlTLS,
-			SASLEnabled:     false,
-			TLSEnabled:      true,
-		},
-		topicName: "initial-topic",
-		sinkName:  "default-event-recorder",
-	}
-)
-
 func TestKafkaSourceUpdate(t *testing.T) {
 	t.Skip("Skip these since they're flaky")
 	testCases := map[string]updateTest{
@@ -546,7 +128,7 @@ func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
 	client.WaitForAllTestResourcesReadyOrFail(context.Background())
 
 	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
-	if test_mt_source == "1" {
+	if testMtSource == "1" {
 		time.Sleep(20 * time.Second)
 	}
 
@@ -592,7 +174,7 @@ func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
 	client.WaitForAllTestResourcesReadyOrFail(context.Background())
 
 	// See https://github.com/knative-sandbox/eventing-kafka/issues/411
-	if test_mt_source == "1" {
+	if testMtSource == "1" {
 		time.Sleep(20 * time.Second)
 	}
 
@@ -608,3 +190,28 @@ func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
 	}
 
 }
+
+type message struct {
+	cloudEventType string
+	payload        []byte
+	headers        map[string]string
+	key            string
+}
+
+type updateTest struct {
+	auth      authSetup
+	topicName string
+	sinkName  string
+}
+
+var (
+	defaultKafkaSource = updateTest{
+		auth: authSetup{
+			bootStrapServer: kafkaBootstrapUrlTLS,
+			SASLEnabled:     false,
+			TLSEnabled:      true,
+		},
+		topicName: "initial-topic",
+		sinkName:  "default-event-recorder",
+	}
+)
diff --git a/test/upgrade/README.md b/test/upgrade/README.md
new file mode 100644
index 000000000..6a8bfe9f7
--- /dev/null
+++ b/test/upgrade/README.md
@@ -0,0 +1,119 @@
+# Upgrade Tests
+
+In order to get coverage for the upgrade process from an operator’s perspective,
+we need an additional suite of tests that perform a complete knative upgrade.
+Running these tests on every commit will ensure that we don’t introduce any
+non-upgradeable changes, so every commit should be releasable.
+
+This is inspired by kubernetes
+[upgrade testing](https://github.com/kubernetes/community/blob/master/contributors/devel/sig-testing/e2e-tests.md#version-skewed-and-upgrade-testing)
+.
+
+These tests are a pretty big hammer in that they cover more than just version
+changes, but it’s one of the only ways to make sure we don’t accidentally make
+breaking changes for now.
+
+## Flow
+
+We’d like to validate that the upgrade doesn’t break any resources (they still
+propagate events) and doesn't break our installation (we can still update
+resources).
+
+At a high level, we want to do this:
+
+1. Install the latest knative release.
+1. Create some resources.
+1. Install knative at HEAD.
+1. Run any post-install jobs that apply for the release to be.
+1. Test those resources, verify that we didn’t break anything.
+
+To achieve that, we created an upgrade framework (knative.dev/pkg/test/upgrade).
+This framework will enforce running upgrade tests in specific order and supports
+continual verification of system under test. In case of Eventing Kafka it is:
+
+1. Install the latest release from GitHub.
+1. Run the `preupgrade` smoke tests.
+1. Start `continual` tests that will propagate events in the background, while
+   upgrading and downgrading.
+1. Install at HEAD (`ko apply -f config/`) and run the post-install jobs.
+1. Run the `postupgrade` smoke tests.
+1. Install the latest release from GitHub.
+1. Run the `postdowngrade` smoke tests.
+1. Stop and verify `continual` tests, checking if every event propagated well.
+
+## Tests
+
+### Smoke tests
+
+Those are just stolen from the e2e tests as one of the simplest cases.
+
+#### preupgrade, postupgrade, postdowngrade
+
+Run the selected smoke tests for channel and source.
+
+### Probe test
+
+In order to verify that we don't have data-plane unavailability during our
+control-plane outages (when we're upgrading the eventing-kafka installation), we
+run a prober test that continually sends events to a Kafka channel and sends
+events to Kafka topic, during the entire upgrade/downgrade process. When the
+upgrade completes, we make sure that all of those events propagated at least
+once.
+
+To achieve that
+a [wathola tool](https://pkg.go.dev/knative.dev/eventing/test/upgrade/prober/wathola)
+has been created. It consists of 4 components: _sender_, _forwarder_,
+_receiver_, and _fetcher_. _Sender_ is the usual Kubernetes deployment that
+publishes events to the system under tests (`KafkaSource` or `KafkaChannel`)
+with given interval. When it terminates (by either `SIGTERM`, or `SIGINT`), a
+`finished` event is generated. _Forwarder_ is a knative serving service that
+scales up from zero to receive the sent events and forward them to given target
+which is the _receiver_ in our case. _Receiver_ is an ordinary deployment that
+collects events from multiple forwarders and has an endpoint `/report` that can
+be polled to get the status of received events. To fetch the report from within
+the cluster _fetcher_ comes in. It's a simple one time job, that will fetch the
+report from _receiver_ and print it on stdout as JSON. That enables the test
+client to download _fetcher_ logs and parse the JSON to get the final report.
+
+Diagram below describe the setup:
+
+```
+                   K8s cluster                            |     Test machine
+                                                          |
+ (deploym.)         (ksvc)            (deploym.)          |
++--------+       +-----------+       +----------+         |    +------------+
+|        |       |           ++      |          |         |    |            |
+| Sender |   +-->| Forwarder ||----->+ Receiver |         |    + TestProber |
+|        |   |   |           ||      |          |<---+    |    |            |
++---+----+   |   +------------|      +----------+    |    |    +------------+
+    |        |    +-----------+                      |    |
+    |        |                                            |    |
+    |        |                               +---------+  |
+    |     +--+------+       +---------+      |         |  |
+    +----->         |       |         |      | Fetcher |  |
+    |     | Channel <-------+ Source  |      |         |  |
+    |     |         |       |         |      +---------+  |
+    |     +---------+       +----^----+        (job)      |
+    |                            |
+    +----------------------------+
+```
+
+#### Probe test configuration
+
+Probe test behavior can be influenced from outside without modifying its source
+code. That can be beneficial if one would like to run upgrade tests in different
+context. One such example might be running Eventing upgrade tests in place that
+have Serving and Eventing both installed. In such environment one can set
+environment variable `E2E_UPGRADE_TESTS_SERVING_USE` to enable usage of ksvc
+forwarder (which is disabled by default):
+
+```
+$ export E2E_UPGRADE_TESTS_SERVING_USE=true
+```
+
+Any option, apart from namespace, in
+[`knative.dev/eventing/test/upgrade/prober.Config`](https://github.com/knative/eventing/blob/022e281/test/upgrade/prober/prober.go#L52-L63)
+struct can be influenced, by using `E2E_UPGRADE_TESTS_XXXXX` environmental
+variable prefix (using
+[kelseyhightower/envconfig](https://github.com/kelseyhightower/envconfig#usage)
+usage).
diff --git a/test/upgrade/continual.go b/test/upgrade/continual.go
index 8948c2de1..b93c6dd7b 100644
--- a/test/upgrade/continual.go
+++ b/test/upgrade/continual.go
@@ -67,6 +67,22 @@ func ChannelContinualTest() pkgupgrade.BackgroundOperation {
 		"ChannelContinualTest", setup, verify)
 }
 
+// SourceContinualTest tests source operation in continual manner during the
+// whole upgrade and downgrade process asserting that all event are propagated
+// well.
+func SourceContinualTest() 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)
+}
+
 func configureKafkaChannelAsDefault(
 	c pkgupgrade.Context,
 	ctx context.Context,
diff --git a/test/upgrade/installation/githead.go b/test/upgrade/installation/githead.go
index fa00deeb2..610ab5d40 100644
--- a/test/upgrade/installation/githead.go
+++ b/test/upgrade/installation/githead.go
@@ -22,5 +22,6 @@ import pkgupgrade "knative.dev/pkg/test/upgrade"
 func GitHead() pkgupgrade.Operation {
 	return pkgupgrade.NewOperation("GitHead", func(c pkgupgrade.Context) {
 		runShellFunc("install_head_consolidated_channel", c)
+		runShellFunc("install_head_consolidated_source", c)
 	})
 }
diff --git a/test/upgrade/installation/stable.go b/test/upgrade/installation/stable.go
index b0338506c..247018b43 100644
--- a/test/upgrade/installation/stable.go
+++ b/test/upgrade/installation/stable.go
@@ -22,5 +22,6 @@ import pkgupgrade "knative.dev/pkg/test/upgrade"
 func LatestStable() pkgupgrade.Operation {
 	return pkgupgrade.NewOperation("LatestStable", func(c pkgupgrade.Context) {
 		runShellFunc("install_released_consolidated_channel", c)
+		runShellFunc("install_released_consolidated_source", c)
 	})
 }
diff --git a/test/upgrade/postdowngrade.go b/test/upgrade/postdowngrade.go
index a07887203..6964973dc 100644
--- a/test/upgrade/postdowngrade.go
+++ b/test/upgrade/postdowngrade.go
@@ -28,3 +28,11 @@ func ChannelPostDowngradeTest() pkgupgrade.Operation {
 			runChannelSmokeTest(c.T)
 		})
 }
+
+// SourcePostDowngradeTest tests source operations after downgrade.
+func SourcePostDowngradeTest() pkgupgrade.Operation {
+	return pkgupgrade.NewOperation("SourcePostDowngradeTest",
+		func(c pkgupgrade.Context) {
+			runSourceSmokeTest(c.T)
+		})
+}
diff --git a/test/upgrade/postupgrade.go b/test/upgrade/postupgrade.go
index cb4b03a53..f6a9f8fad 100644
--- a/test/upgrade/postupgrade.go
+++ b/test/upgrade/postupgrade.go
@@ -27,3 +27,11 @@ func ChannelPostUpgradeTest() pkgupgrade.Operation {
 			runChannelSmokeTest(c.T)
 		})
 }
+
+// SourcePostUpgradeTest tests source operations after upgrade.
+func SourcePostUpgradeTest() pkgupgrade.Operation {
+	return pkgupgrade.NewOperation("SourcePostUpgradeTest",
+		func(c pkgupgrade.Context) {
+			runSourceSmokeTest(c.T)
+		})
+}
diff --git a/test/upgrade/preupgrade.go b/test/upgrade/preupgrade.go
index cab8f8a52..b553cfe61 100644
--- a/test/upgrade/preupgrade.go
+++ b/test/upgrade/preupgrade.go
@@ -27,3 +27,11 @@ func ChannelPreUpgradeTest() pkgupgrade.Operation {
 			runChannelSmokeTest(c.T)
 		})
 }
+
+// SourcePreUpgradeTest tests source operations before upgrade.
+func SourcePreUpgradeTest() pkgupgrade.Operation {
+	return pkgupgrade.NewOperation("SourcePreUpgradeTest",
+		func(c pkgupgrade.Context) {
+			runSourceSmokeTest(c.T)
+		})
+}
diff --git a/test/upgrade/smoke.go b/test/upgrade/smoke.go
index b1b49cbf1..bfe4c0f88 100644
--- a/test/upgrade/smoke.go
+++ b/test/upgrade/smoke.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"knative.dev/eventing-kafka/test/e2e"
 	"knative.dev/eventing/test/e2e/helpers"
 )
 
@@ -38,6 +39,12 @@ func runChannelSmokeTest(t *testing.T) {
 	}
 }
 
+func runSourceSmokeTest(t *testing.T) {
+	e2e.AssureKafkaSourceIsOperational(t, func(auth, testCase, version string) bool {
+		return auth == "plain" && (testCase == "structured" || testCase == "binary")
+	})
+}
+
 type smokeTestCase struct {
 	name     string
 	encoding cloudevents.Encoding
diff --git a/test/upgrade/suite.go b/test/upgrade/suite.go
index 7fa9fb406..6771ccb79 100644
--- a/test/upgrade/suite.go
+++ b/test/upgrade/suite.go
@@ -27,15 +27,19 @@ func Suite() pkgupgrade.Suite {
 		Tests: pkgupgrade.Tests{
 			PreUpgrade: []pkgupgrade.Operation{
 				ChannelPreUpgradeTest(),
+				SourcePreUpgradeTest(),
 			},
 			PostUpgrade: []pkgupgrade.Operation{
 				ChannelPostUpgradeTest(),
+				SourcePostUpgradeTest(),
 			},
 			PostDowngrade: []pkgupgrade.Operation{
 				ChannelPostDowngradeTest(),
+				SourcePostDowngradeTest(),
 			},
 			Continual: []pkgupgrade.BackgroundOperation{
 				ChannelContinualTest(),
+				SourceContinualTest(),
 			},
 		},
 		Installations: pkgupgrade.Installations{