Repository: knative-sandbox/eventing-kafka
Pull request #220 (2020-11-23 17:18:11)

Use SYSTEM_NAMESPACE variable instead of hardcoded value
## Proposed Changes

- 🧽  Remove references to hard-coded "knative-eventing" namespace, preferring system.Namespace() call
- 🧽  Store the environment, loaded in the controller main(), in the context instead of re-reading it whenever NewController is called


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/220#" title="Author self-approved">eric-sap</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **vaikas** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @vaikas` in a comment when ready.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/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":["vaikas"]} -->
knative-prow-robot: Hi @eric-sap. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<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>

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220?src=pr&el=h1) Report
> Merging [#220](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220?src=pr&el=desc) (254d063) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/e71a1b1373574d96e9f951d7eb76eae8192ff466?el=desc) (e71a1b1) will **increase** coverage by `0.08%`.
> The diff coverage is `95.83%`.

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

```diff
@@            Coverage Diff             @@
##           master     #220      +/-   ##
==========================================
+ Coverage   75.99%   76.08%   +0.08%     
==========================================
  Files         112      112              
  Lines        4274     4281       +7     
==========================================
+ Hits         3248     3257       +9     
+ Misses        831      829       -2     
  Partials      195      195              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...l/distributed/controller/kafkasecret/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYXNlY3JldC9jb250cm9sbGVyLmdv) | `68.75% <50.00%> (ø)` | |
| [...kg/channel/distributed/common/kafka/admin/admin.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL2FkbWluL2FkbWluLmdv) | `52.94% <100.00%> (ø)` | |
| [.../channel/distributed/controller/env/environment.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9lbnYvZW52aXJvbm1lbnQuZ28=) | `100.00% <100.00%> (ø)` | |
| [...nel/distributed/controller/kafkachannel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvY2hhbm5lbC5nbw==) | `93.84% <100.00%> (ø)` | |
| [.../distributed/controller/kafkachannel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvY29udHJvbGxlci5nbw==) | `79.68% <100.00%> (ø)` | |
| [.../distributed/controller/kafkachannel/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvZGlzcGF0Y2hlci5nbw==) | `90.09% <100.00%> (ø)` | |
| [...nel/distributed/controller/kafkasecret/receiver.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYXNlY3JldC9yZWNlaXZlci5nbw==) | `93.33% <100.00%> (ø)` | |
| [...tributed/common/kafka/admin/eventhubcache/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL2FkbWluL2V2ZW50aHViY2FjaGUvY2FjaGUuZ28=) | `67.46% <0.00%> (+2.40%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220?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/220?src=pr&el=footer). Last update [e71a1b1...eeeec86](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/220?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

travis-minke-sap: /ok-to-test
travis-minke-sap: /approve
/hold

* holding for some final manual testing...
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/220#" title="Author self-approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/220#issuecomment-732361821" title="Approved">travis-minke-sap</a>*

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/OWNERS)~~ [travis-minke-sap]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @eric-sap: PR needs rebase.

<details>

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

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

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333408792326443008) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333408792326443008) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333408792326443008) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333408792326443008) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333408792364191744) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333408792326443008) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333408792364191744) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333408792435494912) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333408792326443008) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333408792364191744) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333408792364191744) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333408792343220224) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526415550713856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333408792418717696) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526415550713856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526415529742336) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

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

<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: @eric-sap: 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-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-build-tests | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333408792401940480) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526415550713856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526415529742336) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-unit-tests | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333526415634599936) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<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: @eric-sap: 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-distributed | cf0236e418d5b6d014b8dfc8d01f729d1a871633 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333408792380968960) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526415550713856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526415529742336) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-unit-tests | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333526415634599936) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333526415609434112) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<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: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333526415655571456) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526415550713856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526415529742336) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-unit-tests | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-unit-tests/1333526415634599936) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333526415609434112) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526415571685376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @eric-sap: 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 | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526415550713856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-build-tests | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-build-tests/1333526415609434112) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526415571685376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

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

<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: @eric-sap: 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 | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526415508770816) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526415571685376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526767368933376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<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: @eric-sap: 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-distributed | 2dc461b729b8db405eff2c7df58cbc2056781159 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526415571685376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526767368933376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526767331184640) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<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: @eric-sap: 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-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333526767368933376) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526767331184640) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526767398293504) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333530839283142656/artifacts/line-cov.html#file17) | 84.9% | 77.9% | -7.0

knative-prow-robot: @eric-sap: 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-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333526767331184640) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526767398293504) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333532039281905664) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<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: @eric-sap: 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-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333526767398293504) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333532039281905664) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333532039231574016) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<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: @eric-sap: 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-tls | 842eedeff86ea18dd506b3bc60e40e16d1e958e0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333526767347961856) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333532039281905664) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333532039231574016) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333532039307071488) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @eric-sap: 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 | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1333532039281905664) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1333532039231574016) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1333532039307071488) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | ec23044a1acfb9ee1388915c0da6b0f86b1aaee0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1333532039256739840) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

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

<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/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/220/pull-knative-sandbox-eventing-kafka-go-coverage/1333543757785600000/artifacts/line-cov.html#file17) | 84.9% | 77.9% | -7.0


Patch:
diff --git a/cmd/channel/consolidated/dispatcher/main.go b/cmd/channel/consolidated/dispatcher/main.go
index 3d36c63e7..476dd8d0e 100644
--- a/cmd/channel/consolidated/dispatcher/main.go
+++ b/cmd/channel/consolidated/dispatcher/main.go
@@ -35,8 +35,5 @@ func main() {
 		ctx = injection.WithNamespaceScope(ctx, ns)
 	}
 
-	// Do not run the dispatcher in leader-election mode
-	ctx = sharedmain.WithHADisabled(ctx)
-
 	sharedmain.MainWithContext(ctx, component, controller.NewController)
 }
diff --git a/cmd/channel/distributed/controller/main.go b/cmd/channel/distributed/controller/main.go
index e8c386cad..b902f3bf9 100644
--- a/cmd/channel/distributed/controller/main.go
+++ b/cmd/channel/distributed/controller/main.go
@@ -20,6 +20,8 @@ import (
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	// _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
+	"context"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
@@ -46,5 +48,6 @@ func main() {
 		logger.Fatal("Failed To Load Environment Variables - Terminating!", zap.Error(err))
 	}
 	ctx = controller.WithResyncPeriod(ctx, environment.ResyncPeriod)
+	ctx = context.WithValue(ctx, env.Key{}, environment)
 	sharedmain.MainWithContext(ctx, constants.ControllerComponentName, kafkachannel.NewController, kafkasecret.NewController)
 }
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index e25e80ac1..b5901937e 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -83,13 +83,13 @@ func main() {
 	sarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
 
 	// Initialize Tracing (Watches config-tracing ConfigMap, Assumes Context Came From LoggingContext With Embedded K8S Client Key)
-	err = commonconfig.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName)
+	err = commonconfig.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Failed To Initialize Tracing - Terminating", zap.Error(err))
 	}
 
 	// Initialize Observability (Watches config-observability ConfigMap And Starts Profiling Server)
-	err = commonconfig.InitializeObservability(ctx, logger.Sugar(), environment.MetricsDomain, environment.MetricsPort)
+	err = commonconfig.InitializeObservability(ctx, logger.Sugar(), environment.MetricsDomain, environment.MetricsPort, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Failed To Initialize Observability - Terminating", zap.Error(err))
 	}
@@ -115,7 +115,7 @@ func main() {
 	dispatcher = dispatch.NewDispatcher(dispatcherConfig)
 
 	// Watch The Settings ConfigMap For Changes
-	err = commonconfig.InitializeConfigWatcher(ctx, logger.Sugar(), configMapObserver)
+	err = commonconfig.InitializeConfigWatcher(ctx, logger.Sugar(), configMapObserver, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
 	}
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 8a407a7ca..d293d0144 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -86,13 +86,13 @@ func main() {
 	sarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
 
 	// Initialize Tracing (Watches config-tracing ConfigMap, Assumes Context Came From LoggingContext With Embedded K8S Client Key)
-	err = commonconfig.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName)
+	err = commonconfig.InitializeTracing(logger.Sugar(), ctx, environment.ServiceName, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Could Not Initialize Tracing - Terminating", zap.Error(err))
 	}
 
 	// Initialize Observability (Watches config-observability ConfigMap And Starts Profiling Server)
-	err = commonconfig.InitializeObservability(ctx, logger.Sugar(), environment.MetricsDomain, environment.MetricsPort)
+	err = commonconfig.InitializeObservability(ctx, logger.Sugar(), environment.MetricsDomain, environment.MetricsPort, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Could Not Initialize Observability - Terminating", zap.Error(err))
 	}
@@ -112,7 +112,7 @@ func main() {
 	statsReporter := metrics.NewStatsReporter(logger)
 
 	// Watch The Settings ConfigMap For Changes
-	err = commonconfig.InitializeConfigWatcher(ctx, logger.Sugar(), configMapObserver)
+	err = commonconfig.InitializeConfigWatcher(ctx, logger.Sugar(), configMapObserver, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
 	}
diff --git a/go.mod b/go.mod
index 0cec18181..beb5c0f96 100644
--- a/go.mod
+++ b/go.mod
@@ -32,9 +32,9 @@ require (
 	k8s.io/apimachinery v0.18.12
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.19.1-0.20201127005336-6065b380266f
+	knative.dev/eventing v0.19.1-0.20201130190736-241af67e1844
 	knative.dev/hack v0.0.0-20201125230335-c46a6498e9ed
-	knative.dev/pkg v0.0.0-20201127013335-0d896b5c87b8
+	knative.dev/pkg v0.0.0-20201130192436-e5346d90e980
 )
 
 replace (
diff --git a/go.sum b/go.sum
index fc4bd6b43..2332743e7 100644
--- a/go.sum
+++ b/go.sum
@@ -1228,8 +1228,8 @@ k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29/go.mod h1:F+5wygcW0wmRTnM
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.19.1-0.20201127005336-6065b380266f h1:mFXWnW07P2rTRPhKUSBivtsbfKtVOGkSPv3j4tFUFQc=
-knative.dev/eventing v0.19.1-0.20201127005336-6065b380266f/go.mod h1:Rzd5RlKewLIDG/fIelqfDoNWHGuPcwD2UEK4hBXjjZs=
+knative.dev/eventing v0.19.1-0.20201130190736-241af67e1844 h1:IghjQeIkwSa6skZdUASyYM9b7tyyTttEA5r9mbdLxAE=
+knative.dev/eventing v0.19.1-0.20201130190736-241af67e1844/go.mod h1:SD32TaZ0N3jbq5XxsuB8+uBuPUSI3eqaPswStSnjoxE=
 knative.dev/hack v0.0.0-20201112185459-01a34c573bd8 h1:RNbZsAjhswBPtl4C5C5gEFX5/GfWIOZQxfYD9DhkHdY=
 knative.dev/hack v0.0.0-20201112185459-01a34c573bd8/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20201118155651-b31d3bb6bff9 h1:OgYL1cATsWTS6kztSOuQqhmYfcn6ZdpPAQQ6TKCx00I=
@@ -1240,10 +1240,10 @@ knative.dev/hack v0.0.0-20201125230335-c46a6498e9ed h1:DrTU+vxQrNJtySxyNovUtl4si
 knative.dev/hack v0.0.0-20201125230335-c46a6498e9ed/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/pkg v0.0.0-20201117221452-0fccc54273ed h1:4LNEFhvD2Ya4WgCD3SC1WGXz81bjEvQgbcpe5XACI3I=
 knative.dev/pkg v0.0.0-20201117221452-0fccc54273ed/go.mod h1:nxlh3CUvx6WBPr1WKD96AHxFZPD2UKRDo9RUp8ILTyQ=
-knative.dev/pkg v0.0.0-20201125095035-9bf616d2f46a h1:pdJpLaq50mLKrPWYSQgTH2p64Dk7Fq/xID6l0F69cVY=
-knative.dev/pkg v0.0.0-20201125095035-9bf616d2f46a/go.mod h1:wXZqP8MXCxb51yNFlecA13BwG7Hk370SWDFWV4dx4ug=
 knative.dev/pkg v0.0.0-20201127013335-0d896b5c87b8 h1:/JA9H9Q3nP2wb+c+5pD4/6KNr9c4Wb9Cg+OsbAvTIT0=
 knative.dev/pkg v0.0.0-20201127013335-0d896b5c87b8/go.mod h1:wXZqP8MXCxb51yNFlecA13BwG7Hk370SWDFWV4dx4ug=
+knative.dev/pkg v0.0.0-20201130192436-e5346d90e980 h1:44zakwrFjbivCM/YR+ni70e9u1ELtwA7VF1HSmUcj+g=
+knative.dev/pkg v0.0.0-20201130192436-e5346d90e980/go.mod h1:wXZqP8MXCxb51yNFlecA13BwG7Hk370SWDFWV4dx4ug=
 knative.dev/reconciler-test v0.0.0-20201124190335-83a44efcdfef/go.mod h1:YSs1y1rgnjs8w39/drLIOQbWvZUQwqApvd+EizO8UsA=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/pkg/channel/consolidated/kafka/admin.go b/pkg/channel/consolidated/kafka/admin.go
new file mode 100644
index 000000000..f883e6a11
--- /dev/null
+++ b/pkg/channel/consolidated/kafka/admin.go
@@ -0,0 +1,109 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	"go.uber.org/zap"
+
+	"github.com/Shopify/sarama"
+	"knative.dev/pkg/logging"
+)
+
+var mutex sync.Mutex
+
+type ClusterAdminFactory func() (sarama.ClusterAdmin, error)
+
+type AdminClient interface {
+	// ListConsumerGroups Lists the consumer groups
+	ListConsumerGroups() ([]string, error)
+}
+
+// AdminClientManager manages a ClusterAdmin connection and recreates one when needed
+// it is made to overcome https://github.com/Shopify/sarama/issues/1162
+type AdminClientManager struct {
+	logger       *zap.SugaredLogger
+	adminFactory ClusterAdminFactory
+	clusterAdmin sarama.ClusterAdmin
+}
+
+func NewAdminClient(ctx context.Context, caFactory ClusterAdminFactory) (AdminClient, error) {
+	logger := logging.FromContext(ctx)
+	logger.Info("Creating a new AdminClient")
+	kafkaClusterAdmin, err := caFactory()
+	if err != nil {
+		logger.Errorw("error while creating ClusterAdmin", zap.Error(err))
+		return nil, err
+	}
+	return &AdminClientManager{
+		logger:       logger,
+		adminFactory: caFactory,
+		clusterAdmin: kafkaClusterAdmin,
+	}, nil
+}
+
+// ListConsumerGroups Returns a list of the consumer groups.
+//
+// In the occasion of errors, there will be a retry with an exponential backoff.
+// Due to a known issue in Sarama ClusterAdmin https://github.com/Shopify/sarama/issues/1162,
+// a new ClusterAdmin will be created with every retry until the call succeeds or
+// the timeout is reached.
+func (c *AdminClientManager) ListConsumerGroups() ([]string, error) {
+	c.logger.Info("Attempting to list consumer group")
+	mutex.Lock()
+	defer mutex.Unlock()
+	r := 0
+	// This gives us around ~13min of exponential backoff
+	max := 13
+	cgsMap, err := c.clusterAdmin.ListConsumerGroups()
+	for err != nil && r <= max {
+		// There's on error, let's retry and presume a new ClusterAdmin can fix it
+
+		// Calculate incremental delay following this https://docs.aws.amazon.com/general/latest/gr/api-retries.html
+		t := int(math.Pow(2, float64(r)) * 100)
+		d := time.Duration(t) * time.Millisecond
+		c.logger.Errorw("listing consumer group failed. Refreshing the ClusterAdmin and retrying.",
+			zap.Error(err),
+			zap.Duration("retry after", d),
+			zap.Int("Retry attempt", r),
+			zap.Int("Max retries", max),
+		)
+		time.Sleep(d)
+
+		// let's reconnect and try again
+		c.clusterAdmin, err = c.adminFactory()
+		r += 1
+		if err != nil {
+			// skip this attempt
+			continue
+		}
+		cgsMap, err = c.clusterAdmin.ListConsumerGroups()
+	}
+
+	if r > max {
+		return nil, fmt.Errorf("failed to refresh the culster admin and retry: %v", err)
+	}
+
+	return sets.StringKeySet(cgsMap).List(), nil
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
new file mode 100644
index 000000000..8b808c988
--- /dev/null
+++ b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
@@ -0,0 +1,199 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/kafka"
+)
+
+var (
+	watchersMtx sync.RWMutex
+	cacheMtx    sync.RWMutex
+	// Hooks into the poll logic for testing
+	after = time.After
+	done  = func() {}
+)
+
+type ConsumerGroupHandler func()
+type Matcher func(string) bool
+
+type ConsumerGroupWatcher interface {
+	// Start instructs the watcher to start polling for the consumer groups and
+	// notify any observers on the event of any changes
+	Start() error
+
+	// Terminate instructs the watcher to stop polling and clear the watchers cache
+	Terminate()
+
+	// Watch registers callback on the event of any changes observed
+	// on the consumer groups. watcherID is an arbitrary string the user provides
+	// that will be used to identify his callbacks when provided to Forget(watcherID).
+	//
+	// To ensure this is event-triggered, level-driven,
+	// we don't pass the updates to the callback, instead the observer is expected
+	// to use List() to get the updated list of ConsumerGroups.
+	Watch(watcherID string, callback ConsumerGroupHandler) error
+
+	// Forget removes all callbacks that correspond to the watcherID
+	Forget(watcherID string)
+
+	// List returns all the cached consumer groups that match matcher.
+	// It will return an empty slice if none matched or the cache is empty
+	List(matcher Matcher) []string
+}
+
+type WatcherImpl struct {
+	logger *zap.SugaredLogger
+	//TODO name?
+	watchers             map[string]ConsumerGroupHandler
+	cachedConsumerGroups sets.String
+	adminClient          kafka.AdminClient
+	pollDuration         time.Duration
+	done                 chan struct{}
+}
+
+func NewConsumerGroupWatcher(ctx context.Context, ac kafka.AdminClient, pollDuration time.Duration) ConsumerGroupWatcher {
+	return &WatcherImpl{
+		logger:               logging.FromContext(ctx),
+		adminClient:          ac,
+		pollDuration:         pollDuration,
+		watchers:             make(map[string]ConsumerGroupHandler),
+		cachedConsumerGroups: sets.String{},
+	}
+}
+
+func (w *WatcherImpl) Start() error {
+	w.logger.Infow("ConsumerGroupWatcher starting. Polling for consumer groups", zap.Duration("poll duration", w.pollDuration))
+	go func() {
+		for {
+			select {
+			case <-after(w.pollDuration):
+				// let's get current observed consumer groups
+				observedCGs, err := w.adminClient.ListConsumerGroups()
+				if err != nil {
+					w.logger.Errorw("error while listing consumer groups", zap.Error(err))
+					continue
+				}
+				var notify bool
+				var changedGroup string
+				observedCGsSet := sets.String{}.Insert(observedCGs...)
+				// Look for observed CGs
+				for c := range observedCGsSet {
+					if !w.cachedConsumerGroups.Has(c) {
+						// This is the first appearance.
+						w.logger.Debugw("Consumer group observed. Caching.",
+							zap.String("consumer group", c))
+						changedGroup = c
+						notify = true
+						break
+					}
+				}
+				// Look for disappeared CGs
+				for c := range w.cachedConsumerGroups {
+					if !observedCGsSet.Has(c) {
+						// This CG was cached but it's no longer there.
+						w.logger.Debugw("Consumer group deleted.",
+							zap.String("consumer group", c))
+						changedGroup = c
+						notify = true
+						break
+					}
+				}
+				if notify {
+					cacheMtx.Lock()
+					w.cachedConsumerGroups = observedCGsSet
+					cacheMtx.Unlock()
+					w.notify(changedGroup)
+				}
+				done()
+			case <-w.done:
+				break
+			}
+		}
+	}()
+	return nil
+}
+
+func (w *WatcherImpl) Terminate() {
+	watchersMtx.Lock()
+	cacheMtx.Lock()
+	defer watchersMtx.Unlock()
+	defer cacheMtx.Unlock()
+
+	w.watchers = nil
+	w.cachedConsumerGroups = nil
+	w.done <- struct{}{}
+}
+
+// TODO explore returning a channel instead of a taking callback
+func (w *WatcherImpl) Watch(watcherID string, cb ConsumerGroupHandler) error {
+	w.logger.Debugw("Adding a new watcher", zap.String("watcherID", watcherID))
+	watchersMtx.Lock()
+	defer watchersMtx.Unlock()
+	w.watchers[watcherID] = cb
+
+	// notify at least once to get the current state
+	cb()
+	return nil
+}
+
+func (w *WatcherImpl) Forget(watcherID string) {
+	w.logger.Debugw("Forgetting watcher", zap.String("watcherID", watcherID))
+	watchersMtx.Lock()
+	defer watchersMtx.Unlock()
+	delete(w.watchers, watcherID)
+}
+
+func (w *WatcherImpl) List(matcher Matcher) []string {
+	w.logger.Debug("Listing consumer groups")
+	cacheMtx.RLock()
+	defer cacheMtx.RUnlock()
+	cgs := make([]string, 0)
+	for cg := range w.cachedConsumerGroups {
+		if matcher(cg) {
+			cgs = append(cgs, cg)
+		}
+	}
+	return cgs
+}
+
+func (w *WatcherImpl) notify(cg string) {
+	watchersMtx.RLock()
+	defer watchersMtx.RUnlock()
+
+	for _, cb := range w.watchers {
+		cb()
+	}
+}
+
+func Find(list []string, item string) bool {
+	for _, i := range list {
+		if i == item {
+			return true
+		}
+	}
+	return false
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index 7b69a1f47..677d9f222 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -20,36 +20,27 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
+	"time"
 
-	"knative.dev/eventing-kafka/pkg/source"
-
-	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/kafka"
 
 	"github.com/Shopify/sarama"
-
 	"go.uber.org/zap"
-
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	rbacv1listers "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/client-go/tools/cache"
-
-	"knative.dev/eventing/pkg/apis/eventing"
-	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
-	"knative.dev/pkg/logging"
-	"knative.dev/pkg/network"
-
-	"knative.dev/pkg/apis"
-	"knative.dev/pkg/controller"
-	pkgreconciler "knative.dev/pkg/reconciler"
+	"k8s.io/utils/pointer"
 
 	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
@@ -58,6 +49,16 @@ import (
 	kafkaScheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
 	kafkaChannelReconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
 	listers "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/source"
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/apis/eventing"
+	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/apis/duck"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
+	pkgreconciler "knative.dev/pkg/reconciler"
 )
 
 const (
@@ -75,6 +76,8 @@ const (
 	dispatcherRoleBindingCreated    = "DispatcherRoleBindingCreated"
 
 	dispatcherName = "kafka-ch-dispatcher"
+
+	pollInterval = 2 * time.Second
 )
 
 func newReconciledNormal(namespace, name string) pkgreconciler.Event {
@@ -116,11 +119,10 @@ type Reconciler struct {
 	kafkaAuthConfig  *utils.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
-
+	kafkaClusterAdmin    sarama.ClusterAdmin
+	consumerGroupWatcher ConsumerGroupWatcher
 	kafkachannelLister   listers.KafkaChannelLister
 	kafkachannelInformer cache.SharedIndexInformer
 	deploymentLister     appsv1listers.DeploymentLister
@@ -162,7 +164,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 		return r.kafkaConfigError
 	}
 
-	kafkaClusterAdmin, err := r.createClient(ctx, kc)
+	kafkaClusterAdmin, err := r.createClient()
 	if err != nil {
 		kc.Status.MarkConfigFailed("InvalidConfiguration", "Unable to build Kafka admin client for channel %s: %v", kc.Name, err)
 		return err
@@ -177,7 +179,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 	// 4. Dispatcher endpoints to ensure that there's something backing the Service.
 	// 5. K8s service representing the channel that will use ExternalName to point to the Dispatcher k8s service.
 
-	if err := r.createTopic(ctx, kc, kafkaClusterAdmin); err != nil {
+	if err := r.reconcileTopic(ctx, kc, kafkaClusterAdmin); err != nil {
 		kc.Status.MarkTopicFailed("TopicCreateFailed", "error while creating topic: %s", err)
 		return err
 	}
@@ -238,7 +240,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 		Scheme: "http",
 		Host:   network.GetServiceHostname(svc.Name, svc.Namespace),
 	})
-
+	err = r.setupSubscriptionStatusWatcher(ctx, kc)
+	if err != nil {
+		logger.Errorw("error setting up some subscription status watchers", zap.Error(err))
+	}
 	// close the connection
 	err = kafkaClusterAdmin.Close()
 	if err != nil {
@@ -251,6 +256,60 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 	return newReconciledNormal(kc.Namespace, kc.Name)
 }
 
+func (r *Reconciler) setupSubscriptionStatusWatcher(ctx context.Context, channel *v1beta1.KafkaChannel) error {
+	var err error
+	groupIDPrefix := fmt.Sprintf("kafka.%s.%s", channel.Namespace, channel.Name)
+
+	m := func(cg string) bool {
+		return strings.HasPrefix(cg, groupIDPrefix)
+	}
+	err = r.consumerGroupWatcher.Watch(string(channel.ObjectMeta.UID), func() {
+		err := r.markSubscriptionReadiness(ctx, channel, r.consumerGroupWatcher.List(m))
+		if err != nil {
+			logging.FromContext(ctx).Errorw("error updating subscription readiness", zap.Error(err))
+		}
+	})
+	return err
+}
+
+func (r *Reconciler) markSubscriptionReadiness(ctx context.Context, ch *v1beta1.KafkaChannel, cgs []string) error {
+	after := ch.DeepCopy()
+	after.Status.Subscribers = make([]v1.SubscriberStatus, 0)
+
+	for _, s := range ch.Spec.Subscribers {
+		cg := fmt.Sprintf("kafka.%s.%s.%s", ch.Namespace, ch.Name, s.UID)
+		if Find(cgs, cg) {
+			logging.FromContext(ctx).Debugw("marking subscription", zap.Any("subscription", s))
+			after.Status.Subscribers = append(after.Status.Subscribers, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionTrue,
+			})
+		}
+	}
+
+	jsonPatch, err := duck.CreatePatch(ch, after)
+	if err != nil {
+		return fmt.Errorf("creating JSON patch: %w", err)
+	}
+	// If there is nothing to patch, we are good, just return.
+	// Empty patch is [], hence we check for that.
+	if len(jsonPatch) == 0 {
+		return nil
+	}
+	patch, err := jsonPatch.MarshalJSON()
+	if err != nil {
+		return fmt.Errorf("marshaling JSON patch: %w", err)
+	}
+	patched, err := r.kafkaClientSet.MessagingV1beta1().KafkaChannels(ch.Namespace).Patch(ctx, ch.Name, types.JSONPatchType, patch, metav1.PatchOptions{}, "status")
+
+	if err != nil {
+		return fmt.Errorf("Failed patching: %w", err)
+	}
+	logging.FromContext(ctx).Debugw("Patched resource", zap.Any("patch", patch), zap.Any("patched", patched))
+	return nil
+}
+
 func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope string, dispatcherNamespace string, kc *v1beta1.KafkaChannel) (*appsv1.Deployment, error) {
 	if scope == scopeNamespace {
 		// Configure RBAC in namespace to access the configmaps
@@ -464,7 +523,7 @@ func (r *Reconciler) reconcileChannelService(ctx context.Context, dispatcherName
 	return svc, nil
 }
 
-func (r *Reconciler) createClient(ctx context.Context, kc *v1beta1.KafkaChannel) (sarama.ClusterAdmin, error) {
+func (r *Reconciler) createClient() (sarama.ClusterAdmin, error) {
 	// We don't currently initialize r.kafkaClusterAdmin, hence we end up creating the cluster admin client every time.
 	// This is because of an issue with Shopify/sarama. See https://github.com/Shopify/sarama/issues/1162.
 	// Once the issue is fixed we should use a shared cluster admin client. Also, r.kafkaClusterAdmin is currently
@@ -480,7 +539,7 @@ func (r *Reconciler) createClient(ctx context.Context, kc *v1beta1.KafkaChannel)
 	return kafkaClusterAdmin, nil
 }
 
-func (r *Reconciler) createTopic(ctx context.Context, channel *v1beta1.KafkaChannel, kafkaClusterAdmin sarama.ClusterAdmin) error {
+func (r *Reconciler) reconcileTopic(ctx context.Context, channel *v1beta1.KafkaChannel, kafkaClusterAdmin sarama.ClusterAdmin) error {
 	logger := logging.FromContext(ctx)
 
 	topicName := utils.TopicName(utils.KafkaChannelSeparator, channel.Namespace, channel.Name)
@@ -516,10 +575,11 @@ func (r *Reconciler) deleteTopic(ctx context.Context, channel *v1beta1.KafkaChan
 }
 
 func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.ConfigMap) {
-	logging.FromContext(ctx).Info("Reloading Kafka configuration")
+	logger := logging.FromContext(ctx)
+	logger.Info("Reloading Kafka configuration")
 	kafkaConfig, err := utils.GetKafkaConfig(configMap.Data)
 	if err != nil {
-		logging.FromContext(ctx).Errorw("Error reading Kafka configuration", zap.Error(err))
+		logger.Errorw("Error reading Kafka configuration", zap.Error(err))
 	}
 
 	if kafkaConfig.AuthSecretName != "" {
@@ -530,15 +590,32 @@ 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
+	ac, err := kafka.NewAdminClient(ctx, func() (sarama.ClusterAdmin, error) {
+		return source.MakeAdminClient(controllerAgentName, r.kafkaAuthConfig, kafkaConfig.Brokers)
+	})
+
+	if err != nil {
+		logger.Errorw("Error creating AdminClient", zap.Error(err))
+		return
+	}
+
+	if r.consumerGroupWatcher != nil {
+		r.consumerGroupWatcher.Terminate()
+	}
+
+	r.consumerGroupWatcher = NewConsumerGroupWatcher(ctx, ac, pollInterval)
+	//TODO handle error
+	r.consumerGroupWatcher.Start()
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
 	// Do not attempt retrying creating the client because it might be a permanent error
 	// in which case the finalizer will never get removed.
-	if kafkaClusterAdmin, err := r.createClient(ctx, kc); err == nil && r.kafkaConfig != nil {
+	if kafkaClusterAdmin, err := r.createClient(); err == nil && r.kafkaConfig != nil {
 		if err := r.deleteTopic(ctx, kc, kafkaClusterAdmin); err != nil {
 			return err
 		}
 	}
+	r.consumerGroupWatcher.Forget(string(kc.ObjectMeta.UID))
 	return newReconciledNormal(kc.Namespace, kc.Name) //ok to remove finalizer
 }
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index 25824e22c..0e1cf95a0 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -21,13 +21,10 @@ import (
 	"fmt"
 
 	"go.uber.org/zap"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
 
-	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/apis/eventing"
 	"knative.dev/eventing/pkg/channel/fanout"
 	"knative.dev/eventing/pkg/kncloudevents"
@@ -117,7 +114,9 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		kafkachannelLister:   kafkaChannelInformer.Lister(),
 		kafkachannelInformer: kafkaChannelInformer.Informer(),
 	}
-	r.impl = kafkachannelreconciler.NewImpl(ctx, r)
+	r.impl = kafkachannelreconciler.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
+		return controller.Options{SkipStatusUpdates: true}
+	})
 
 	logger.Info("Setting up event handlers")
 
@@ -146,6 +145,16 @@ func filterWithAnnotation(namespaced bool) func(obj interface{}) bool {
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
+	logging.FromContext(ctx).Debugw("ReconcileKind for channel", zap.String("channel", kc.Name))
+	return r.syncDispatcher(ctx)
+}
+
+func (r *Reconciler) ObserveKind(ctx context.Context, kc *v1beta1.KafkaChannel) pkgreconciler.Event {
+	logging.FromContext(ctx).Debugw("ObserveKind for channel", zap.String("channel", kc.Name))
+	return r.syncDispatcher(ctx)
+}
+
+func (r *Reconciler) syncDispatcher(ctx context.Context) pkgreconciler.Event {
 	channels, err := r.kafkachannelLister.List(labels.Everything())
 	if err != nil {
 		logging.FromContext(ctx).Error("Error listing kafka channels")
@@ -172,7 +181,6 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 		logging.FromContext(ctx).Error("Error updating kafka consumers in dispatcher")
 		return err
 	}
-	kc.Status.SubscribableStatus = r.createSubscribableStatus(&kc.Spec.SubscribableSpec, failedSubscriptions)
 	if len(failedSubscriptions) > 0 {
 		logging.FromContext(ctx).Error("Some kafka subscriptions failed to subscribe")
 		return fmt.Errorf("Some kafka subscriptions failed to subscribe")
@@ -180,28 +188,6 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 	return nil
 }
 
-func (r *Reconciler) createSubscribableStatus(subscribable *eventingduckv1.SubscribableSpec, failedSubscriptions map[types.UID]error) eventingduckv1.SubscribableStatus {
-	if subscribable == nil {
-		return eventingduckv1.SubscribableStatus{}
-	}
-	subscriberStatus := make([]eventingduckv1.SubscriberStatus, 0)
-	for _, sub := range subscribable.Subscribers {
-		status := eventingduckv1.SubscriberStatus{
-			UID:                sub.UID,
-			ObservedGeneration: sub.Generation,
-			Ready:              corev1.ConditionTrue,
-		}
-		if err, ok := failedSubscriptions[sub.UID]; ok {
-			status.Ready = corev1.ConditionFalse
-			status.Message = err.Error()
-		}
-		subscriberStatus = append(subscriberStatus, status)
-	}
-	return eventingduckv1.SubscribableStatus{
-		Subscribers: subscriberStatus,
-	}
-}
-
 // newConfigFromKafkaChannels creates a new Config from the list of kafka channels.
 func (r *Reconciler) newChannelConfigFromKafkaChannel(c *v1beta1.KafkaChannel) *dispatcher.ChannelConfig {
 	channelConfig := dispatcher.ChannelConfig{
diff --git a/pkg/channel/distributed/common/config/configwatcher.go b/pkg/channel/distributed/common/config/configwatcher.go
index 3cedada30..f159abd59 100644
--- a/pkg/channel/distributed/common/config/configwatcher.go
+++ b/pkg/channel/distributed/common/config/configwatcher.go
@@ -26,7 +26,6 @@ import (
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/injection/sharedmain"
-	"knative.dev/pkg/system"
 )
 
 // The EventingKafkaConfig and these EK sub-structs contain our custom configuration settings,
@@ -75,7 +74,7 @@ type EventingKafkaConfig struct {
 // Initialize The Specified Context With A ConfigMap Watcher
 // Much Of This Function Is Taken From The knative.dev sharedmain Package
 //
-func InitializeConfigWatcher(ctx context.Context, logger *zap.SugaredLogger, handler configmap.Observer) error {
+func InitializeConfigWatcher(ctx context.Context, logger *zap.SugaredLogger, handler configmap.Observer, namespace string) error {
 
 	// Create A Watcher On The Configuration Settings ConfigMap & Dynamically Update Configuration
 	// Since this is designed to be called by the main() function, the default KNative package behavior here
@@ -84,7 +83,7 @@ func InitializeConfigWatcher(ctx context.Context, logger *zap.SugaredLogger, han
 
 	// Start The ConfigMap Watcher
 	// Taken from knative.dev/pkg/injection/sharedmain/main.go::WatchObservabilityConfigOrDie
-	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, SettingsConfigMapName, metav1.GetOptions{}); err == nil {
+	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(namespace).Get(ctx, SettingsConfigMapName, metav1.GetOptions{}); err == nil {
 		watcher.Watch(SettingsConfigMapName, handler)
 	} else if !apierrors.IsNotFound(err) {
 		logger.Error("Error reading ConfigMap "+SettingsConfigMapName, zap.Error(err))
diff --git a/pkg/channel/distributed/common/config/observability.go b/pkg/channel/distributed/common/config/observability.go
index c6ed5de50..6cb5bc521 100644
--- a/pkg/channel/distributed/common/config/observability.go
+++ b/pkg/channel/distributed/common/config/observability.go
@@ -30,14 +30,13 @@ import (
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/profiling"
-	"knative.dev/pkg/system"
 )
 
 //
 // Initialize The Specified Context With A Profiling Server (ConfigMap Watcher And HTTP Endpoint)
 // Much Of This Function Is Taken From The knative.dev sharedmain Package
 //
-func InitializeObservability(ctx context.Context, logger *zap.SugaredLogger, metricsDomain string, metricsPort int) error {
+func InitializeObservability(ctx context.Context, logger *zap.SugaredLogger, metricsDomain string, metricsPort int, namespace string) error {
 
 	// Initialize the profiling server
 	// Taken from knative.dev/pkg/injection/sharedmain/main.go::MainWithConfig
@@ -71,7 +70,7 @@ func InitializeObservability(ctx context.Context, logger *zap.SugaredLogger, met
 	// Start The Observability ConfigMap Watcher
 	// Taken from knative.dev/pkg/injection/sharedmain/main.go::WatchObservabilityConfigOrDie
 	// and knative.dev/pkg/metrics/exporter.go::ConfigMapWatcher
-	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, metrics.ConfigMapName(),
+	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(namespace).Get(ctx, metrics.ConfigMapName(),
 		metav1.GetOptions{}); err == nil {
 		cmw.Watch(metrics.ConfigMapName(),
 			func(configMap *corev1.ConfigMap) {
diff --git a/pkg/channel/distributed/common/config/tracing.go b/pkg/channel/distributed/common/config/tracing.go
index cb56e9c25..112f2d959 100644
--- a/pkg/channel/distributed/common/config/tracing.go
+++ b/pkg/channel/distributed/common/config/tracing.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/client-go/kubernetes"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
-	"knative.dev/pkg/system"
 	"knative.dev/pkg/tracing"
 	tracingconfig "knative.dev/pkg/tracing/config"
 )
@@ -33,7 +32,7 @@ import (
 // Initialize The Specified Context With A Tracer (ConfigMap Watcher)
 // Assumes ctx Has K8S Client Injected Via LoggingContext (Or Similar)
 //
-func InitializeTracing(logger *zap.SugaredLogger, ctx context.Context, service string) error {
+func InitializeTracing(logger *zap.SugaredLogger, ctx context.Context, service string, namespace string) error {
 	k8sClient, ok := ctx.Value(injectionclient.Key{}).(kubernetes.Interface)
 	if !ok {
 		err := errors.New("error getting kubernetes client from context (invalid interface)")
@@ -42,7 +41,7 @@ func InitializeTracing(logger *zap.SugaredLogger, ctx context.Context, service s
 	}
 
 	// Create A Watcher On The Tracing ConfigMap & Dynamically Update Tracing Configuration
-	cmw := configmap.NewInformedWatcher(k8sClient, system.Namespace())
+	cmw := configmap.NewInformedWatcher(k8sClient, namespace)
 
 	if err := tracing.SetupDynamicPublishing(logger, cmw, service, tracingconfig.ConfigName); err != nil {
 		logger.Error("Error setting up dynamic trace publishing", zap.Error(err))
diff --git a/pkg/channel/distributed/common/kafka/admin/admin.go b/pkg/channel/distributed/common/kafka/admin/admin.go
index 98f527136..8092e067c 100644
--- a/pkg/channel/distributed/common/kafka/admin/admin.go
+++ b/pkg/channel/distributed/common/kafka/admin/admin.go
@@ -22,7 +22,7 @@ import (
 	"fmt"
 
 	"github.com/Shopify/sarama"
-	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/pkg/system"
 )
 
 // Sarama ClusterAdmin Wrapping Interface To Facilitate Other Implementations (e.g. Azure EventHubs)
@@ -69,11 +69,11 @@ const (
 func CreateAdminClient(ctx context.Context, saramaConfig *sarama.Config, clientId string, adminClientType AdminClientType) (AdminClientInterface, error) {
 	switch adminClientType {
 	case Kafka:
-		return NewKafkaAdminClientWrapper(ctx, saramaConfig, clientId, constants.KnativeEventingNamespace)
+		return NewKafkaAdminClientWrapper(ctx, saramaConfig, clientId, system.Namespace())
 	case EventHub:
-		return NewEventHubAdminClientWrapper(ctx, constants.KnativeEventingNamespace)
+		return NewEventHubAdminClientWrapper(ctx, system.Namespace())
 	case Custom:
-		return NewCustomAdminClientWrapper(ctx, constants.KnativeEventingNamespace)
+		return NewCustomAdminClientWrapper(ctx, system.Namespace())
 	case Unknown:
 		return nil, errors.New("received unknown AdminClientType") // Should Never Happen But...
 	default:
diff --git a/pkg/channel/distributed/controller/env/environment.go b/pkg/channel/distributed/controller/env/environment.go
index 470dea79b..8cd27db64 100644
--- a/pkg/channel/distributed/controller/env/environment.go
+++ b/pkg/channel/distributed/controller/env/environment.go
@@ -17,9 +17,13 @@ limitations under the License.
 package env
 
 import (
+	"context"
+	"errors"
 	"strconv"
 	"time"
 
+	"knative.dev/pkg/system"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 	"knative.dev/pkg/controller"
@@ -39,10 +43,11 @@ const (
 type Environment struct {
 
 	// Eventing-kafka Configuration
-	ServiceAccount string        // Required
-	MetricsPort    int           // Required
-	MetricsDomain  string        // Required
-	ResyncPeriod   time.Duration // Optional
+	SystemNamespace string        // Required
+	ServiceAccount  string        // Required
+	MetricsPort     int           // Required
+	MetricsDomain   string        // Required
+	ResyncPeriod    time.Duration // Optional
 
 	// Dispatcher Configuration
 	DispatcherImage string // Required
@@ -51,6 +56,9 @@ type Environment struct {
 	ReceiverImage string // Required
 }
 
+// Key is used as the key for associating information with a context.Context.
+type Key struct{}
+
 // Get The Environment
 func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 
@@ -60,6 +68,12 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	// The ControllerConfig Reference
 	environment := &Environment{}
 
+	// Get The Required System Namespace Config Value
+	environment.SystemNamespace, err = env.GetRequiredConfigValue(logger, system.NamespaceEnvKey)
+	if err != nil {
+		return nil, err
+	}
+
 	// Get The Required K8S ServiceAccount Config Value
 	environment.ServiceAccount, err = env.GetRequiredConfigValue(logger, env.ServiceAccountEnvVarKey)
 	if err != nil {
@@ -115,3 +129,12 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	// Return The Populated ControllerConfig
 	return environment, nil
 }
+
+// Obtain the Environment struct contained in a Context, if present
+func FromContext(ctx context.Context) (*Environment, error) {
+	environment, ok := ctx.Value(Key{}).(*Environment)
+	if !ok {
+		return nil, errors.New("could not extract Environment from context")
+	}
+	return environment, nil
+}
diff --git a/pkg/channel/distributed/controller/kafkachannel/channel.go b/pkg/channel/distributed/controller/kafkachannel/channel.go
index 19e76819e..0f1c5c5e1 100644
--- a/pkg/channel/distributed/controller/kafkachannel/channel.go
+++ b/pkg/channel/distributed/controller/kafkachannel/channel.go
@@ -29,7 +29,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/network"
@@ -130,7 +129,7 @@ func (r *Reconciler) newKafkaChannelService(channel *kafkav1beta1.KafkaChannel)
 
 	// Get The Receiver Service Name For The Kafka Secret (One Receiver Service Per Kafka Secret)
 	deploymentName := util.ReceiverDnsSafeName(r.kafkaSecretName(channel))
-	serviceAddress := network.GetServiceHostname(deploymentName, commonconstants.KnativeEventingNamespace)
+	serviceAddress := network.GetServiceHostname(deploymentName, r.environment.SystemNamespace)
 
 	// Create & Return The Service Model
 	return &corev1.Service{
diff --git a/pkg/channel/distributed/controller/kafkachannel/controller.go b/pkg/channel/distributed/controller/kafkachannel/controller.go
index e31a1c4fd..d9e45f46c 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller.go
@@ -55,9 +55,9 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	serviceInformer := service.Get(ctx)
 
 	// Load The Environment Variables
-	environment, err := env.GetEnvironment(logger)
+	environment, err := env.FromContext(ctx)
 	if err != nil {
-		logger.Fatal("Failed To Load Environment Variables - Terminating!", zap.Error(err))
+		logger.Fatal("Failed To Get Environment From Context - Terminating!", zap.Error(err))
 	}
 
 	// Load the Sarama and other eventing-kafka settings from our configmap
@@ -102,7 +102,7 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	}
 
 	// Watch The Settings ConfigMap For Changes
-	err = commonconfig.InitializeConfigWatcher(ctx, logger.Sugar(), rec.configMapObserver)
+	err = commonconfig.InitializeConfigWatcher(ctx, logger.Sugar(), rec.configMapObserver, environment.SystemNamespace)
 	if err != nil {
 		logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
 	}
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index 856300321..5596efb78 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -34,7 +34,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
@@ -208,7 +207,7 @@ func (r *Reconciler) getDispatcherService(channel *kafkav1beta1.KafkaChannel) (*
 	serviceName := util.DispatcherDnsSafeName(channel)
 
 	// Get The Service By Namespace / Name
-	service, err := r.serviceLister.Services(commonconstants.KnativeEventingNamespace).Get(serviceName)
+	service, err := r.serviceLister.Services(r.environment.SystemNamespace).Get(serviceName)
 
 	// Return The Results
 	return service, err
@@ -228,7 +227,7 @@ func (r *Reconciler) newDispatcherService(channel *kafkav1beta1.KafkaChannel) *c
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      serviceName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: r.environment.SystemNamespace,
 			Labels: map[string]string{
 				constants.KafkaChannelDispatcherLabel:   "true",                                  // Identifies the Service as being a KafkaChannel "Dispatcher"
 				constants.KafkaChannelNameLabel:         channel.Name,                            // Identifies the Service's Owning KafkaChannel's Name
@@ -361,7 +360,7 @@ func (r *Reconciler) getDispatcherDeployment(channel *kafkav1beta1.KafkaChannel)
 	deploymentName := util.DispatcherDnsSafeName(channel)
 
 	// Get The Dispatcher Deployment By Namespace / Name
-	deployment, err := r.deploymentLister.Deployments(commonconstants.KnativeEventingNamespace).Get(deploymentName)
+	deployment, err := r.deploymentLister.Deployments(r.environment.SystemNamespace).Get(deploymentName)
 
 	// Return The Results
 	return deployment, err
@@ -391,7 +390,7 @@ func (r *Reconciler) newDispatcherDeployment(logger *zap.Logger, channel *kafkav
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      deploymentName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: r.environment.SystemNamespace,
 			Labels: map[string]string{
 				constants.AppLabel:                    deploymentName,    // Matches K8S Service Selector Key/Value Below
 				constants.KafkaChannelDispatcherLabel: "true",            // Identifies the Deployment as being a KafkaChannel "Dispatcher"
@@ -474,7 +473,7 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1beta1.KafkaChan
 	envVars := []corev1.EnvVar{
 		{
 			Name:  system.NamespaceEnvKey,
-			Value: commonconstants.KnativeEventingNamespace,
+			Value: r.environment.SystemNamespace,
 		},
 		{
 			Name: commonenv.PodNameEnvVarKey,
diff --git a/pkg/channel/distributed/controller/kafkasecret/controller.go b/pkg/channel/distributed/controller/kafkasecret/controller.go
index 52fb00f54..9276a1b85 100644
--- a/pkg/channel/distributed/controller/kafkasecret/controller.go
+++ b/pkg/channel/distributed/controller/kafkasecret/controller.go
@@ -32,7 +32,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/kafkasecretinjection"
 	injectionclient "knative.dev/eventing-kafka/pkg/client/injection/client"
 	"knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/service"
@@ -54,9 +53,9 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	serviceInformer := service.Get(ctx)
 
 	// Load The Environment Variables
-	environment, err := env.GetEnvironment(logger)
+	environment, err := env.FromContext(ctx)
 	if err != nil {
-		logger.Fatal("Failed To Load Environment Variables - Terminating!", zap.Error(err))
+		logger.Fatal("Failed To Get Environment From Context - Terminating!", zap.Error(err))
 	}
 
 	// Load the Sarama and other eventing-kafka settings from our configmap
@@ -100,7 +99,7 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 		Handler:    controller.HandleAll(controllerImpl.EnqueueControllerOf),
 	})
 	kafkachannelInformer.Informer().AddEventHandler(
-		controller.HandleAll(enqueueSecretOfKafkaChannel(controllerImpl)),
+		controller.HandleAll(enqueueSecretOfKafkaChannel(controllerImpl, environment.SystemNamespace)),
 	)
 
 	// Return The KafkaSecret Controller Impl
@@ -113,7 +112,7 @@ func Shutdown() {
 }
 
 // Enqueue The Kafka Secret Associated With The Specified KafkaChannel
-func enqueueSecretOfKafkaChannel(controller *controller.Impl) func(obj interface{}) {
+func enqueueSecretOfKafkaChannel(controller *controller.Impl, namespace string) func(obj interface{}) {
 	return func(obj interface{}) {
 		if object, ok := obj.(metav1.Object); ok {
 			labels := object.GetLabels()
@@ -121,7 +120,7 @@ func enqueueSecretOfKafkaChannel(controller *controller.Impl) func(obj interface
 				secretName := labels[constants.KafkaSecretLabel]
 				if len(secretName) > 0 {
 					controller.EnqueueKey(types.NamespacedName{
-						Namespace: commonconstants.KnativeEventingNamespace,
+						Namespace: namespace,
 						Name:      secretName,
 					})
 				}
diff --git a/pkg/channel/distributed/controller/kafkasecret/receiver.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
index 65bfec161..50c868a10 100644
--- a/pkg/channel/distributed/controller/kafkasecret/receiver.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -33,7 +33,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
@@ -135,7 +134,7 @@ func (r *Reconciler) getReceiverService(secret *corev1.Secret) (*corev1.Service,
 	deploymentName := util.ReceiverDnsSafeName(secret.Name)
 
 	// Get The Receiver Service By Namespace / Name
-	service, err := r.serviceLister.Services(commonconstants.KnativeEventingNamespace).Get(deploymentName)
+	service, err := r.serviceLister.Services(r.environment.SystemNamespace).Get(deploymentName)
 
 	// Return The Results
 	return service, err
@@ -155,7 +154,7 @@ func (r *Reconciler) newReceiverService(secret *corev1.Secret) *corev1.Service {
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      deploymentName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: r.environment.SystemNamespace,
 			Labels: map[string]string{
 				constants.KafkaChannelReceiverLabel:  "true",                               // Allows for identification of Receivers
 				constants.K8sAppChannelSelectorLabel: constants.K8sAppChannelSelectorValue, // Prometheus ServiceMonitor
@@ -241,7 +240,7 @@ func (r *Reconciler) getReceiverDeployment(secret *corev1.Secret) (*appsv1.Deplo
 	deploymentName := util.ReceiverDnsSafeName(secret.Name)
 
 	// Get The Receiver Deployment By Namespace / Name
-	deployment, err := r.deploymentLister.Deployments(commonconstants.KnativeEventingNamespace).Get(deploymentName)
+	deployment, err := r.deploymentLister.Deployments(r.environment.SystemNamespace).Get(deploymentName)
 
 	// Return The Results
 	return deployment, err
@@ -271,7 +270,7 @@ func (r *Reconciler) newReceiverDeployment(logger *zap.Logger, secret *corev1.Se
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      deploymentName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: r.environment.SystemNamespace,
 			Labels: map[string]string{
 				constants.AppLabel:                  deploymentName, // Matches Service Selector Key/Value Below
 				constants.KafkaChannelReceiverLabel: "true",         // Allows for identification of Receivers
@@ -355,7 +354,7 @@ func (r *Reconciler) receiverDeploymentEnvVars(secret *corev1.Secret) ([]corev1.
 	envVars := []corev1.EnvVar{
 		{
 			Name:  system.NamespaceEnvKey,
-			Value: commonconstants.KnativeEventingNamespace,
+			Value: r.environment.SystemNamespace,
 		},
 		{
 			Name: commonenv.PodNameEnvVarKey,
diff --git a/pkg/channel/distributed/dispatcher/env/environment.go b/pkg/channel/distributed/dispatcher/env/environment.go
index 8b295d3fc..27f1d5562 100644
--- a/pkg/channel/distributed/dispatcher/env/environment.go
+++ b/pkg/channel/distributed/dispatcher/env/environment.go
@@ -20,6 +20,8 @@ import (
 	"strconv"
 	"time"
 
+	"knative.dev/pkg/system"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 	"knative.dev/pkg/controller"
@@ -28,6 +30,9 @@ import (
 // Environment Structure
 type Environment struct {
 
+	// Eventing Configuration
+	SystemNamespace string // Required
+
 	// Metrics Configuration
 	MetricsPort   int    // Required
 	MetricsDomain string // Required
@@ -60,6 +65,12 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	// The Environment Struct To Be Populated
 	environment := &Environment{}
 
+	// Get The Required System Namespace Config Value
+	environment.SystemNamespace, err = env.GetRequiredConfigValue(logger, system.NamespaceEnvKey)
+	if err != nil {
+		return nil, err
+	}
+
 	// Get The Required Metrics Port Config Value & Convert To Int
 	environment.MetricsPort, err = env.GetRequiredConfigInt(logger, env.MetricsPortEnvVarKey, "MetricsPort")
 	if err != nil {
diff --git a/pkg/channel/distributed/receiver/env/environment.go b/pkg/channel/distributed/receiver/env/environment.go
index b39335118..73f87eb08 100644
--- a/pkg/channel/distributed/receiver/env/environment.go
+++ b/pkg/channel/distributed/receiver/env/environment.go
@@ -20,6 +20,8 @@ import (
 	"strconv"
 	"time"
 
+	"knative.dev/pkg/system"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 	"knative.dev/pkg/controller"
@@ -28,6 +30,9 @@ import (
 // The Environment Struct
 type Environment struct {
 
+	// Eventing Configuration
+	SystemNamespace string // Required
+
 	// Metrics Configuration
 	MetricsPort   int    // Required
 	MetricsDomain string // Required
@@ -58,6 +63,12 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	// The Environment Struct To Be Populated
 	environment := &Environment{}
 
+	// Get The Required System Namespace Config Value
+	environment.SystemNamespace, err = env.GetRequiredConfigValue(logger, system.NamespaceEnvKey)
+	if err != nil {
+		return nil, err
+	}
+
 	// Get The Required Metrics Port Config Value & Convert To Int
 	environment.MetricsPort, err = env.GetRequiredConfigInt(logger, env.MetricsPortEnvVarKey, "MetricsPort")
 	if err != nil {
diff --git a/pkg/common/constants/constants.go b/pkg/common/constants/constants.go
index 32fe8b34b..2534edc89 100644
--- a/pkg/common/constants/constants.go
+++ b/pkg/common/constants/constants.go
@@ -21,7 +21,4 @@ const (
 	// KafkaChannel Spec Defaults
 	DefaultNumPartitions     = 1
 	DefaultReplicationFactor = 1
-
-	// Knative Eventing Namespace
-	KnativeEventingNamespace = "knative-eventing"
 )
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index b01addb02..7d6110ff0 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -136,6 +136,10 @@ func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool
 		a.logger.Debug("Error while sending the message", zap.Error(err))
 		return false, err // Error while sending, don't commit offset
 	}
+	// Always try to close body so the connection can be reused afterwards
+	if res.Body != nil {
+		res.Body.Close()
+	}
 
 	if res.StatusCode/100 != 2 {
 		a.logger.Debug("Unexpected status code", zap.Int("status code", res.StatusCode))
diff --git a/pkg/source/reconciler/source/resources/labels.go b/pkg/source/reconciler/source/resources/labels.go
index b0658d674..048b91850 100644
--- a/pkg/source/reconciler/source/resources/labels.go
+++ b/pkg/source/reconciler/source/resources/labels.go
@@ -25,6 +25,6 @@ const (
 func GetLabels(name string) map[string]string {
 	return map[string]string{
 		"eventing.knative.dev/source":     controllerAgentName,
-		"eventing.knative.dev/SourceName": name,
+		"eventing.knative.dev/sourceName": name,
 	}
 }
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 5ea17417b..aa8410391 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1072,7 +1072,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.19.1-0.20201127005336-6065b380266f
+# knative.dev/eventing v0.19.1-0.20201130190736-241af67e1844
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/util/crstatusevent
@@ -1206,7 +1206,7 @@ knative.dev/eventing/test/test_images/recordevents
 # knative.dev/hack v0.0.0-20201125230335-c46a6498e9ed
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20201127013335-0d896b5c87b8
+# knative.dev/pkg v0.0.0-20201130192436-e5346d90e980
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/channel/consolidated/kafka/admin_test.go b/pkg/channel/consolidated/kafka/admin_test.go
new file mode 100644
index 000000000..3e671c22a
--- /dev/null
+++ b/pkg/channel/consolidated/kafka/admin_test.go
@@ -0,0 +1,93 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kafka
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Shopify/sarama"
+	pkgtesting "knative.dev/pkg/logging/testing"
+)
+
+const testCG = "cg1"
+
+var m sync.RWMutex
+
+type FakeClusterAdmin struct {
+	sarama.ClusterAdmin
+	faulty bool
+}
+
+func (f *FakeClusterAdmin) ListConsumerGroups() (map[string]string, error) {
+	cgs := map[string]string{
+		testCG: "cg",
+	}
+	m.RLock()
+	defer m.RUnlock()
+	if f.faulty {
+		return nil, fmt.Errorf("Error")
+	}
+	return cgs, nil
+}
+
+func TestAdminClient(t *testing.T) {
+	var wg sync.WaitGroup
+	wg.Add(10)
+	ctx := pkgtesting.TestContextWithLogger(t)
+	f := &FakeClusterAdmin{}
+	ac, err := NewAdminClient(ctx, func() (sarama.ClusterAdmin, error) {
+		return f, nil
+	})
+	if err != nil {
+		t.Error("failed to obtain new client", err)
+	}
+	for i := 0; i < 10; i += 1 {
+		go func() {
+			doList(t, ac)
+			check := make(chan struct{})
+			go func() {
+				m.Lock()
+				f.faulty = true
+				m.Unlock()
+				check <- struct{}{}
+				time.Sleep(2 * time.Second)
+				m.Lock()
+				f.faulty = false
+				m.Unlock()
+				check <- struct{}{}
+			}()
+			<-check
+			doList(t, ac)
+			<-check
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+}
+
+func doList(t *testing.T, ac AdminClient) {
+	cgs, _ := ac.ListConsumerGroups()
+	if len(cgs) != 1 {
+		t.Fatalf("list consumer group: got %d, want %d", len(cgs), 1)
+	}
+	if cgs[0] != testCG {
+		t.Fatalf("consumer group: got %s, want %s", cgs[0], testCG)
+	}
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher_test.go b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher_test.go
new file mode 100644
index 000000000..e543c01ce
--- /dev/null
+++ b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher_test.go
@@ -0,0 +1,84 @@
+/*
+Copyright 2020 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	pkgtesting "knative.dev/pkg/logging/testing"
+)
+
+//TODO how to mock the sarama AdminClient
+type FakeClusterAdmin struct {
+	mutex sync.RWMutex
+	cgs   sets.String
+}
+
+func (fake *FakeClusterAdmin) ListConsumerGroups() ([]string, error) {
+	fake.mutex.RLock()
+	defer fake.mutex.RUnlock()
+	return fake.cgs.List(), nil
+}
+
+func (fake *FakeClusterAdmin) deleteCG(cg string) {
+	fake.mutex.Lock()
+	defer fake.mutex.Unlock()
+	fake.cgs.Delete(cg)
+}
+
+func TestKafkaWatcher(t *testing.T) {
+	cgname := "kafka.event-example.default-kne-trigger.0d9c4383-1e68-42b5-8c3a-3788274404c5"
+	wid := "channel-abc"
+	cgs := sets.String{}
+	cgs.Insert(cgname)
+	ca := FakeClusterAdmin{
+		cgs: cgs,
+	}
+
+	ch := make(chan sets.String, 1)
+
+	w := NewConsumerGroupWatcher(pkgtesting.TestContextWithLogger(t), &ca, 2*time.Second)
+	w.Watch(wid, func() {
+		cgs := w.List(func(cg string) bool {
+			return cgname == cg
+		})
+		result := sets.String{}
+		result.Insert(cgs...)
+		ch <- result
+	})
+
+	w.Start()
+	<-ch
+	assertSync(t, ch, cgs)
+	ca.deleteCG(cgname)
+	assertSync(t, ch, sets.String{})
+}
+
+func assertSync(t *testing.T, ch chan sets.String, cgs sets.String) {
+	select {
+	case syncedCGs := <-ch:
+		if !syncedCGs.Equal(cgs) {
+			t.Errorf("observed and expected consumer groups do not match. got %v expected %v", syncedCGs, cgs)
+		}
+	case <-time.After(6 * time.Second):
+		t.Errorf("timedout waiting for consumer groups to sync")
+	}
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
index b58dc384e..0bccf4490 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
@@ -20,6 +20,10 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
+
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
 
 	"github.com/Shopify/sarama"
 
@@ -58,6 +62,8 @@ const (
 	channelServiceAddress = "test-kc-kn-channel.test-namespace.svc.cluster.local"
 	brokerName            = "test-broker"
 	finalizerName         = "kafkachannels.messaging.knative.dev"
+	sub1UID               = "2f9b5e8e-deb6-11e8-9f32-f2801f1b9fd1"
+	sub2UID               = "34c5aec8-deb6-11e8-9f32-f2801f1b9fd1"
 )
 
 var (
@@ -235,12 +241,14 @@ func TestAllCases(t *testing.T) {
 				makeService(),
 				makeReadyEndpoints(),
 				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaChannelSubscribers(subscribers()),
 					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 				makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 			},
 			WantErr: false,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaChannelSubscribers(subscribers()),
 					reconcilertesting.WithInitKafkaChannelConditions,
 					reconcilertesting.WithKafkaFinalizer(finalizerName),
 					reconcilertesting.WithKafkaChannelConfigReady(),
@@ -326,7 +334,8 @@ func TestAllCases(t *testing.T) {
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
-			kafkachannelLister: listers.GetKafkaChannelLister(),
+			consumerGroupWatcher: NewConsumerGroupWatcher(ctx, &FakeClusterAdmin{}, 100*time.Millisecond),
+			kafkachannelLister:   listers.GetKafkaChannelLister(),
 			// TODO fix
 			kafkachannelInformer: nil,
 			deploymentLister:     listers.GetDeploymentLister(),
@@ -383,7 +392,8 @@ func TestTopicExists(t *testing.T) {
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
-			kafkachannelLister: listers.GetKafkaChannelLister(),
+			consumerGroupWatcher: NewConsumerGroupWatcher(ctx, &FakeClusterAdmin{}, 100*time.Millisecond),
+			kafkachannelLister:   listers.GetKafkaChannelLister(),
 			// TODO fix
 			kafkachannelInformer: nil,
 			deploymentLister:     listers.GetDeploymentLister(),
@@ -452,7 +462,8 @@ func TestDeploymentUpdatedOnImageChange(t *testing.T) {
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
-			kafkachannelLister: listers.GetKafkaChannelLister(),
+			consumerGroupWatcher: NewConsumerGroupWatcher(ctx, &FakeClusterAdmin{}, 100*time.Millisecond),
+			kafkachannelLister:   listers.GetKafkaChannelLister(),
 			// TODO fix
 			kafkachannelInformer: nil,
 			deploymentLister:     listers.GetDeploymentLister(),
@@ -521,7 +532,8 @@ func TestDeploymentZeroReplicas(t *testing.T) {
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
-			kafkachannelLister: listers.GetKafkaChannelLister(),
+			consumerGroupWatcher: NewConsumerGroupWatcher(ctx, &FakeClusterAdmin{}, 100*time.Millisecond),
+			kafkachannelLister:   listers.GetKafkaChannelLister(),
 			// TODO fix
 			kafkachannelInformer: nil,
 			deploymentLister:     listers.GetDeploymentLister(),
@@ -587,7 +599,8 @@ func TestDeploymentMoreThanOneReplicas(t *testing.T) {
 			kafkaConfig: &KafkaConfig{
 				Brokers: []string{brokerName},
 			},
-			kafkachannelLister: listers.GetKafkaChannelLister(),
+			consumerGroupWatcher: NewConsumerGroupWatcher(ctx, &FakeClusterAdmin{}, 100*time.Millisecond),
+			kafkachannelLister:   listers.GetKafkaChannelLister(),
 			// TODO fix
 			kafkachannelInformer: nil,
 			deploymentLister:     listers.GetDeploymentLister(),
@@ -682,7 +695,11 @@ func (ca *mockClusterAdmin) DeleteACL(filter sarama.AclFilter, validateOnly bool
 }
 
 func (ca *mockClusterAdmin) ListConsumerGroups() (map[string]string, error) {
-	return nil, nil
+	cgs := map[string]string{
+		fmt.Sprintf("kafka.%s.%s.%s", kcName, testNS, sub1UID): "consumer",
+		fmt.Sprintf("kafka.%s.%s.%s", kcName, testNS, sub2UID): "consumer",
+	}
+	return cgs, nil
 }
 
 func (ca *mockClusterAdmin) DescribeConsumerGroups(groups []string) ([]*sarama.GroupDescription, error) {
@@ -796,3 +813,18 @@ func patchFinalizers(namespace, name string) clientgotesting.PatchActionImpl {
 	action.Patch = []byte(patch)
 	return action
 }
+
+func subscribers() []eventingduckv1.SubscriberSpec {
+
+	return []eventingduckv1.SubscriberSpec{{
+		UID:           sub1UID,
+		Generation:    1,
+		SubscriberURI: apis.HTTP("call1"),
+		ReplyURI:      apis.HTTP("sink2"),
+	}, {
+		UID:           sub2UID,
+		Generation:    2,
+		SubscriberURI: apis.HTTP("call2"),
+		ReplyURI:      apis.HTTP("sink2"),
+	}}
+}
diff --git a/pkg/channel/consolidated/reconciler/testing/kafkachannel.go b/pkg/channel/consolidated/reconciler/testing/kafkachannel.go
index 5296a84aa..413d7736b 100644
--- a/pkg/channel/consolidated/reconciler/testing/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/testing/kafkachannel.go
@@ -20,6 +20,10 @@ import (
 	"context"
 	"time"
 
+	"k8s.io/apimachinery/pkg/types"
+
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -38,6 +42,7 @@ func NewKafkaChannel(name, namespace string, ncopt ...KafkaChannelOption) *v1bet
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
+			UID:       types.UID("abc-xyz"),
 		},
 		Spec: v1beta1.KafkaChannelSpec{},
 	}
@@ -133,3 +138,23 @@ func WithKafkaFinalizer(finalizerName string) KafkaChannelOption {
 		nc.SetFinalizers(finalizers.List())
 	}
 }
+
+func WithKafkaChannelSubscribers(subs []v1.SubscriberSpec) KafkaChannelOption {
+	return func(nc *v1beta1.KafkaChannel) {
+		nc.Spec.Subscribers = subs
+	}
+}
+
+func WithKafkaChannelStatusSubscribers() KafkaChannelOption {
+	return func(nc *v1beta1.KafkaChannel) {
+		ss := make([]v1.SubscriberStatus, len(nc.Spec.Subscribers))
+		for _, s := range nc.Spec.Subscribers {
+			ss = append(ss, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionTrue,
+			})
+		}
+		nc.Status.Subscribers = ss
+	}
+}
diff --git a/pkg/channel/distributed/common/config/configwatcher_test.go b/pkg/channel/distributed/common/config/configwatcher_test.go
index 058b3ab3b..f648e980c 100644
--- a/pkg/channel/distributed/common/config/configwatcher_test.go
+++ b/pkg/channel/distributed/common/config/configwatcher_test.go
@@ -18,7 +18,6 @@ package config
 
 import (
 	"context"
-	"os"
 	"sync"
 	"testing"
 	"time"
@@ -28,7 +27,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/system"
@@ -43,14 +41,11 @@ var (
 // Test The InitializeObservability() Functionality
 func TestInitializeConfigWatcher(t *testing.T) {
 
-	// Test Data
-	ctx := context.TODO()
-
 	// Obtain a Test Logger (Required By be InitializeConfigWatcher function)
 	logger := logtesting.TestLogger(t)
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 
 	// Create A Test Observability ConfigMap For The InitializeObservability() Call To Watch
 	configMap := commontesting.GetTestSaramaConfigMap(commontesting.OldSaramaConfig, commontesting.TestEKConfig)
@@ -59,7 +54,7 @@ func TestInitializeConfigWatcher(t *testing.T) {
 	fakeK8sClient := fake.NewSimpleClientset(configMap)
 
 	// Add The Fake K8S Client To The Context (Required By InitializeObservability)
-	ctx = context.WithValue(ctx, injectionclient.Key{}, fakeK8sClient)
+	ctx := context.WithValue(context.TODO(), injectionclient.Key{}, fakeK8sClient)
 
 	// The configWatcherHandler should change the nil "watchedConfigMap" to a valid ConfigMap when the watcher triggers
 
@@ -68,7 +63,7 @@ func TestInitializeConfigWatcher(t *testing.T) {
 	assert.Equal(t, testConfigMap.Data["sarama"], commontesting.OldSaramaConfig)
 
 	// Perform The Test (Initialize The Observability Watcher)
-	err = InitializeConfigWatcher(ctx, logger, configWatcherHandler)
+	err = InitializeConfigWatcher(ctx, logger, configWatcherHandler, system.Namespace())
 	assert.Nil(t, err)
 
 	// Wait for the configWatcherHandler to be called (happens pretty quickly; loop usually only runs once)
diff --git a/pkg/channel/distributed/common/config/observability_test.go b/pkg/channel/distributed/common/config/observability_test.go
index 9d588493f..13828b8a1 100644
--- a/pkg/channel/distributed/common/config/observability_test.go
+++ b/pkg/channel/distributed/common/config/observability_test.go
@@ -29,7 +29,6 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/metrics"
@@ -48,7 +47,7 @@ func TestInitializeObservability(t *testing.T) {
 	logger := logtesting.TestLogger(t)
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	assert.Nil(t, os.Setenv(env.MetricsDomainEnvVarKey, metricsDomain))
 
 	// Create A Test Observability ConfigMap For The InitializeObservability() Call To Watch
@@ -74,7 +73,7 @@ func TestInitializeObservability(t *testing.T) {
 	ctx = context.WithValue(ctx, injectionclient.Key{}, fakeK8sClient)
 
 	// Perform The Test (Initialize The Observability Watcher)
-	err := InitializeObservability(ctx, logger, metricsDomain, metricsPort)
+	err := InitializeObservability(ctx, logger, metricsDomain, metricsPort, system.Namespace())
 	assert.Nil(t, err)
 
 	// Verify that the profiling endpoint exists and responds to requests
diff --git a/pkg/channel/distributed/common/config/tracing_test.go b/pkg/channel/distributed/common/config/tracing_test.go
index 883dd3ca9..1ab2c1668 100644
--- a/pkg/channel/distributed/common/config/tracing_test.go
+++ b/pkg/channel/distributed/common/config/tracing_test.go
@@ -18,14 +18,14 @@ package config
 
 import (
 	"context"
-	"os"
 	"testing"
 
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/system"
@@ -43,7 +43,7 @@ func TestInitializeTracing(t *testing.T) {
 	logger := logtesting.TestLogger(t)
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 
 	// Create A Test Tracing ConfigMap For The SetupDynamicPublishing() Call To Watch
 	tracingConfigMap := &corev1.ConfigMap{
@@ -64,7 +64,7 @@ func TestInitializeTracing(t *testing.T) {
 	ctx = context.WithValue(ctx, injectionclient.Key{}, fakeK8sClient)
 
 	// Perform The Test (Initialize The Tracing Watcher)
-	err := InitializeTracing(logger, ctx, service)
+	err := InitializeTracing(logger, ctx, service, system.Namespace())
 	assert.Nil(t, err)
 
 	// If the InitializeTracing Succeeds, it will not fatally exit
@@ -73,16 +73,16 @@ func TestInitializeTracing(t *testing.T) {
 }
 
 func TestInitializeTracing_Failure(t *testing.T) {
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	ctx := context.TODO()
 
 	// If there is no Kubernetes client in the context, the server will not start
-	err := InitializeTracing(logtesting.TestLogger(t), ctx, "TestService")
+	err := InitializeTracing(logtesting.TestLogger(t), ctx, "TestService", system.Namespace())
 	assert.NotNil(t, err)
 
 	// If there is no "config-tracing" configmap, the server will not start
 	fakeK8sClient := fake.NewSimpleClientset()
 	ctx = context.WithValue(ctx, injectionclient.Key{}, fakeK8sClient)
-	err = InitializeTracing(logtesting.TestLogger(t), ctx, "TestService")
+	err = InitializeTracing(logtesting.TestLogger(t), ctx, "TestService", system.Namespace())
 	assert.NotNil(t, err)
 }
diff --git a/pkg/channel/distributed/common/k8s/context_test.go b/pkg/channel/distributed/common/k8s/context_test.go
index 09e2e0f8d..cfd39e8b8 100644
--- a/pkg/channel/distributed/common/k8s/context_test.go
+++ b/pkg/channel/distributed/common/k8s/context_test.go
@@ -18,17 +18,16 @@ package k8s
 
 import (
 	"context"
-	"os"
 	"testing"
 	"time"
 
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
@@ -42,8 +41,7 @@ func TestLoggingContext(t *testing.T) {
 	component := "TestComponent"
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
-	assert.Nil(t, os.Setenv(env.KnativeLoggingConfigMapNameEnvVarKey, logging.ConfigMapName()))
+	commontesting.SetTestEnvironment(t)
 
 	// Create A Test Logging ConfigMap
 	loggingConfigMap := &corev1.ConfigMap{
diff --git a/pkg/channel/distributed/common/kafka/admin/admin_kafka_test.go b/pkg/channel/distributed/common/kafka/admin/admin_kafka_test.go
index 608339640..3a8ec0dcf 100644
--- a/pkg/channel/distributed/common/kafka/admin/admin_kafka_test.go
+++ b/pkg/channel/distributed/common/kafka/admin/admin_kafka_test.go
@@ -18,7 +18,6 @@ package admin
 
 import (
 	"context"
-	"os"
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
@@ -29,7 +28,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
@@ -62,7 +60,7 @@ Metadata:
 `
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 
 	// Create Test Kafka Secret And ConfigMap
 	kafkaSecret := createKafkaSecret(kafkaSecretName, namespace, kafkaSecretBrokers, kafkaSecretUsername, kafkaSecretPassword)
diff --git a/pkg/channel/distributed/common/kafka/admin/admin_test.go b/pkg/channel/distributed/common/kafka/admin/admin_test.go
index 7e6ed9865..0c96cf81d 100644
--- a/pkg/channel/distributed/common/kafka/admin/admin_test.go
+++ b/pkg/channel/distributed/common/kafka/admin/admin_test.go
@@ -20,10 +20,13 @@ import (
 	"context"
 	"testing"
 
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
+
+	"knative.dev/pkg/system"
+
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 )
 
 // Mock AdminClient Reference
@@ -33,7 +36,8 @@ var mockAdminClient AdminClientInterface
 func TestCreateAdminClientKafka(t *testing.T) {
 
 	// Test Data
-	ctx := context.TODO()
+	commontesting.SetTestEnvironment(t)
+	ctx := context.WithValue(context.TODO(), env.Key{}, &env.Environment{SystemNamespace: system.Namespace()})
 	clientId := "TestClientId"
 	adminClientType := Kafka
 	mockAdminClient = &MockAdminClient{}
@@ -43,7 +47,7 @@ func TestCreateAdminClientKafka(t *testing.T) {
 	NewKafkaAdminClientWrapper = func(ctxArg context.Context, saramaConfig *sarama.Config, clientIdArg string, namespaceArg string) (AdminClientInterface, error) {
 		assert.Equal(t, ctx, ctxArg)
 		assert.Equal(t, clientId, clientIdArg)
-		assert.Equal(t, constants.KnativeEventingNamespace, namespaceArg)
+		assert.Equal(t, system.Namespace(), namespaceArg)
 		assert.Equal(t, adminClientType, adminClientType)
 		return mockAdminClient, nil
 	}
@@ -62,7 +66,8 @@ func TestCreateAdminClientKafka(t *testing.T) {
 func TestCreateAdminClientEventHub(t *testing.T) {
 
 	// Test Data
-	ctx := context.TODO()
+	commontesting.SetTestEnvironment(t)
+	ctx := context.WithValue(context.TODO(), env.Key{}, &env.Environment{SystemNamespace: system.Namespace()})
 	clientId := "TestClientId"
 	adminClientType := EventHub
 	mockAdminClient = &MockAdminClient{}
@@ -71,7 +76,7 @@ func TestCreateAdminClientEventHub(t *testing.T) {
 	NewEventHubAdminClientWrapperRef := NewEventHubAdminClientWrapper
 	NewEventHubAdminClientWrapper = func(ctxArg context.Context, namespaceArg string) (AdminClientInterface, error) {
 		assert.Equal(t, ctx, ctxArg)
-		assert.Equal(t, constants.KnativeEventingNamespace, namespaceArg)
+		assert.Equal(t, system.Namespace(), namespaceArg)
 		assert.Equal(t, adminClientType, adminClientType)
 		return mockAdminClient, nil
 	}
@@ -90,7 +95,8 @@ func TestCreateAdminClientEventHub(t *testing.T) {
 func TestCreateAdminClientCustom(t *testing.T) {
 
 	// Test Data
-	ctx := context.TODO()
+	commontesting.SetTestEnvironment(t)
+	ctx := context.WithValue(context.TODO(), env.Key{}, &env.Environment{SystemNamespace: system.Namespace()})
 	clientId := "TestClientId"
 	adminClientType := Custom
 	mockAdminClient = &MockAdminClient{}
@@ -99,7 +105,7 @@ func TestCreateAdminClientCustom(t *testing.T) {
 	NewCustomAdminClientWrapperRef := NewCustomAdminClientWrapper
 	NewCustomAdminClientWrapper = func(ctxArg context.Context, namespaceArg string) (AdminClientInterface, error) {
 		assert.Equal(t, ctx, ctxArg)
-		assert.Equal(t, constants.KnativeEventingNamespace, namespaceArg)
+		assert.Equal(t, system.Namespace(), namespaceArg)
 		assert.Equal(t, adminClientType, adminClientType)
 		return mockAdminClient, nil
 	}
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index 65b0f9617..a2710bb40 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"os"
 	"regexp"
 	"strconv"
 	"strings"
@@ -33,9 +32,7 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/system"
 )
 
 const (
@@ -199,7 +196,7 @@ func TestUpdateSaramaConfig(t *testing.T) {
 // This test is specifically to validate that our default settings (used in 200-eventing-kafka-configmap.yaml)
 // are valid.  If the defaults in the file change, change this test to match for verification purposes.
 func TestLoadDefaultSaramaSettings(t *testing.T) {
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	configMap := commontesting.GetTestSaramaConfigMap(EKDefaultSaramaConfig, EKDefaultConfigYaml)
 	fakeK8sClient := fake.NewSimpleClientset(configMap)
 	ctx := context.WithValue(context.Background(), injectionclient.Key{}, fakeK8sClient)
@@ -236,7 +233,7 @@ func TestLoadDefaultSaramaSettings(t *testing.T) {
 // Verify that the JSON fragment can be loaded into a sarama.Config struct
 func TestMergeSaramaSettings(t *testing.T) {
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 
 	// Get a default Sarama config for verification that we don't overwrite settings when we merge
 	defaultConfig := sarama.NewConfig()
@@ -356,7 +353,7 @@ func TestSaramaConfigEqual(t *testing.T) {
 
 func TestLoadEventingKafkaSettings(t *testing.T) {
 	// Set up a configmap and verify that the sarama settings are loaded properly from it
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	configMap := commontesting.GetTestSaramaConfigMap(commontesting.OldSaramaConfig, commontesting.TestEKConfig)
 	fakeK8sClient := fake.NewSimpleClientset(configMap)
 
@@ -431,7 +428,7 @@ func verifyTestEKConfigSettings(t *testing.T, saramaConfig *sarama.Config, event
 
 func getTestSaramaContext(t *testing.T, saramaConfig string, eventingKafkaConfig string) context.Context {
 	// Set up a configmap and return a context containing that configmap (for tests)
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	configMap := commontesting.GetTestSaramaConfigMap(saramaConfig, eventingKafkaConfig)
 	fakeK8sClient := fake.NewSimpleClientset(configMap)
 	ctx := context.WithValue(context.Background(), injectionclient.Key{}, fakeK8sClient)
diff --git a/pkg/channel/distributed/common/metrics/stats_reporter_test.go b/pkg/channel/distributed/common/metrics/stats_reporter_test.go
index cce90f6c9..b5e295eb0 100644
--- a/pkg/channel/distributed/common/metrics/stats_reporter_test.go
+++ b/pkg/channel/distributed/common/metrics/stats_reporter_test.go
@@ -35,7 +35,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/metrics"
@@ -54,7 +53,7 @@ func TestMetricsServer_Report(t *testing.T) {
 	msgCount := 13579
 
 	// Initialize The Environment For The Test
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	assert.Nil(t, os.Setenv(env.MetricsDomainEnvVarKey, metricsDomain))
 
 	// Create An Observability ConfigMap For The InitializeObservability() Call To Watch
@@ -82,7 +81,7 @@ func TestMetricsServer_Report(t *testing.T) {
 	logger := logtesting.TestLogger(t).Desugar()
 
 	// Initialize The Observability Watcher
-	err := config.InitializeObservability(ctx, logger.Sugar(), metricsDomain, metricsPort)
+	err := config.InitializeObservability(ctx, logger.Sugar(), metricsDomain, metricsPort, system.Namespace())
 	assert.Nil(t, err)
 
 	// Create A New StatsReporter To Test
diff --git a/pkg/channel/distributed/common/testing/constants.go b/pkg/channel/distributed/common/testing/constants.go
index 4da092a7b..046da1543 100644
--- a/pkg/channel/distributed/common/testing/constants.go
+++ b/pkg/channel/distributed/common/testing/constants.go
@@ -17,6 +17,8 @@ limitations under the License.
 package testing
 
 const (
+	SystemNamespace = "eventing-test-ns"
+
 	OldClientId = "TestOldClientId"
 	NewClientId = "TestNewClientId"
 	OldUsername = "TestOldUsername"
diff --git a/pkg/channel/distributed/common/testing/util.go b/pkg/channel/distributed/common/testing/util.go
index 55693a1d9..e5e0f975e 100644
--- a/pkg/channel/distributed/common/testing/util.go
+++ b/pkg/channel/distributed/common/testing/util.go
@@ -18,9 +18,14 @@ package testing
 
 import (
 	"net/http"
+	"os"
 	"testing"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/system"
+
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 
 	"github.com/Shopify/sarama"
@@ -32,7 +37,7 @@ import (
 
 // Returns A ConfigMap Containing The Desired Sarama Config YAML
 func GetTestSaramaConfigMap(saramaConfig string, configuration string) *corev1.ConfigMap {
-	return GetTestSaramaConfigMapNamespaced(SettingsConfigMapName, KnativeEventingNamespace, saramaConfig, configuration)
+	return GetTestSaramaConfigMapNamespaced(SettingsConfigMapName, system.Namespace(), saramaConfig, configuration)
 }
 
 // Returns A ConfigMap Containing The Desired Sarama Config YAML, Name And Namespace
@@ -84,3 +89,13 @@ func RetryGet(url string, pause time.Duration, retryCount int, retryAgain int) (
 	// Request failed too many times; return the error for caller to process
 	return resp, err
 }
+
+// Sets the environment variables that are necessary for common components
+func SetTestEnvironment(t *testing.T) {
+	// The system.Namespace() call panics if the SYSTEM_NAMESPACE variable isn't set, so
+	// this sets an example namespace value explicitly for testing purposes
+	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, SystemNamespace))
+	// The logging.ConfigMapName() has a default if it isn't present, so this just uses that
+	// function directly to ensure that the CONFIG_LOGGING_NAME variable is set
+	assert.Nil(t, os.Setenv(env.KnativeLoggingConfigMapNameEnvVarKey, logging.ConfigMapName()))
+}
diff --git a/pkg/channel/distributed/controller/env/environment_test.go b/pkg/channel/distributed/controller/env/environment_test.go
index e3996b4af..1361b796f 100644
--- a/pkg/channel/distributed/controller/env/environment_test.go
+++ b/pkg/channel/distributed/controller/env/environment_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package env
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"os"
@@ -24,6 +25,8 @@ import (
 	"testing"
 	"time"
 
+	"knative.dev/pkg/system"
+
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
@@ -31,6 +34,7 @@ import (
 
 // Test Constants
 const (
+	systemNamespace     = "test-system-namespace"
 	serviceAccount      = "TestServiceAccount"
 	metricsPort         = "9999"
 	metricsDomain       = "example.com/kafka-eventing"
@@ -46,6 +50,7 @@ const (
 // Define The TestCase Struct
 type TestCase struct {
 	name                  string
+	systemNamespace       string
 	serviceAccount        string
 	metricsPort           string
 	metricsDomain         string
@@ -68,6 +73,11 @@ func TestGetEnvironment(t *testing.T) {
 	testCase := getValidTestCase("Valid Complete Config")
 	testCases = append(testCases, testCase)
 
+	testCase = getValidTestCase("Missing Required Config - SystemNamespace")
+	testCase.systemNamespace = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(system.NamespaceEnvKey)
+	testCases = append(testCases, testCase)
+
 	testCase = getValidTestCase("Missing Required Config - ServiceAccount")
 	testCase.serviceAccount = ""
 	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.ServiceAccountEnvVarKey)
@@ -122,6 +132,7 @@ func TestGetEnvironment(t *testing.T) {
 
 				assert.Nil(t, err)
 				assert.NotNil(t, environment)
+				assert.Equal(t, testCase.systemNamespace, environment.SystemNamespace)
 				assert.Equal(t, testCase.serviceAccount, environment.ServiceAccount)
 				assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
 				assert.Equal(t, testCase.channelImage, environment.ReceiverImage)
@@ -136,9 +147,23 @@ func TestGetEnvironment(t *testing.T) {
 	}
 }
 
+// Test that the FromContext function loads an Environment struct from a Context properly
+func TestFromContext(t *testing.T) {
+	ctx := context.TODO()
+	environment, err := FromContext(ctx)
+	assert.Nil(t, environment)
+	assert.NotNil(t, err)
+	ctx = context.WithValue(ctx, Key{}, &Environment{SystemNamespace: "test-namespace"})
+	environment, err = FromContext(ctx)
+	assert.NotNil(t, environment)
+	assert.Nil(t, err)
+	assert.Equal(t, "test-namespace", environment.SystemNamespace)
+}
+
 // Add TestCase Variables To The Environment
 func setupTestEnvironment(t *testing.T, testCase TestCase) {
 	os.Clearenv()
+	assertSetenv(t, system.NamespaceEnvKey, testCase.systemNamespace)
 	assertSetenv(t, env.ServiceAccountEnvVarKey, testCase.serviceAccount)
 	assertSetenv(t, env.MetricsDomainEnvVarKey, testCase.metricsDomain)
 	assertSetenvNonempty(t, env.MetricsPortEnvVarKey, testCase.metricsPort)
@@ -152,6 +177,7 @@ func getValidTestCase(name string) TestCase {
 	return TestCase{
 		name:                  name,
 		serviceAccount:        serviceAccount,
+		systemNamespace:       systemNamespace,
 		metricsPort:           metricsPort,
 		metricsDomain:         metricsDomain,
 		resyncPeriodMinutes:   resyncPeriodMinutes,
diff --git a/pkg/channel/distributed/controller/kafkachannel/controller_test.go b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
index 680da5ce0..02a834905 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
@@ -35,14 +35,12 @@ import (
 	controllertesting "knative.dev/eventing-kafka/pkg/channel/distributed/controller/testing"
 	fakeKafkaClient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake" // Knative Fake Informer Injection
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment/fake" // Knative Fake Informer Injection
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"    // Knative Fake Informer Injection
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
-	"knative.dev/pkg/system"
 )
 
 // Test The NewController() Functionality
@@ -52,7 +50,8 @@ func TestNewController(t *testing.T) {
 	populateEnvironmentVariables(t)
 
 	// Create A Context With Test Logger
-	ctx := logging.WithLogger(context.Background(), logtesting.TestLogger(t))
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
 
 	// Register Fake Informers (See Injection "_" Imports Above!)
 	ctx, fakeInformers := injection.Fake.SetupInformers(ctx, &rest.Config{})
@@ -68,6 +67,9 @@ func TestNewController(t *testing.T) {
 	assert.NotNil(t, fakeKafkaClientset)
 
 	// Perform The Test (Create The KafkaChannel Controller)
+	environment, err := controllerenv.GetEnvironment(logger.Desugar())
+	assert.Nil(t, err)
+	ctx = context.WithValue(ctx, controllerenv.Key{}, environment)
 	controller := NewController(ctx, nil)
 
 	// Verify The Results
@@ -202,7 +204,7 @@ func TestShutdown(t *testing.T) {
 
 // Utility Function For Populating Required Environment Variables For Testing
 func populateEnvironmentVariables(t *testing.T) {
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, controllertesting.ServiceAccount))
 	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, controllertesting.MetricsDomain))
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index 4d893b310..41ff9bd5a 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -21,6 +21,9 @@ import (
 	"sync"
 	"testing"
 
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
+
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
@@ -78,7 +81,8 @@ func TestSetKafkaAdminClient(t *testing.T) {
 	}
 
 	// Perform The Test
-	reconciler.SetKafkaAdminClient(context.TODO())
+	ctx := context.WithValue(context.TODO(), env.Key{}, &env.Environment{SystemNamespace: commontesting.SystemNamespace})
+	reconciler.SetKafkaAdminClient(ctx)
 
 	// Verify Results
 	assert.True(t, mockAdminClient1.CloseCalled())
@@ -124,6 +128,7 @@ func TestReconcile(t *testing.T) {
 	// Note - Knative reconciler framework expects Events (not errors) from ReconcileKind()
 	//        so WantErr is only for higher level failures in the injected Reconcile() function.
 	//
+	commontesting.SetTestEnvironment(t)
 	tableTest := TableTest{
 
 		//
diff --git a/pkg/channel/distributed/controller/kafkasecret/controller_test.go b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
index 0d92cbda1..a50cf0be9 100644
--- a/pkg/channel/distributed/controller/kafkasecret/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
@@ -32,14 +32,12 @@ import (
 	controllertesting "knative.dev/eventing-kafka/pkg/channel/distributed/controller/testing"
 	fakeKafkaClient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake" // Knative Fake Informer Injection
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment/fake" // Knative Fake Informer Injection
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"    // Knative Fake Informer Injection
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
-	"knative.dev/pkg/system"
 )
 
 // Test The NewController() Functionality
@@ -49,7 +47,8 @@ func TestNewController(t *testing.T) {
 	populateEnvironmentVariables(t)
 
 	// Create A Context With Test Logger
-	ctx := logging.WithLogger(context.Background(), logtesting.TestLogger(t))
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
 
 	// Register Fake Informers (See Injection "_" Imports Above!)
 	ctx, fakeInformers := injection.Fake.SetupInformers(ctx, &rest.Config{})
@@ -65,6 +64,9 @@ func TestNewController(t *testing.T) {
 	assert.NotNil(t, fakeKafkaClientset)
 
 	// Perform The Test (Create The KafkaChannel Controller)
+	environment, err := controllerenv.GetEnvironment(logger.Desugar())
+	assert.Nil(t, err)
+	ctx = context.WithValue(ctx, controllerenv.Key{}, environment)
 	controller := NewController(ctx, nil)
 
 	// Verify The Results
@@ -80,7 +82,7 @@ func TestShutdown(t *testing.T) {
 
 // Utility Function For Populating Required Environment Variables For Testing
 func populateEnvironmentVariables(t *testing.T) {
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, controllertesting.ServiceAccount))
 	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, controllertesting.MetricsDomain))
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
diff --git a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
index cd07c8430..cf9e67ae2 100644
--- a/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/reconciler_test.go
@@ -20,10 +20,12 @@ import (
 	"context"
 	"testing"
 
+	clientgotesting "k8s.io/client-go/testing"
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
-	clientgotesting "k8s.io/client-go/testing"
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/kafkasecretinjection"
@@ -56,6 +58,7 @@ func TestReconcile(t *testing.T) {
 	// Note - Knative reconciler framework expects Events (not errors) from ReconcileKind()
 	//        so WantErr is only for higher level failures in the injected Reconcile() function.
 	//
+	commontesting.SetTestEnvironment(t)
 	tableTest := TableTest{
 
 		//
@@ -204,7 +207,7 @@ func TestReconcile(t *testing.T) {
 			},
 			WantErr: true,
 			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, event.ReceiverServiceReconciliationFailed.String(), "Failed To Reconcile Receiver Service: encountered Receiver Service with DeletionTimestamp knative-eventing/kafkasecret-name-b9176d5f-receiver - potential race condition"),
+				Eventf(corev1.EventTypeWarning, event.ReceiverServiceReconciliationFailed.String(), "Failed To Reconcile Receiver Service: encountered Receiver Service with DeletionTimestamp "+controllertesting.KafkaSecretNamespace+"/kafkasecret-name-b9176d5f-receiver - potential race condition"),
 				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
 			},
 		},
@@ -266,7 +269,7 @@ func TestReconcile(t *testing.T) {
 			},
 			WantErr: true,
 			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, event.ReceiverDeploymentReconciliationFailed.String(), "Failed To Reconcile Receiver Deployment: encountered Receiver Deployment with DeletionTimestamp knative-eventing/kafkasecret-name-b9176d5f-receiver - potential race condition"),
+				Eventf(corev1.EventTypeWarning, event.ReceiverDeploymentReconciliationFailed.String(), "Failed To Reconcile Receiver Deployment: encountered Receiver Deployment with DeletionTimestamp "+controllertesting.KafkaSecretNamespace+"/kafkasecret-name-b9176d5f-receiver - potential race condition"),
 				controllertesting.NewKafkaSecretFailedReconciliationEvent(),
 			},
 		},
diff --git a/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go b/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go
index 394a236c6..3d3269d8e 100644
--- a/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go
+++ b/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go
@@ -23,6 +23,8 @@ import (
 	"testing"
 	"time"
 
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+
 	"github.com/stretchr/testify/assert"
 	"k8s.io/client-go/kubernetes/fake"
 	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
@@ -61,6 +63,7 @@ func TestGet(t *testing.T) {
 // Utility Function For Populating Required Environment Variables For Testing
 func populateEnvironmentVariables(t *testing.T) {
 	// Most of these are not actually used, but they need to exist or the GetEnvironment call will fail
+	commontesting.SetTestEnvironment(t)
 	assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, controllertesting.ServiceAccount))
 	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, controllertesting.MetricsDomain))
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index 6ec6f4dd1..998e4dd47 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -21,6 +21,8 @@ import (
 	"strconv"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -38,7 +40,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/eventing/pkg/apis/messaging"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
@@ -70,7 +71,7 @@ const (
 	KafkaChannelNamespace  = "kafkachannel-namespace"
 	KafkaChannelName       = "kafkachannel-name"
 	KafkaChannelKey        = KafkaChannelNamespace + "/" + KafkaChannelName
-	KafkaSecretNamespace   = commonconstants.KnativeEventingNamespace // Needs To Match Hardcoded Value In Reconciliation
+	KafkaSecretNamespace   = "eventing-test-ns" // Needs To Match system.Namespace() Call In Reconciliation
 	KafkaSecretName        = "kafkasecret-name"
 	KafkaSecretKey         = KafkaSecretNamespace + "/" + KafkaSecretName
 	ReceiverDeploymentName = KafkaSecretName + "-b9176d5f-receiver" // Truncated MD5 Hash Of KafkaSecretName
@@ -191,6 +192,7 @@ func WithoutFinalizersDeployment(deployment *appsv1.Deployment) {
 // Set The Required Environment Variables
 func NewEnvironment() *env.Environment {
 	return &env.Environment{
+		SystemNamespace: testing.SystemNamespace,
 		ServiceAccount:  ServiceAccount,
 		MetricsPort:     MetricsPort,
 		MetricsDomain:   MetricsDomain,
@@ -476,7 +478,7 @@ func NewKafkaChannelService(options ...ServiceOption) *corev1.Service {
 		},
 		Spec: corev1.ServiceSpec{
 			Type:         corev1.ServiceTypeExternalName,
-			ExternalName: ReceiverDeploymentName + "." + commonconstants.KnativeEventingNamespace + ".svc.cluster.local",
+			ExternalName: ReceiverDeploymentName + "." + system.Namespace() + ".svc.cluster.local",
 		},
 	}
 
@@ -500,7 +502,7 @@ func NewKafkaChannelReceiverService(options ...ServiceOption) *corev1.Service {
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      ReceiverDeploymentName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: system.Namespace(),
 			Labels: map[string]string{
 				"k8s-app":               "eventing-kafka-channels",
 				"kafkachannel-receiver": "true",
@@ -540,6 +542,7 @@ func NewKafkaChannelReceiverService(options ...ServiceOption) *corev1.Service {
 // Utility Function For Creating A Receiver Deployment For The Test Channel
 func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Deployment {
 
+	systemNamespace := system.Namespace()
 	// Replicas Int Reference
 	replicas := int32(ReceiverReplicas)
 
@@ -551,7 +554,7 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      ReceiverDeploymentName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: systemNamespace,
 			Labels: map[string]string{
 				"app":                   ReceiverDeploymentName,
 				"kafkachannel-receiver": "true",
@@ -608,7 +611,7 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 							Env: []corev1.EnvVar{
 								{
 									Name:  system.NamespaceEnvKey,
-									Value: commonconstants.KnativeEventingNamespace,
+									Value: systemNamespace,
 								},
 								{
 									Name: commonenv.PodNameEnvVarKey,
@@ -717,7 +720,7 @@ func NewKafkaChannelDispatcherService(options ...ServiceOption) *corev1.Service
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      serviceName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: system.Namespace(),
 			Labels: map[string]string{
 				constants.KafkaChannelNameLabel:       KafkaChannelName,
 				constants.KafkaChannelNamespaceLabel:  KafkaChannelNamespace,
@@ -756,6 +759,7 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 	sparseKafkaChannel := &kafkav1beta1.KafkaChannel{ObjectMeta: metav1.ObjectMeta{Namespace: KafkaChannelNamespace, Name: KafkaChannelName}}
 	dispatcherName := util.DispatcherDnsSafeName(sparseKafkaChannel)
 	topicName := util.TopicName(sparseKafkaChannel)
+	systemNamespace := system.Namespace()
 
 	// Replicas Int Reference
 	replicas := int32(DispatcherReplicas)
@@ -767,7 +771,7 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      dispatcherName,
-			Namespace: commonconstants.KnativeEventingNamespace,
+			Namespace: systemNamespace,
 			Labels: map[string]string{
 				constants.AppLabel:                    dispatcherName,
 				constants.KafkaChannelNameLabel:       KafkaChannelName,
@@ -818,7 +822,7 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 							Env: []corev1.EnvVar{
 								{
 									Name:  system.NamespaceEnvKey,
-									Value: commonconstants.KnativeEventingNamespace,
+									Value: systemNamespace,
 								},
 								{
 									Name: commonenv.PodNameEnvVarKey,
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index 3fb850202..2db7ff071 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -19,7 +19,6 @@ package dispatcher
 import (
 	"encoding/json"
 	"fmt"
-	"os"
 
 	"github.com/Shopify/sarama"
 	"github.com/ghodss/yaml"
@@ -30,7 +29,7 @@ import (
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	kafkaconsumer "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer"
 	kafkatesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/testing"
-	"knative.dev/eventing-kafka/pkg/common/constants"
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 	logtesting "knative.dev/pkg/logging/testing"
@@ -366,7 +365,7 @@ func TestConfigChanged(t *testing.T) {
 	logger := logtesting.TestLogger(t).Desugar()
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, constants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 
 	// Create Mocks
 	var dispatcher Dispatcher
diff --git a/pkg/channel/distributed/dispatcher/env/environment_test.go b/pkg/channel/distributed/dispatcher/env/environment_test.go
index fc86c05b4..bfdcf9d42 100644
--- a/pkg/channel/distributed/dispatcher/env/environment_test.go
+++ b/pkg/channel/distributed/dispatcher/env/environment_test.go
@@ -24,6 +24,8 @@ import (
 	"testing"
 	"time"
 
+	"knative.dev/pkg/system"
+
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
 	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
@@ -31,23 +33,25 @@ import (
 
 // Test Constants
 const (
-	metricsPort   = "9999"
-	metricsDomain = "kafka-eventing"
-	healthPort    = "1234"
-	resyncPeriod  = "3600"
-	kafkaBrokers  = "TestKafkaBrokers"
-	kafkaTopic    = "TestKafkaTopic"
-	channelKey    = "TestChannelKey"
-	serviceName   = "TestServiceName"
-	kafkaUsername = "TestKafkaUsername"
-	kafkaPassword = "TestKafkaPassword"
-	podName       = "TestPod"
-	containerName = "TestContainer"
+	systemNamespace = "test-system-namespace"
+	metricsPort     = "9999"
+	metricsDomain   = "kafka-eventing"
+	healthPort      = "1234"
+	resyncPeriod    = "3600"
+	kafkaBrokers    = "TestKafkaBrokers"
+	kafkaTopic      = "TestKafkaTopic"
+	channelKey      = "TestChannelKey"
+	serviceName     = "TestServiceName"
+	kafkaUsername   = "TestKafkaUsername"
+	kafkaPassword   = "TestKafkaPassword"
+	podName         = "TestPod"
+	containerName   = "TestContainer"
 )
 
 // Define The TestCase Struct
 type TestCase struct {
 	name                 string
+	systemNamespace      string
 	metricsPort          string
 	metricsDomain        string
 	healthPort           string
@@ -75,6 +79,11 @@ func TestGetEnvironment(t *testing.T) {
 	testCase := getValidTestCase("Valid Complete Config")
 	testCases = append(testCases, testCase)
 
+	testCase = getValidTestCase("Missing Required Config - SystemNamespace")
+	testCase.systemNamespace = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(system.NamespaceEnvKey)
+	testCases = append(testCases, testCase)
+
 	testCase = getValidTestCase("Missing Required Config - MetricsDomain")
 	testCase.metricsDomain = ""
 	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.MetricsDomainEnvVarKey)
@@ -146,6 +155,7 @@ func TestGetEnvironment(t *testing.T) {
 
 			// (Re)Setup The Environment Variables From TestCase
 			os.Clearenv()
+			assertSetenv(t, system.NamespaceEnvKey, testCase.systemNamespace)
 			assertSetenv(t, commonenv.MetricsDomainEnvVarKey, testCase.metricsDomain)
 			assertSetenvNonempty(t, commonenv.MetricsPortEnvVarKey, testCase.metricsPort)
 			assertSetenvNonempty(t, commonenv.HealthPortEnvVarKey, testCase.healthPort)
@@ -167,6 +177,7 @@ func TestGetEnvironment(t *testing.T) {
 
 				assert.Nil(t, err)
 				assert.NotNil(t, environment)
+				assert.Equal(t, testCase.systemNamespace, environment.SystemNamespace)
 				assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
 				assert.Equal(t, testCase.healthPort, strconv.Itoa(environment.HealthPort))
 				assert.Equal(t, testCase.kafkaBrokers, environment.KafkaBrokers)
@@ -201,6 +212,7 @@ func assertSetenvNonempty(t *testing.T, envKey string, value string) {
 func getValidTestCase(name string) TestCase {
 	return TestCase{
 		name:                 name,
+		systemNamespace:      systemNamespace,
 		metricsPort:          metricsPort,
 		metricsDomain:        metricsDomain,
 		healthPort:           healthPort,
diff --git a/pkg/channel/distributed/receiver/env/environment_test.go b/pkg/channel/distributed/receiver/env/environment_test.go
index 3455f3783..e09bc67f9 100644
--- a/pkg/channel/distributed/receiver/env/environment_test.go
+++ b/pkg/channel/distributed/receiver/env/environment_test.go
@@ -24,6 +24,8 @@ import (
 	"testing"
 	"time"
 
+	"knative.dev/pkg/system"
+
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
@@ -31,6 +33,7 @@ import (
 
 // Test Constants
 const (
+	systemNamespace     = "test-system-namespace"
 	metricsPort         = "9999"
 	metricsDomain       = "kafka-eventing"
 	healthPort          = "1234"
@@ -46,6 +49,7 @@ const (
 // Define The TestCase Struct
 type TestCase struct {
 	name                 string
+	systemNamespace      string
 	metricsPort          string
 	metricsDomain        string
 	healthPort           string
@@ -71,6 +75,11 @@ func TestGetEnvironment(t *testing.T) {
 	testCase := getValidTestCase("Valid Complete Config")
 	testCases = append(testCases, testCase)
 
+	testCase = getValidTestCase("Missing Required Config - SystemNamespace")
+	testCase.systemNamespace = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(system.NamespaceEnvKey)
+	testCases = append(testCases, testCase)
+
 	testCase = getValidTestCase("Missing Required Config - MetricsDomain")
 	testCase.metricsDomain = ""
 	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.MetricsDomainEnvVarKey)
@@ -132,6 +141,7 @@ func TestGetEnvironment(t *testing.T) {
 
 			// (Re)Setup The Environment Variables From TestCase
 			os.Clearenv()
+			assertSetenv(t, system.NamespaceEnvKey, testCase.systemNamespace)
 			assertSetenv(t, env.MetricsDomainEnvVarKey, testCase.metricsDomain)
 			assertSetenvNonempty(t, env.MetricsPortEnvVarKey, testCase.metricsPort)
 			assertSetenvNonempty(t, env.HealthPortEnvVarKey, testCase.healthPort)
@@ -151,6 +161,7 @@ func TestGetEnvironment(t *testing.T) {
 
 				assert.Nil(t, err)
 				assert.NotNil(t, environment)
+				assert.Equal(t, testCase.systemNamespace, environment.SystemNamespace)
 				assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
 				assert.Equal(t, testCase.healthPort, strconv.Itoa(environment.HealthPort))
 				assert.Equal(t, testCase.kafkaBrokers, environment.KafkaBrokers)
@@ -183,6 +194,7 @@ func assertSetenvNonempty(t *testing.T, envKey string, value string) {
 func getValidTestCase(name string) TestCase {
 	return TestCase{
 		name:                 name,
+		systemNamespace:      systemNamespace,
 		metricsPort:          metricsPort,
 		metricsDomain:        metricsDomain,
 		healthPort:           healthPort,
diff --git a/pkg/channel/distributed/receiver/producer/producer_test.go b/pkg/channel/distributed/receiver/producer/producer_test.go
index 96474638a..3e6b6246e 100644
--- a/pkg/channel/distributed/receiver/producer/producer_test.go
+++ b/pkg/channel/distributed/receiver/producer/producer_test.go
@@ -19,7 +19,6 @@ package producer
 import (
 	"context"
 	"encoding/json"
-	"os"
 
 	"github.com/Shopify/sarama"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -30,10 +29,10 @@ import (
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	commonconfig "knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/constants"
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	receivertesting "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/testing"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/system"
 
@@ -188,7 +187,7 @@ func TestConfigChanged(t *testing.T) {
 	defer func() { createSyncProducerWrapper = createSyncProducerWrapperPlaceholder }()
 
 	// Setup Environment
-	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, commonconstants.KnativeEventingNamespace))
+	commontesting.SetTestEnvironment(t)
 	// Create Mocks
 	mockSyncProducer := receivertesting.NewMockSyncProducer()
 	producer := createTestProducer(t, mockSyncProducer)
diff --git a/pkg/source/reconciler/source/resources/labels_test.go b/pkg/source/reconciler/source/resources/labels_test.go
index 0f7b4bc21..d2d302878 100644
--- a/pkg/source/reconciler/source/resources/labels_test.go
+++ b/pkg/source/reconciler/source/resources/labels_test.go
@@ -28,7 +28,7 @@ func TestGetLabels(t *testing.T) {
 
 	wantLabels := map[string]string{
 		"eventing.knative.dev/source":     "kafka-source-controller",
-		"eventing.knative.dev/SourceName": "testSourceName",
+		"eventing.knative.dev/sourceName": "testSourceName",
 	}
 
 	eq := cmp.Equal(testLabels, wantLabels)
diff --git a/test/config/config-tracing.yaml b/test/config/config-tracing.yaml
index 35b08ea09..b1c6c6397 100644
--- a/test/config/config-tracing.yaml
+++ b/test/config/config-tracing.yaml
@@ -16,9 +16,9 @@ apiVersion: v1
 kind: ConfigMap
 metadata:
   name: config-tracing
-  namespace: knative-eventing
+  namespace: ${SYSTEM_NAMESPACE}
 data:
   backend: "zipkin"
-  zipkin-endpoint: "http://zipkin.knative-eventing.svc.cluster.local:9411/api/v2/spans"
+  zipkin-endpoint: "http://zipkin.${SYSTEM_NAMESPACE}.svc.cluster.local:9411/api/v2/spans"
   debug: "true"
   sample-rate: "1.0"
diff --git a/test/config/monitoring.yaml b/test/config/monitoring.yaml
index f219821aa..0f226f52d 100644
--- a/test/config/monitoring.yaml
+++ b/test/config/monitoring.yaml
@@ -16,7 +16,7 @@ apiVersion: v1
 kind: Service
 metadata:
   name: zipkin
-  namespace: knative-eventing
+  namespace: ${SYSTEM_NAMESPACE}
 spec:
   ports:
   - name: http
@@ -28,7 +28,7 @@ apiVersion: apps/v1
 kind: Deployment
 metadata:
   name: zipkin
-  namespace: knative-eventing
+  namespace: ${SYSTEM_NAMESPACE}
 spec:
   replicas: 1
   selector:
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index cd6af83e3..c7a4491df 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -94,13 +94,25 @@ readonly KAFKA_SASL_CLUSTER_URL="my-cluster-kafka-bootstrap.kafka.svc.cluster.lo
 KAFKA_CLUSTER_URL=${KAFKA_PLAIN_CLUSTER_URL}
 # Kafka channel CRD config template file. It needs to be modified to be the real config file.
 readonly KAFKA_CRD_CONFIG_TEMPLATE="400-kafka-config.yaml"
-# Real Kafka channel CRD config , generated from the template directory and modified template file.
+
+# Real Kafka channel CRD config, generated from the template directory and modified template file.
 readonly KAFKA_CRD_CONFIG_DIR="$(mktemp -d)"
+# Real Kafka Source CRD config, generated from the template directory and modified template file.
+readonly KAFKA_SOURCE_CRD_CONFIG_DIR="$(mktemp -d)"
+
+# Remove the temporary directories on exit (avoiding "rm -rf" to prevent disaster if something is wrong with the variables)
+trap "{ for dirrm in \"${KAFKA_CRD_CONFIG_DIR}\" \"${KAFKA_SOURCE_CRD_CONFIG_DIR}\"; do rm \"\${dirrm}\"/*; rmdir \"\${dirrm}\"; done }" EXIT
+
 # Kafka channel CRD config template directory.
-readonly KAFKA_SOURCE_CRD_CONFIG_DIR="config/source"
+readonly KAFKA_SOURCE_TEMPLATE_DIR="config/source"
+
+# Namespaces where we install Eventing components
+# This is the namespace of knative-eventing itself
+export EVENTING_NAMESPACE="knative-eventing"
 
-# Namespace where we install Eventing components
+# Namespace where we install eventing-kafka components (may be different than EVENTING_NAMESPACE)
 readonly SYSTEM_NAMESPACE="knative-eventing"
+export SYSTEM_NAMESPACE
 
 # Zipkin setup
 readonly KNATIVE_EVENTING_MONITORING_YAML="test/config/monitoring.yaml"
@@ -164,7 +176,7 @@ function knative_setup() {
     ko apply -f "${EVENTING_IN_MEMORY_CHANNEL_CONFIG}"
     popd
   fi
-  wait_until_pods_running knative-eventing || fail_test "Knative Eventing did not come up"
+  wait_until_pods_running "${EVENTING_NAMESPACE}" || fail_test "Knative Eventing did not come up"
 
   install_zipkin
 }
@@ -172,18 +184,18 @@ function knative_setup() {
 # Setup zipkin
 function install_zipkin() {
   echo "Installing Zipkin..."
-  kubectl apply -f "${KNATIVE_EVENTING_MONITORING_YAML}"
-  wait_until_pods_running knative-eventing || fail_test "Zipkin inside eventing did not come up"
+  sed "s/\${SYSTEM_NAMESPACE}/${SYSTEM_NAMESPACE}/g" < "${KNATIVE_EVENTING_MONITORING_YAML}" | kubectl apply -f -
+  wait_until_pods_running "${SYSTEM_NAMESPACE}" || fail_test "Zipkin inside eventing did not come up"
   # Setup config tracing for tracing tests
-  kubectl apply -f "${CONFIG_TRACING_CONFIG}"
+  sed "s/\${SYSTEM_NAMESPACE}/${SYSTEM_NAMESPACE}/g" <  "${CONFIG_TRACING_CONFIG}" | kubectl apply -f -
 }
 
 # Remove zipkin
 function uninstall_zipkin() {
   echo "Uninstalling Zipkin..."
-  kubectl delete -f "${KNATIVE_EVENTING_MONITORING_YAML}"
-  wait_until_object_does_not_exist deployment zipkin knative-eventing || fail_test "Zipkin deployment was unable to be deleted"
-  kubectl delete -n knative-eventing configmap config-tracing
+  sed "s/\${SYSTEM_NAMESPACE}/${SYSTEM_NAMESPACE}/g" <  "${KNATIVE_EVENTING_MONITORING_YAML}" | kubectl delete -f -
+  wait_until_object_does_not_exist deployment zipkin "${SYSTEM_NAMESPACE}" || fail_test "Zipkin deployment was unable to be deleted"
+  kubectl delete -n "${SYSTEM_NAMESPACE}" configmap config-tracing
 }
 
 function knative_teardown() {
@@ -203,7 +215,7 @@ function knative_teardown() {
     ko delete -f "${EVENTING_CONFIG}"
     popd
   fi
-  wait_until_object_does_not_exist namespaces knative-eventing
+  wait_until_object_does_not_exist namespaces "${EVENTING_NAMESPACE}"
 }
 
 # Add function call to trap
@@ -252,21 +264,25 @@ function install_consolidated_channel_crds() {
   echo "Installing consolidated Kafka Channel CRD"
   rm "${KAFKA_CRD_CONFIG_DIR}/"*yaml
   cp "${CONSOLIDATED_TEMPLATE_DIR}/"*yaml "${KAFKA_CRD_CONFIG_DIR}"
+  sed -i "s/namespace: knative-eventing/namespace: ${SYSTEM_NAMESPACE}/g" "${KAFKA_CRD_CONFIG_DIR}/"*yaml
   sed -i "s/REPLACE_WITH_CLUSTER_URL/${KAFKA_CLUSTER_URL}/" ${KAFKA_CRD_CONFIG_DIR}/${KAFKA_CRD_CONFIG_TEMPLATE}
   ko apply -f "${KAFKA_CRD_CONFIG_DIR}" || return 1
-  wait_until_pods_running knative-eventing || fail_test "Failed to install the consolidated Kafka Channel CRD"
+  wait_until_pods_running "${SYSTEM_NAMESPACE}" || fail_test "Failed to install the consolidated Kafka Channel CRD"
 }
 
 function install_consolidated_sources_crds() {
   echo "Installing consolidated Kafka Source CRD"
+  rm "${KAFKA_SOURCE_CRD_CONFIG_DIR}/"*yaml
+  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
-  wait_until_pods_running knative-eventing || fail_test "Failed to install the consolidated Kafka Source CRD"
+  wait_until_pods_running "${EVENTING_NAMESPACE}" || fail_test "Failed to install the consolidated Kafka Source CRD"
 }
 
 # Uninstall The eventing-kafka KafkaChannel Implementation Via Ko
 function uninstall_channel_crds() {
   echo "Uninstalling Kafka Channel CRD"
-  kubectl delete secret -n knative-eventing kafka-cluster
+  kubectl delete secret -n "${SYSTEM_NAMESPACE}" kafka-cluster
   sleep 10 # Give Controller Time To React To Kafka Secret Deletion ; )
   echo "Current namespaces:"
   kubectl get namespaces
@@ -284,6 +300,7 @@ function install_distributed_channel_crds() {
   echo "Installing distributed Kafka Channel CRD"
   rm "${KAFKA_CRD_CONFIG_DIR}/"*yaml
   cp "${DISTRIBUTED_TEMPLATE_DIR}/"*yaml "${KAFKA_CRD_CONFIG_DIR}"
+  sed -i "s/namespace: knative-eventing/namespace: ${SYSTEM_NAMESPACE}/g" "${KAFKA_CRD_CONFIG_DIR}/"*yaml
 
   # Update The Kafka Secret With Strimzi Kafka Cluster Brokers (No Authentication)
   sed -i "s/brokers: \"\"/brokers: ${STRIMZI_KAFKA_CLUSTER_BROKERS_ENCODED}/" "${KAFKA_CRD_CONFIG_DIR}/${EVENTING_KAFKA_SECRET_TEMPLATE}"
@@ -295,9 +312,9 @@ function install_distributed_channel_crds() {
   ko apply -f "${KAFKA_CRD_CONFIG_DIR}" || return 1
 
    # Add The kn-eventing-test-pull-secret (If Present) To ServiceAccount & Restart eventing-kafka Deployment
-  add_kn_eventing_test_pull_secret knative-eventing eventing-kafka-channel-controller eventing-kafka-channel-controller
+  add_kn_eventing_test_pull_secret "${SYSTEM_NAMESPACE}" eventing-kafka-channel-controller eventing-kafka-channel-controller
 
-  wait_until_pods_running knative-eventing || fail_test "Failed to install the distributed Kafka Channel CRD"
+  wait_until_pods_running "${SYSTEM_NAMESPACE}" || fail_test "Failed to install the distributed Kafka Channel CRD"
 }
 
 function kafka_setup() {
@@ -344,7 +361,7 @@ function create_tls_secrets() {
   TLSUSER_CRT=$(kubectl -n ${STRIMZI_KAFKA_NAMESPACE} get secret my-tls-user --template='{{index .data "user.crt"}}' | base64 --decode )
   TLSUSER_KEY=$(kubectl -n ${STRIMZI_KAFKA_NAMESPACE} get secret my-tls-user --template='{{index .data "user.key"}}' | base64 --decode )
 
-  kubectl create secret --namespace knative-eventing generic strimzi-tls-secret \
+  kubectl create secret --namespace "${EVENTING_NAMESPACE}" generic strimzi-tls-secret \
     --from-literal=ca.crt="$STRIMZI_CRT" \
     --from-literal=user.crt="$TLSUSER_CRT" \
     --from-literal=user.key="$TLSUSER_KEY"
@@ -355,7 +372,7 @@ function create_sasl_secrets() {
   STRIMZI_CRT=$(kubectl -n ${STRIMZI_KAFKA_NAMESPACE} get secret my-cluster-cluster-ca-cert --template='{{index .data "ca.crt"}}' | base64 --decode )
   SASL_PASSWD=$(kubectl -n ${STRIMZI_KAFKA_NAMESPACE} get secret my-sasl-user --template='{{index .data "password"}}' | base64 --decode )
 
-  kubectl create secret --namespace knative-eventing generic strimzi-sasl-secret \
+  kubectl create secret --namespace "${EVENTING_NAMESPACE}" generic strimzi-sasl-secret \
     --from-literal=ca.crt="$STRIMZI_CRT" \
     --from-literal=password="$SASL_PASSWD" \
     --from-literal=saslType="SCRAM-SHA-512" \
@@ -380,7 +397,7 @@ function test_consolidated_channel_tls() {
   # Test the consolidated channel with TLS
   echo "Testing the consolidated channel with TLS"
   # Set the URL to the TLS listeners config
-  cp ${KAFKA_TLS_CONFIG} "${CONSOLIDATED_TEMPLATE_DIR}/configmaps/kafka-config.yaml"
+  cp "${KAFKA_TLS_CONFIG}" "${CONSOLIDATED_TEMPLATE_DIR}/configmaps/kafka-config.yaml"
   KAFKA_CLUSTER_URL=${KAFKA_TLS_CLUSTER_URL}
 
   install_consolidated_channel_crds || return 1
@@ -394,7 +411,7 @@ function test_consolidated_channel_sasl() {
   # Test the consolidated channel with SASL
   echo "Testing the consolidated channel with SASL"
   # Set the URL to the SASL listeners config
-  cp ${KAFKA_SASL_CONFIG} "${CONSOLIDATED_TEMPLATE_DIR}/configmaps/kafka-config.yaml"
+  cp "${KAFKA_SASL_CONFIG}" "${CONSOLIDATED_TEMPLATE_DIR}/configmaps/kafka-config.yaml"
   KAFKA_CLUSTER_URL=${KAFKA_SASL_CLUSTER_URL}
 
   install_consolidated_channel_crds || return 1
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 8d09be4f7..87bbab93f 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -44,6 +44,9 @@ TEST_PARALLEL=${MAX_PARALLEL_TESTS:-12}
 
 source "$(dirname "$0")/e2e-common.sh"
 
+# Create the system namespace if it doesn't already exist (may be the same as the EVENTING_NAMESPACE)
+kubectl get namespace "${SYSTEM_NAMESPACE}" || kubectl create namespace "${SYSTEM_NAMESPACE}"
+
 TEST_CONSOLIDATED_CHANNEL=${TEST_CONSOLIDATED_CHANNEL:-0}
 TEST_CONSOLIDATED_CHANNEL_TLS=${TEST_CONSOLIDATED_CHANNEL_TLS:-0}
 TEST_CONSOLIDATED_CHANNEL_SASL=${TEST_CONSOLIDATED_CHANNEL_SASL:-0}
@@ -51,25 +54,33 @@ TEST_DISTRIBUTED_CHANNEL=${TEST_DISTRIBUTED_CHANNEL:-0}
 
 echo "e2e-tests.sh command line: $@"
 
-# Note:  The setting of gcp-project-id option here has no effect when testing locally; it is only for the kubetest2 utility
 # If you wish to use this script just as test setup, *without* teardown, add "--skip-teardowns" to the initialize command
 initialize $@ --skip-istio-addon
 
+# Copy some resources if the SYSTEM_NAMESPACE is not the same as the EVENTING_NAMESPACE
+if [[ $SYSTEM_NAMESPACE != $EVENTING_NAMESPACE ]]; then
+  for configmap in config-leader-election config-logging config-observability config-kafka; do
+    kubectl get configmap "${configmap}" "--namespace=${EVENTING_NAMESPACE}" -o yaml | sed "s/namespace: ${EVENTING_NAMESPACE}/namespace: ${SYSTEM_NAMESPACE}/" | kubectl create -f -
+  done
+fi
+
 echo "e2e-tests.sh environment:"
+echo "EVENTING_NAMESPACE: ${EVENTING_NAMESPACE}"
+echo "SYSTEM_NAMESPACE: ${SYSTEM_NAMESPACE}"
 echo "TEST_CONSOLIDATED_CHANNEL: ${TEST_CONSOLIDATED_CHANNEL}"
 echo "TEST_DISTRIBUTED_CHANNEL: ${TEST_DISTRIBUTED_CHANNEL}"
 echo "TEST_CONSOLIDATED_CHANNEL_TLS: ${TEST_CONSOLIDATED_CHANNEL_TLS}"
 echo "TEST_CONSOLIDATED_CHANNEL_SASL: ${TEST_CONSOLIDATED_CHANNEL_SASL}"
 
-# If neither of the four test was specified, run both plain tests
+# If none of the tests were explicitly specified, run both plain tests
 if [[ $TEST_CONSOLIDATED_CHANNEL != 1 ]] && [[ $TEST_CONSOLIDATED_CHANNEL_TLS != 1 ]] && [[ $TEST_CONSOLIDATED_CHANNEL_SASL != 1 ]] && [[ $TEST_DISTRIBUTED_CHANNEL != 1 ]]; then
   TEST_DISTRIBUTED_CHANNEL=1
   TEST_CONSOLIDATED_CHANNEL=1
 fi
 
-export SYSTEM_NAMESPACE
 create_tls_secrets
 create_sasl_secrets
+
 if [[ $TEST_CONSOLIDATED_CHANNEL == 1 ]]; then
   echo "Launching the PLAIN TESTS:"
   test_consolidated_channel_plain || exit 1
diff --git a/test/e2e/helpers/kafka_helper.go b/test/e2e/helpers/kafka_helper.go
index 2d36acf1b..012eaa01b 100644
--- a/test/e2e/helpers/kafka_helper.go
+++ b/test/e2e/helpers/kafka_helper.go
@@ -269,7 +269,7 @@ func CheckKafkaSourceState(ctx context.Context, c *testlib.Client, name string,
 //On timeout reports error
 func CheckRADeployment(ctx context.Context, c *testlib.Client, name string, inState func(deps *appsv1.DeploymentList) (bool, error)) error {
 	listOptions := metav1.ListOptions{
-		LabelSelector: fmt.Sprintf("%s=%s", "eventing.knative.dev/SourceName", name),
+		LabelSelector: fmt.Sprintf("%s=%s", "eventing.knative.dev/sourceName", name),
 	}
 	kDeps := c.Kube.AppsV1().Deployments(c.Namespace)
 	var lastState *appsv1.DeploymentList
diff --git a/vendor/knative.dev/pkg/test/mako/sidecar.go b/vendor/knative.dev/pkg/test/mako/sidecar.go
index ad1531f34..ff890f624 100644
--- a/vendor/knative.dev/pkg/test/mako/sidecar.go
+++ b/vendor/knative.dev/pkg/test/mako/sidecar.go
@@ -93,7 +93,7 @@ func SetupHelper(ctx context.Context, benchmarkKey *string, benchmarkName *strin
 	// Get the commit of the benchmarks
 	commitID, err := changeset.Get()
 	if err != nil {
-		return nil, err
+		log.Println("Cannot find commit ID")
 	}
 
 	// Setup a deployment informer, so that we can use the lister to track