Repository: knative-sandbox/eventing-kafka
Pull request #192 (2020-11-13 17:00:19)

Adding global TLS/SASL for the consolidated KafkaChannel
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

This is work in progress, and needs a bit more of polishing... Details below:

## Proposed Changes

- Adding a few fields to the  `config-kafka` ConfigMap, to resolve a `Secret` in a given namespace. It's global due to the nature of the channel (e.g: broker URLs are also global)
- support for TLS/SASL

## Usage

Assuming some TLS Kafka (e.g. [strimzi](https://github.com/strimzi/strimzi-kafka-operator/tree/release-0.20.x/examples/security/tls-auth)), the usage goes like:

Create `tls secret` in your namesapce. Below we leverage the settings from Strimzi:

```shell
STRIMZI_CRT=$(k -n kafka get secret my-cluster-cluster-ca-cert --template='{{index .data "ca.crt"}}' | base64 --decode )
TLSUSER_CRT=$(k -n kafka get secret my-user --template='{{index .data "user.crt"}}' | base64 --decode )
TLSUSER_KEY=$(k -n kafka get secret my-user --template='{{index .data "user.key"}}' | base64 --decode )


kubectl create secret --namespace default generic my-tls-secret \
    --from-literal=ca.crt="$STRIMZI_CRT" \
    --from-literal=user.crt="$TLSUSER_CRT" \
    --from-literal=user.key="$TLSUSER_KEY" \
```

Add to the `config-kafka` ConfigMap, like:

```yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: config-kafka
  namespace: knative-eventing
data:
  bootstrapServers: my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093
  authSecretNamespace: default
  authSecretName: my-tls-secret
```

## Follow up

- docs (will do follow up PR, once merged)

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/192#" title="Author self-approved">matzew</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)~~ [matzew]

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327295250636476416/artifacts/line-cov.html#file0) | 63.3% | 58.6% | -4.8
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327295250636476416/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327295250636476416/artifacts/line-cov.html#file2) | Do not exist | 48.5% | 

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

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

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

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 2af10900c9c94040428c3360d676a91dcc6b31fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-unit-tests/1327295250619699200) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 2af10900c9c94040428c3360d676a91dcc6b31fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327295250636476416) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<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 -->
lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 2af10900c9c94040428c3360d676a91dcc6b31fc into 628e07a66ef302df51d7b201369a208902482509 - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-13ad0e3476af471c09701a4efa34aa262c19b373)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192?src=pr&el=h1) Report
> Merging [#192](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192?src=pr&el=desc) (6801dbf) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/25f218b21aaf3fea656e484bd4655983352e6c9d?el=desc) (25f218b) will **decrease** coverage by `0.81%`.
> The diff coverage is `12.06%`.

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

```diff
@@            Coverage Diff             @@
##           master     #192      +/-   ##
==========================================
- Coverage   76.05%   75.24%   -0.82%     
==========================================
  Files         112      111       -1     
  Lines        4206     4257      +51     
==========================================
+ Hits         3199     3203       +4     
- Misses        817      862      +45     
- Partials      190      192       +2     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `45.97% <0.00%> (-3.72%)` | :arrow_down: |
| [...consolidated/reconciler/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9rYWZrYWNoYW5uZWwuZ28=) | `46.40% <0.00%> (-0.57%)` | :arrow_down: |
| [pkg/common/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9jbGllbnQuZ28=) | `44.28% <0.00%> (ø)` | |
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `54.16% <25.00%> (-29.17%)` | :arrow_down: |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `69.23% <100.00%> (ø)` | |
| [...tributed/common/kafka/admin/eventhubcache/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/192/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL2FkbWluL2V2ZW50aHViY2FjaGUvY2FjaGUuZ28=) | `65.06% <0.00%> (-2.41%)` | :arrow_down: |

------

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327518059878944768/artifacts/line-cov.html#file0) | 63.3% | 58.6% | -4.8
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327518059878944768/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327518059878944768/artifacts/line-cov.html#file2) | 90.5% | 60.5% | -29.9
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327518059878944768/artifacts/line-cov.html#file3) | Do not exist | 48.5% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 6801dbf605739b8ab90904e7ad7b6ab0c2bd93b1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327518059878944768) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<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 -->
lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 6801dbf605739b8ab90904e7ad7b6ab0c2bd93b1 into 25f218b21aaf3fea656e484bd4655983352e6c9d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-c542d6329e279300537e5403e2747afa51f8d604)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327545798086365184/artifacts/line-cov.html#file0) | 63.3% | 58.6% | -4.8
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327545798086365184/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327545798086365184/artifacts/line-cov.html#file2) | 90.5% | 60.5% | -29.9
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327545798086365184/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 6801dbf605739b8ab90904e7ad7b6ab0c2bd93b1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1327518059878944768) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-unit-tests | 811be0f00c0ad666b422fcc540f06962900ac492 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-unit-tests/1327545798077976576) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<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 -->
lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 811be0f00c0ad666b422fcc540f06962900ac492 into 25f218b21aaf3fea656e484bd4655983352e6c9d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-1e761bf0ed15eb82add1f03c780de8721c3c8ac4)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
matzew: /retest
slinkydeveloper: /assign
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328325902177669120/artifacts/line-cov.html#file0) | 63.3% | 58.6% | -4.8
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328325902177669120/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328325902177669120/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328325902177669120/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 0d3aaa95447d72d3a73a4750e18422f21ee71863 into 25f218b21aaf3fea656e484bd4655983352e6c9d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-ff0b7c4eff7561da079de566dd332ccdc82ae01e)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328334380367613952/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328334380367613952/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328334380367613952/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328334380367613952/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

matzew: /retest
lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 7ac2c739fe4723290389d5faaa39e8969658cbf7 into 25f218b21aaf3fea656e484bd4655983352e6c9d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-1d5269b11187cb2afa726cb659ff89cc0f1a4563)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
matzew: > Missing e2e test and we're good to go!

I will install some `strimzi-tls` and we run the (consolidated) tests w/ that 

slinkydeveloper: @matzew In the installed `config-kafka` CM, can you add **commented** the fields `authSecretNamespace` and `authSecretName`? I always find extremely useful the comments in the config maps
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328654962615390208/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328654962615390208/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328654962615390208/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328654962615390208/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging f729b491867a611437b42b87eaedb557b7a5af14 into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-cb56062b556dc2a6caac2c5a96cc47953006171a)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | f729b491867a611437b42b87eaedb557b7a5af14 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328654962560864256) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @matzew: 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 | f729b491867a611437b42b87eaedb557b7a5af14 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328654962560864256) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | f729b491867a611437b42b87eaedb557b7a5af14 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328654962539892736) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328675463626756096/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328675463626756096/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328675463626756096/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328675463626756096/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 338a11761b71d86b1246c059aee7b6a2b2c0cec8 into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-0cd483e4ed988efc94b2fa409e496a7b88fb27d4)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: 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 | f729b491867a611437b42b87eaedb557b7a5af14 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328654962539892736) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 338a11761b71d86b1246c059aee7b6a2b2c0cec8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328675463551258624) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @matzew: 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 | 338a11761b71d86b1246c059aee7b6a2b2c0cec8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328675463551258624) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 338a11761b71d86b1246c059aee7b6a2b2c0cec8 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328675463526092800) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328726261660389376/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328726261660389376/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328726261660389376/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328726261660389376/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging cc42b6c75e86bb897d25d34f3b3dbed72e6824c6 into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-35dea15c367713a326db34247e7f0b83d6e40dc1)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328774696115638272/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328774696115638272/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328774696115638272/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328774696115638272/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

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

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

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

<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 -->
lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 6767a6115f308a8ce5118671ad85c9136db2b342 into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-6713fdddd822621565af630aeec2c5b3b28366ab)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 6767a6115f308a8ce5118671ad85c9136db2b342 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-unit-tests/1328774696086278144) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 6767a6115f308a8ce5118671ad85c9136db2b342 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328774696014974976) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328788452187049984/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328788452187049984/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328788452187049984/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328788452187049984/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 4f1fe4b3fb7f34b5df3ba67b837dbe8eedcbb6ef into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-212fd488400ef96da120a060a20c308431986c82)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: 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 | 6767a6115f308a8ce5118671ad85c9136db2b342 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328774696014974976) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 4f1fe4b3fb7f34b5df3ba67b837dbe8eedcbb6ef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328788452128329728) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @matzew: 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 | 4f1fe4b3fb7f34b5df3ba67b837dbe8eedcbb6ef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328788452128329728) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 4f1fe4b3fb7f34b5df3ba67b837dbe8eedcbb6ef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328788452111552512) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328868354412253184/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328868354412253184/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328868354412253184/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328868354412253184/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 78619bc756ac8aa9ab6de4c457ff81df55c434cf into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-8777d19c5dc702ae4da0762b481015eab39d2f85)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

<details>

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

On Wed 18. Nov 2020 at 02:40, Knative Prow Robot <notifications@github.com>
wrote:

> @matzew <https://github.com/matzew>: The following test *failed*, say
> /retest to rerun all failed tests:
> Test name Commit Details Rerun command
> pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated
> 78619bc
> <https://github.com/knative-sandbox/eventing-kafka/commit/78619bc756ac8aa9ab6de4c457ff81df55c434cf>
> link
> <https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328868354340950016> /test
> pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated
>
> Full PR test history
> <https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/192>. Your
> PR dashboard <https://gubernator.knative.dev/pr/matzew>.
>
> 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>.
>
> —
>
>
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/knative-sandbox/eventing-kafka/pull/192#issuecomment-729320542>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AABGPTVFF43LMMJHEWXLM53SQMQXPANCNFSM4TUZ6SEQ>
> .
>
-- 
Sent from Gmail Mobile

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

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

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328885562874007552/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328885562874007552/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328885562874007552/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328885562874007552/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging f24d18c980fcaed43decbaa73f3000457f69df05 into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-77b9d4243ec53c4376abf592f58d6bb8ff685c68)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
knative-prow-robot: @matzew: The following test **failed**, say `/retest` to rerun all failed tests:

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

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

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

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

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

<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: @matzew: 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 | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328908807090212864) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-build-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-build-tests/1328908807127961600) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

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

<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: @matzew: 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 | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328908807090212864) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-build-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-build-tests/1328908807127961600) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-unit-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-unit-tests/1328908807144738816) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<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: @matzew: 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 | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328908807090212864) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-build-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-build-tests/1328908807127961600) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-unit-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-unit-tests/1328908807144738816) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328908807106990080) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

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

<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: @matzew: 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 | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1328908807090212864) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-build-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-build-tests/1328908807127961600) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-unit-tests | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-unit-tests/1328908807144738816) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1328908807106990080) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-go-coverage | 377320d425c0c505bb0e9777335355b3d803548f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328908807161516032) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<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 -->
lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 377320d425c0c505bb0e9777335355b3d803548f into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-f3a83cebdfc31e26a7e6f8a4979a4e6d078e5271)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328939810622017536/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328939810622017536/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328939810622017536/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328939810622017536/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 895e4670c19bda716ea690065746c231ad134daf into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-371d4aa7154876b059845b3956f362f2932bfe72)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328955450565070848/artifacts/line-cov.html#file0) | 63.3% | 59.3% | -4.0
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328955450565070848/artifacts/line-cov.html#file1) | 54.2% | 53.5% | -0.7
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328955450565070848/artifacts/line-cov.html#file2) | 90.5% | 94.7% | 4.3
[pkg/common/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/192/pull-knative-sandbox-eventing-kafka-go-coverage/1328955450565070848/artifacts/line-cov.html#file3) | Do not exist | 55.9% | 

lgtm-com[bot]: This pull request **introduces 1 alert** and **fixes 1** when merging 68245527752052d31190cd713aa22b225c4cfc1a into 37e721f972c757ac14dbee6b01384f0b6a60308d - [view on LGTM.com](https://lgtm.com/projects/g/knative-sandbox/eventing-kafka/rev/pr-f38df0f63d18d2f597ff415fcd11f31b36ccd904)

**new alerts:**

* 1 for Disabled TLS certificate check

**fixed alerts:**

* 1 for Disabled TLS certificate check
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/192#" title="Author self-approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/192#pullrequestreview-533201094" title="Approved">slinkydeveloper</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)~~ [matzew,slinkydeveloper]

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":[]} -->

Patch:
diff --git a/config/channel/consolidated/configmaps/kafka-config.yaml b/config/channel/consolidated/configmaps/kafka-config.yaml
index 81b2d2cb5..8303d4d51 100644
--- a/config/channel/consolidated/configmaps/kafka-config.yaml
+++ b/config/channel/consolidated/configmaps/kafka-config.yaml
@@ -21,3 +21,5 @@ data:
   # Broker URL. Replace this with the URLs for your kafka cluster,
   # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
   bootstrapServers: REPLACE_WITH_CLUSTER_URL
+  #authSecretName: name-of-your-secret-for-kafka-auth
+  #authSecretNamespace: namespace-of-your-secret-for-kafka-auth
diff --git a/config/channel/consolidated/roles/controller-clusterrole.yaml b/config/channel/consolidated/roles/controller-clusterrole.yaml
index b9ba99c26..6b4dc9867 100644
--- a/config/channel/consolidated/roles/controller-clusterrole.yaml
+++ b/config/channel/consolidated/roles/controller-clusterrole.yaml
@@ -41,6 +41,7 @@ rules:
     resources:
       - services
       - configmaps
+      - secrets
     verbs:
       - get
       - list
diff --git a/config/channel/consolidated/roles/dispatcher-clusterrole.yaml b/config/channel/consolidated/roles/dispatcher-clusterrole.yaml
index bfe23095c..fe430fc7a 100644
--- a/config/channel/consolidated/roles/dispatcher-clusterrole.yaml
+++ b/config/channel/consolidated/roles/dispatcher-clusterrole.yaml
@@ -57,6 +57,7 @@ rules:
       - ""
     resources:
       - configmaps
+      - secrets
     verbs:
       - get
       - list
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index 237da189c..09d9af74b 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -35,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	kafkaclient "knative.dev/eventing-kafka/pkg/common"
 	"knative.dev/eventing-kafka/pkg/common/consumer"
 	eventingchannels "knative.dev/eventing/pkg/channel"
 	"knative.dev/eventing/pkg/channel/fanout"
@@ -93,6 +94,25 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 	conf.Consumer.Return.Errors = true    // Returns the errors in ConsumerGroup#Errors() https://godoc.org/github.com/Shopify/sarama#ConsumerGroup
 	conf.Producer.Return.Successes = true // Must be enabled for sync producer
 
+	// Get the auth info
+	if args.KafkaAuthConfig != nil {
+		// tls
+		if args.KafkaAuthConfig.TLS != nil {
+			conf.Net.TLS.Enable = true
+			tlsConfig, err := kafkaclient.NewTLSConfig(args.KafkaAuthConfig.TLS.Usercert, args.KafkaAuthConfig.TLS.Userkey, args.KafkaAuthConfig.TLS.Cacert)
+			if err != nil {
+				return nil, err
+			}
+			conf.Net.TLS.Config = tlsConfig
+		}
+		// SASL
+		if args.KafkaAuthConfig.SASL != nil {
+			conf.Net.SASL.Enable = true
+			conf.Net.SASL.User = args.KafkaAuthConfig.SASL.User
+			conf.Net.SASL.Password = args.KafkaAuthConfig.SASL.Password
+		}
+	}
+
 	producer, err := sarama.NewSyncProducer(args.Brokers, conf)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create kafka producer against Kafka bootstrap servers %v : %v", args.Brokers, err)
@@ -160,6 +180,7 @@ type KafkaDispatcherArgs struct {
 	KnCEConnectionArgs *kncloudevents.ConnectionArgs
 	ClientID           string
 	Brokers            []string
+	KafkaAuthConfig    *utils.KafkaAuthConfig
 	TopicFunc          TopicFunc
 	Logger             *zap.SugaredLogger
 }
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index 2e209687b..a5fc196e9 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -21,6 +21,8 @@ import (
 	"errors"
 	"fmt"
 
+	"knative.dev/eventing-kafka/pkg/common"
+
 	"k8s.io/utils/pointer"
 
 	"github.com/Shopify/sarama"
@@ -111,6 +113,7 @@ type Reconciler struct {
 	dispatcherImage string
 
 	kafkaConfig      *utils.KafkaConfig
+	kafkaAuthConfig  *utils.KafkaAuthConfig
 	kafkaConfigError error
 	kafkaClientSet   kafkaclientset.Interface
 
@@ -469,7 +472,7 @@ func (r *Reconciler) createClient(ctx context.Context, kc *v1beta1.KafkaChannel)
 	kafkaClusterAdmin := r.kafkaClusterAdmin
 	if kafkaClusterAdmin == nil {
 		var err error
-		kafkaClusterAdmin, err = resources.MakeClient(controllerAgentName, r.kafkaConfig.Brokers)
+		kafkaClusterAdmin, err = common.MakeAdminClient(controllerAgentName, r.kafkaAuthConfig, r.kafkaConfig.Brokers)
 		if err != nil {
 			return nil, err
 		}
@@ -518,6 +521,11 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	if err != nil {
 		logging.FromContext(ctx).Errorw("Error reading Kafka configuration", zap.Error(err))
 	}
+
+	if kafkaConfig.AuthSecretName != "" {
+		kafkaAuthConfig := utils.GetKafkaAuthData(ctx, kafkaConfig.AuthSecretName, kafkaConfig.AuthSecretNamespace)
+		r.kafkaAuthConfig = kafkaAuthConfig
+	}
 	// For now just override the previous config.
 	// Eventually the previous config should be snapshotted to delete Kafka topics
 	r.kafkaConfig = kafkaConfig
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/client.go b/pkg/channel/consolidated/reconciler/controller/resources/client.go
deleted file mode 100644
index d2af5852e..000000000
--- a/pkg/channel/consolidated/reconciler/controller/resources/client.go
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
-Copyright 2019 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 resources
-
-import (
-	"github.com/Shopify/sarama"
-)
-
-func MakeClient(clientID string, bootstrapServers []string) (sarama.ClusterAdmin, error) {
-	saramaConf := sarama.NewConfig()
-	saramaConf.Version = sarama.V1_1_0_0
-	saramaConf.ClientID = clientID
-	return sarama.NewClusterAdmin(bootstrapServers, saramaConf)
-}
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index 58188374c..25824e22c 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -88,6 +88,8 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		logger.Fatalw("Error loading kafka config", zap.Error(err))
 	}
 
+	kafkaAuthCfg := utils.GetKafkaAuthData(ctx, kafkaConfig.AuthSecretName, kafkaConfig.AuthSecretNamespace)
+
 	connectionArgs := &kncloudevents.ConnectionArgs{
 		MaxIdleConns:        int(kafkaConfig.MaxIdleConns),
 		MaxIdleConnsPerHost: int(kafkaConfig.MaxIdleConnsPerHost),
@@ -98,6 +100,7 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		KnCEConnectionArgs: connectionArgs,
 		ClientID:           "kafka-ch-dispatcher",
 		Brokers:            kafkaConfig.Brokers,
+		KafkaAuthConfig:    kafkaAuthCfg,
 		TopicFunc:          utils.TopicName,
 		Logger:             logger,
 	}
diff --git a/pkg/channel/consolidated/utils/util.go b/pkg/channel/consolidated/utils/util.go
index a4423918e..d0895f2f5 100644
--- a/pkg/channel/consolidated/utils/util.go
+++ b/pkg/channel/consolidated/utils/util.go
@@ -17,10 +17,15 @@ limitations under the License.
 package utils
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 
@@ -29,9 +34,17 @@ import (
 
 const (
 	BrokerConfigMapKey           = "bootstrapServers"
+	AuthSecretName               = "authSecretName"
+	AuthSecretNamespace          = "authSecretNamespace"
 	MaxIdleConnectionsKey        = "maxIdleConns"
 	MaxIdleConnectionsPerHostKey = "maxIdleConnsPerHost"
 
+	TlsCacert    = "ca.crt"
+	TlsUsercert  = "user.crt"
+	TlsUserkey   = "user.key"
+	SaslUser     = "user"
+	SaslPassword = "password"
+
 	KafkaChannelSeparator = "."
 
 	knativeKafkaTopicPrefix = "knative-messaging-kafka"
@@ -44,6 +57,55 @@ type KafkaConfig struct {
 	Brokers             []string
 	MaxIdleConns        int32
 	MaxIdleConnsPerHost int32
+	AuthSecretName      string
+	AuthSecretNamespace string
+}
+
+type KafkaAuthConfig struct {
+	TLS  *KafkaTlsConfig
+	SASL *KafkaSaslConfig
+}
+
+type KafkaTlsConfig struct {
+	Cacert   string
+	Usercert string
+	Userkey  string
+}
+
+type KafkaSaslConfig struct {
+	User     string
+	Password string
+}
+
+func GetKafkaAuthData(ctx context.Context, secretname string, secretNS string) *KafkaAuthConfig {
+
+	k8sClient := kubeclient.Get(ctx)
+	secret, err := k8sClient.CoreV1().Secrets(secretNS).Get(ctx, secretname, metav1.GetOptions{})
+
+	if err != nil || secret == nil {
+		logging.FromContext(ctx).Errorf("Referenced Auth Secret not found")
+		return nil
+	}
+
+	kafkaAuthConfig := &KafkaAuthConfig{}
+	// check for TLS
+	if string(secret.Data[TlsCacert]) != "" {
+		tls := &KafkaTlsConfig{
+			Cacert:   string(secret.Data[TlsCacert]),
+			Usercert: string(secret.Data[TlsUsercert]),
+			Userkey:  string(secret.Data[TlsUserkey]),
+		}
+		kafkaAuthConfig.TLS = tls
+	}
+
+	if string(secret.Data[SaslUser]) != "" {
+		sasl := &KafkaSaslConfig{
+			User:     string(secret.Data[SaslUser]),
+			Password: string(secret.Data[SaslPassword]),
+		}
+		kafkaAuthConfig.SASL = sasl
+	}
+	return kafkaAuthConfig
 }
 
 // GetKafkaConfig returns the details of the Kafka cluster.
@@ -58,9 +120,13 @@ func GetKafkaConfig(configMap map[string]string) (*KafkaConfig, error) {
 	}
 
 	var bootstrapServers string
+	var authSecretNamespace string
+	var authSecretName string
 
 	err := configmap.Parse(configMap,
 		configmap.AsString(BrokerConfigMapKey, &bootstrapServers),
+		configmap.AsString(AuthSecretName, &authSecretName),
+		configmap.AsString(AuthSecretNamespace, &authSecretNamespace),
 		configmap.AsInt32(MaxIdleConnectionsKey, &config.MaxIdleConns),
 		configmap.AsInt32(MaxIdleConnectionsPerHostKey, &config.MaxIdleConnsPerHost),
 	)
@@ -78,6 +144,8 @@ func GetKafkaConfig(configMap map[string]string) (*KafkaConfig, error) {
 		}
 	}
 	config.Brokers = bootstrapServersSplitted
+	config.AuthSecretName = authSecretName
+	config.AuthSecretNamespace = authSecretNamespace
 
 	return config, nil
 }
diff --git a/pkg/source/client.go b/pkg/common/client.go
similarity index 78%
rename from pkg/source/client.go
rename to pkg/common/client.go
index 7e8795771..f3cf480d6 100644
--- a/pkg/source/client.go
+++ b/pkg/common/client.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package source
+package common
 
 import (
 	"context"
@@ -22,6 +22,8 @@ import (
 	"crypto/x509"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
 )
@@ -68,7 +70,7 @@ func NewConfig(ctx context.Context) ([]string, *sarama.Config, error) {
 
 	if env.Net.TLS.Enable {
 		cfg.Net.TLS.Enable = true
-		tlsConfig, err := newTLSConfig(env.Net.TLS.Cert, env.Net.TLS.Key, env.Net.TLS.CACert)
+		tlsConfig, err := NewTLSConfig(env.Net.TLS.Cert, env.Net.TLS.Key, env.Net.TLS.CACert)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -91,9 +93,9 @@ func NewProducer(ctx context.Context) (sarama.Client, error) {
 	return sarama.NewClient(bs, cfg)
 }
 
-// newTLSConfig returns a *tls.Config using the given ceClient cert, ceClient key,
+// NewTLSConfig returns a *tls.Config using the given ceClient cert, ceClient key,
 // and CA certificate. If none are appropriate, a nil *tls.Config is returned.
-func newTLSConfig(clientCert, clientKey, caCert string) (*tls.Config, error) {
+func NewTLSConfig(clientCert, clientKey, caCert string) (*tls.Config, error) {
 	valid := false
 
 	config := &tls.Config{}
@@ -125,6 +127,31 @@ func newTLSConfig(clientCert, clientKey, caCert string) (*tls.Config, error) {
 	return config, nil
 }
 
+func MakeAdminClient(clientID string, kafkaAuthCfg *utils.KafkaAuthConfig, bootstrapServers []string) (sarama.ClusterAdmin, error) {
+	saramaConf := sarama.NewConfig()
+	saramaConf.Version = sarama.V1_1_0_0
+	saramaConf.ClientID = clientID
+
+	if kafkaAuthCfg != nil {
+		// tls
+		if kafkaAuthCfg.TLS != nil {
+			saramaConf.Net.TLS.Enable = true
+			tlsConfig, err := NewTLSConfig(kafkaAuthCfg.TLS.Usercert, kafkaAuthCfg.TLS.Userkey, kafkaAuthCfg.TLS.Cacert)
+			if err != nil {
+				return nil, err
+			}
+			saramaConf.Net.TLS.Config = tlsConfig
+		}
+		// SASL
+		if kafkaAuthCfg.SASL != nil {
+			saramaConf.Net.SASL.Enable = true
+			saramaConf.Net.SASL.User = kafkaAuthCfg.SASL.User
+			saramaConf.Net.SASL.Password = kafkaAuthCfg.SASL.Password
+		}
+	}
+	return sarama.NewClusterAdmin(bootstrapServers, saramaConf)
+}
+
 // verifyCertSkipHostname verifies certificates in the same way that the
 // default TLS handshake does, except it skips hostname verification. It must
 // be used with InsecureSkipVerify.
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 0805f0331..f4f8d73ea 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -21,7 +21,7 @@ import (
 	"net/http"
 	"strings"
 
-	kafkasource "knative.dev/eventing-kafka/pkg/source"
+	"knative.dev/eventing-kafka/pkg/common"
 
 	"go.opencensus.io/trace"
 	"knative.dev/eventing/pkg/adapter/v2"
@@ -91,7 +91,7 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 	)
 
 	// init consumer group
-	addrs, config, err := kafkasource.NewConfig(context.Background())
+	addrs, config, err := common.NewConfig(context.Background())
 	if err != nil {
 		return fmt.Errorf("failed to create the config: %w", err)
 	}


Test patch:
diff --git a/pkg/channel/consolidated/utils/util_test.go b/pkg/channel/consolidated/utils/util_test.go
index 8a687f7ec..ad6d660a5 100644
--- a/pkg/channel/consolidated/utils/util_test.go
+++ b/pkg/channel/consolidated/utils/util_test.go
@@ -17,15 +17,32 @@ limitations under the License.
 package utils
 
 import (
+	"context"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+	clientcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	injectionclient "knative.dev/pkg/client/injection/kube/client"
+
 	"testing"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/google/go-cmp/cmp"
 	_ "knative.dev/pkg/system/testing"
 )
 
+type KubernetesAPI struct {
+	Client kubernetes.Interface
+}
+
+const (
+	secretName      = "kafka-auth"
+	secretNamespace = "my-namespace"
+)
+
 func TestGenerateTopicNameWithDot(t *testing.T) {
 	expected := "knative-messaging-kafka.channel-namespace.channel-name"
 	actual := TopicName(".", "channel-namespace", "channel-name")
@@ -42,6 +59,152 @@ func TestGenerateTopicNameWithHyphen(t *testing.T) {
 	}
 }
 
+func TestGetKafkaAuthData(t *testing.T) {
+
+	api := &KubernetesAPI{
+		Client: fake.NewSimpleClientset(),
+	}
+
+	testCases := map[string]struct {
+		corev1Input clientcorev1.CoreV1Interface
+		secretName  string
+		secretNS    string
+		cacert      string
+		usercert    string
+		userkey     string
+		user        string
+		password    string
+		expected    *KafkaAuthConfig
+	}{
+		"empty secret": {
+			corev1Input: api.Client.CoreV1(),
+			secretName:  secretName,
+			secretNS:    secretNamespace,
+			cacert:      "",
+			usercert:    "",
+			userkey:     "",
+			user:        "",
+			password:    "",
+			expected:    &KafkaAuthConfig{},
+		},
+		"wrong secret name": {
+			corev1Input: api.Client.CoreV1(),
+			secretName:  "wrong",
+			secretNS:    secretNamespace,
+			cacert:      "cacert",
+			usercert:    "usercert",
+			userkey:     "userkey",
+			user:        "user",
+			password:    "password",
+			expected:    nil,
+		},
+		"tls and sasl secret": {
+			corev1Input: api.Client.CoreV1(),
+			secretName:  secretName,
+			secretNS:    secretNamespace,
+			cacert:      "cacert",
+			usercert:    "usercert",
+			userkey:     "userkey",
+			user:        "user",
+			password:    "password",
+			expected: &KafkaAuthConfig{
+				TLS: &KafkaTlsConfig{
+					Cacert:   "cacert",
+					Usercert: "usercert",
+					Userkey:  "userkey",
+				},
+				SASL: &KafkaSaslConfig{
+					User:     "user",
+					Password: "password",
+				},
+			},
+		},
+		"tls secret": {
+			corev1Input: api.Client.CoreV1(),
+			secretName:  secretName,
+			secretNS:    secretNamespace,
+			cacert:      "cacert",
+			usercert:    "usercert",
+			userkey:     "userkey",
+			user:        "",
+			password:    "",
+			expected: &KafkaAuthConfig{
+				TLS: &KafkaTlsConfig{
+					Cacert:   "cacert",
+					Usercert: "usercert",
+					Userkey:  "userkey",
+				},
+			},
+		},
+		"sasl secret": {
+			corev1Input: api.Client.CoreV1(),
+			secretName:  secretName,
+			secretNS:    secretNamespace,
+			cacert:      "",
+			usercert:    "",
+			userkey:     "",
+			user:        "user",
+			password:    "password",
+			expected: &KafkaAuthConfig{
+				TLS: nil,
+				SASL: &KafkaSaslConfig{
+					User:     "user",
+					Password: "password",
+				},
+			},
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+
+			// set up tgt namespace and service account to copy secret into.
+			kafkAuthSecretNS := tc.corev1Input.Namespaces()
+			namespace := &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: secretNamespace,
+				}}
+
+			_, nsCreateError := kafkAuthSecretNS.Create(context.Background(), namespace, metav1.CreateOptions{})
+			if nsCreateError != nil {
+				t.Error("error creating namespace for test case:", nsCreateError)
+			}
+
+			// set up secret
+			kafkAuthSecret := tc.corev1Input.Secrets(secretNamespace)
+			secret := createSecret(tc.cacert, tc.usercert, tc.userkey, tc.user, tc.password)
+			_, secretCreateError := kafkAuthSecret.Create(context.Background(), secret, metav1.CreateOptions{})
+			if secretCreateError != nil {
+				t.Error("error creating secret resources for test case:", secretCreateError)
+			}
+
+			ctx := context.WithValue(context.Background(), injectionclient.Key{}, fake.NewSimpleClientset(secret, namespace))
+			receivedSecret := GetKafkaAuthData(ctx, tc.secretName, tc.secretNS)
+
+			if tc.expected == nil {
+				if receivedSecret != nil {
+					t.Errorf("something is wrong")
+				}
+			} else {
+				if tc.expected.TLS != nil {
+					if receivedSecret.TLS == nil {
+						t.Errorf("TLS wrong")
+					}
+				}
+				if tc.expected.SASL != nil {
+					if receivedSecret.SASL == nil {
+						t.Errorf("SASL wrong")
+					}
+				}
+			}
+
+			// clean up after test
+			tc.corev1Input.Secrets(secretNamespace).Delete(context.Background(), secretName, metav1.DeleteOptions{})
+			tc.corev1Input.Namespaces().Delete(context.Background(), secretNamespace, metav1.DeleteOptions{})
+		})
+	}
+}
+
 func TestGetKafkaConfig(t *testing.T) {
 
 	testCases := []struct {
@@ -79,6 +242,17 @@ func TestGetKafkaConfig(t *testing.T) {
 				MaxIdleConnsPerHost: 100,
 			},
 		},
+		{
+			name: "single bootstrapServers and auth secret name and namespace",
+			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "authSecretName": "kafka-auth-secret", "authSecretNamespace": "default"},
+			expected: &KafkaConfig{
+				Brokers:             []string{"kafkabroker.kafka:9092"},
+				MaxIdleConns:        1000,
+				MaxIdleConnsPerHost: 100,
+				AuthSecretName:      "kafka-auth-secret",
+				AuthSecretNamespace: "default",
+			},
+		},
 		{
 			name: "multiple bootstrapServers",
 			data: map[string]string{"bootstrapServers": "kafkabroker1.kafka:9092,kafkabroker2.kafka:9092"},
@@ -140,7 +314,6 @@ func TestGetKafkaConfig(t *testing.T) {
 
 		})
 	}
-
 }
 
 func TestFindContainer(t *testing.T) {
@@ -213,3 +386,19 @@ func TestFindContainer(t *testing.T) {
 	}
 
 }
+
+func createSecret(cacert string, usercert string, userkey string, user string, password string) *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: secretNamespace,
+			Name:      secretName,
+		},
+		Data: map[string][]byte{
+			TlsCacert:    []byte(cacert),
+			TlsUsercert:  []byte(usercert),
+			TlsUserkey:   []byte(userkey),
+			SaslUser:     []byte(user),
+			SaslPassword: []byte(password),
+		},
+	}
+}
diff --git a/pkg/source/client_test.go b/pkg/common/client_test.go
similarity index 88%
rename from pkg/source/client_test.go
rename to pkg/common/client_test.go
index 296a32359..c0c13e378 100644
--- a/pkg/source/client_test.go
+++ b/pkg/common/client_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package source
+package common
 
 import (
 	"bytes"
@@ -29,6 +29,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Shopify/sarama"
+
 	"github.com/stretchr/testify/require"
 )
 
@@ -79,7 +81,7 @@ func TestNewTLSConfig(t *testing.T) {
 		wantServer: true,
 	}} {
 		t.Run(tt.name, func(t *testing.T) {
-			c, err := newTLSConfig(tt.cert, tt.key, tt.caCert)
+			c, err := NewTLSConfig(tt.cert, tt.key, tt.caCert)
 			if tt.wantErr {
 				if err == nil {
 					t.Fatal("wanted error")
@@ -220,3 +222,24 @@ func TestNewConfig(t *testing.T) {
 	require.NotNil(t, config)
 	require.Equal(t, []string{"my-cluster-kafka-bootstrap.my-kafka-namespace:9092"}, servers)
 }
+
+func TestAdminClient(t *testing.T) {
+
+	seedBroker := sarama.NewMockBroker(t, 1)
+	defer seedBroker.Close()
+
+	seedBroker.SetHandlerByMap(map[string]sarama.MockResponse{
+		"MetadataRequest": sarama.NewMockMetadataResponse(t).
+			SetController(seedBroker.BrokerID()).
+			SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
+	})
+
+	// mock broker does not support TLS ...
+	admin, err := MakeAdminClient("test-client", nil, []string{seedBroker.Addr()})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	require.NoError(t, err)
+	require.NotNil(t, admin)
+}
diff --git a/test/config/100-kafka-ephemeral-triple-2.6.0.yaml b/test/config/100-kafka-ephemeral-triple-2.6.0.yaml
index 25d174989..018521dc9 100644
--- a/test/config/100-kafka-ephemeral-triple-2.6.0.yaml
+++ b/test/config/100-kafka-ephemeral-triple-2.6.0.yaml
@@ -15,6 +15,14 @@ spec:
         port: 9093
         type: internal
         tls: true
+        authentication:
+          type: tls
+      - name: sasl
+        port: 9094
+        type: internal
+        tls: true
+        authentication:
+          type: scram-sha-512
     config:
       offsets.topic.replication.factor: 3
       transaction.state.log.replication.factor: 3
diff --git a/test/config/100-strimzi-users-0.20.0.yaml b/test/config/100-strimzi-users-0.20.0.yaml
new file mode 100644
index 000000000..f495e32b5
--- /dev/null
+++ b/test/config/100-strimzi-users-0.20.0.yaml
@@ -0,0 +1,44 @@
+apiVersion: kafka.strimzi.io/v1beta1
+kind: KafkaUser
+metadata:
+  name: my-tls-user
+  labels:
+    strimzi.io/cluster: my-cluster
+spec:
+  authentication:
+    type: tls
+  authorization:
+    type: simple
+    acls:
+      # Example ACL rules for consuming from knative-messaging-kafka using consumer group my-group
+      - resource:
+          type: topic
+          name: "*"
+        operation: Read
+        host: "*"
+      - resource:
+          type: topic
+          name: "*"
+        operation: Describe
+        host: "*"
+      - resource:
+          type: group
+          name: "*"
+        operation: Read
+        host: "*"
+      # Example ACL rules for producing to topic knative-messaging-kafka
+      - resource:
+          type: topic
+          name: "*"
+        operation: Write
+        host: "*"
+      - resource:
+          type: topic
+          name: "*"
+        operation: Create
+        host: "*"
+      - resource:
+          type: topic
+          name: "*"
+        operation: Describe
+        host: "*"
diff --git a/test/config/config-kafka-sasl.yaml b/test/config/config-kafka-sasl.yaml
new file mode 100644
index 000000000..301d9f05e
--- /dev/null
+++ b/test/config/config-kafka-sasl.yaml
@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka
+  namespace: knative-eventing
+data:
+  # Broker URL. Replace this with the URLs for your kafka cluster,
+  # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
+  bootstrapServers: REPLACE_WITH_CLUSTER_URL
+  authSecretName: strimzi-sasl-secret
+  authSecretNamespace: knative-eventing
diff --git a/test/config/config-kafka-tls.yaml b/test/config/config-kafka-tls.yaml
new file mode 100644
index 000000000..f08472aca
--- /dev/null
+++ b/test/config/config-kafka-tls.yaml
@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka
+  namespace: knative-eventing
+data:
+  # Broker URL. Replace this with the URLs for your kafka cluster,
+  # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
+  bootstrapServers: REPLACE_WITH_CLUSTER_URL
+  authSecretName: strimzi-tls-secret
+  authSecretNamespace: knative-eventing
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 55993d232..c3c3d82ed 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -88,8 +88,16 @@ readonly STRIMZI_INSTALLATION_CONFIG_TEMPLATE="test/config/100-strimzi-cluster-o
 readonly STRIMZI_INSTALLATION_CONFIG="$(mktemp)"
 # Kafka cluster CR config file.
 readonly KAFKA_INSTALLATION_CONFIG="test/config/100-kafka-ephemeral-triple-2.6.0.yaml"
-# Kafka cluster URL for our installation
-readonly KAFKA_CLUSTER_URL="my-cluster-kafka-bootstrap.kafka:9092"
+# Kafka TLS ConfigMap.
+readonly KAFKA_TLS_CONFIG="test/config/config-kafka-tls.yaml"
+# Kafka Users CR config file.
+readonly KAFKA_USERS_CONFIG="test/config/100-strimzi-users-0.20.0.yaml"
+# Kafka PLAIN cluster URL
+readonly KAFKA_PLAIN_CLUSTER_URL="my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"
+# Kafka TLS cluster URL
+readonly KAFKA_TLS_CLUSTER_URL="my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093"
+# Kafka cluster URL for our installation, during tests
+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.
@@ -325,6 +333,9 @@ function kafka_setup() {
 
   # Wait For The Strimzi Kafka Cluster Operator To Be Ready (Forcing Delay To Ensure CRDs Are Installed To Prevent Race Condition)
   wait_until_pods_running "${STRIMZI_KAFKA_NAMESPACE}" || fail_test "Failed to start up a Strimzi Kafka Instance"
+
+  # Create some Strimzi Kafka Users
+  kubectl apply -f "${KAFKA_USERS_CONFIG}" -n "${STRIMZI_KAFKA_NAMESPACE}"
 }
 
 function kafka_teardown() {
@@ -334,10 +345,22 @@ function kafka_teardown() {
   kubectl delete namespace "${STRIMZI_KAFKA_NAMESPACE}"
 }
 
+function create_auth_secrets() {
+  echo "Creating TLS Kafka secret"
+  STRIMZI_CRT=$(kubectl -n kafka get secret my-cluster-cluster-ca-cert --template='{{index .data "ca.crt"}}' | base64 --decode )
+  TLSUSER_CRT=$(kubectl -n kafka get secret my-tls-user --template='{{index .data "user.crt"}}' | base64 --decode )
+  TLSUSER_KEY=$(kubectl -n kafka get secret my-tls-user --template='{{index .data "user.key"}}' | base64 --decode )
+
+  kubectl create secret --namespace knative-eventing generic strimzi-tls-secret \
+    --from-literal=ca.crt="$STRIMZI_CRT" \
+    --from-literal=user.crt="$TLSUSER_CRT" \
+    --from-literal=user.key="$TLSUSER_KEY"
+}
+
 # Installs the resources necessary to test the consolidated channel, runs those tests, and then cleans up those resources
-function test_consolidated_channel() {
-  # Test the consolidated channel
-  echo "Testing the consolidated channel"
+function test_consolidated_channel_plain() {
+  # Test the consolidated channel with no auth
+  echo "Testing the consolidated channel and source"
   install_consolidated_channel_crds || return 1
   install_consolidated_sources_crds || return 1
 
@@ -348,6 +371,20 @@ function test_consolidated_channel() {
   uninstall_channel_crds || return 1
 }
 
+function test_consolidated_channel_tls() {
+  # Test the consolidated channel with no auth
+  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"
+  KAFKA_CLUSTER_URL=${KAFKA_TLS_CLUSTER_URL}
+
+  install_consolidated_channel_crds || return 1
+
+  go_test_e2e -tags=e2e -timeout=40m -test.parallel=${TEST_PARALLEL} ./test/e2e -channels=messaging.knative.dev/v1beta1:KafkaChannel  || fail_test
+
+  uninstall_channel_crds || return 1
+}
+
 # Installs the resources necessary to test the distributed channel, runs those tests, and then cleans up those resources
 function test_distributed_channel() {
   # Test the distributed channel
@@ -401,7 +438,9 @@ fi
 export SYSTEM_NAMESPACE
 
 if [[ $TEST_CONSOLIDATED_CHANNEL == 1 ]]; then
-  test_consolidated_channel || exit 1
+  test_consolidated_channel_plain || exit 1
+  create_auth_secrets
+  test_consolidated_channel_tls || exit 1
 fi
 
 # Terminate any zipkin port-forward processes that are still present on the system
diff --git a/test/test_images/kafka-publisher/main.go b/test/test_images/kafka-publisher/main.go
index 3de776408..4450b6cd7 100644
--- a/test/test_images/kafka-publisher/main.go
+++ b/test/test_images/kafka-publisher/main.go
@@ -19,7 +19,7 @@ package main
 import (
 	"log"
 
-	"knative.dev/eventing-kafka/pkg/source"
+	"knative.dev/eventing-kafka/pkg/common"
 
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
@@ -45,7 +45,7 @@ func main() {
 	log.Printf("GOT: %v", s)
 
 	// Create a Kafka client from our Binding.
-	client, err := source.NewProducer(ctx)
+	client, err := common.NewProducer(ctx)
 	if err != nil {
 		log.Fatal(err.Error())
 	}