Repository: knative-sandbox/eventing-kafka-broker
Pull request #1134 (2021-08-11 07:41:31)

Channel api
First 'proper' PR from branch https://github.com/aliok/eventing-kafka-broker/tree/first-stab-at-kafka-ch

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Introduce channel API in the project
- Configure tooling for code generation for it
- Generate files for it
- Had issues with Golang linter like `Error: string `expected at least one of, got none` has 4 occurrences, make it a constant (goconst)`. Made the linter config same with knative/eventing

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
:gift: Introduce KafkaChannel API which is to be implemented
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["markusthoemmes"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808/artifacts/line-cov.html#file2) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1134](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4415366) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/e49a86c50d2dbd04dc2cceff1d117841095a5f71?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (e49a86c) will **decrease** coverage by `4.59%`.
> The diff coverage is `81.81%`.

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

```diff
@@             Coverage Diff              @@
##               main    #1134      +/-   ##
============================================
- Coverage     75.86%   71.26%   -4.60%     
============================================
  Files            87       35      -52     
  Lines          3033     1434    -1599     
  Branches        136        0     -136     
============================================
- Hits           2301     1022    -1279     
+ Misses          567      317     -250     
+ Partials        165       95      -70     
```

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

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ntrol-plane/pkg/apis/messaging/v1beta1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9yZWdpc3Rlci5nbw==) | `0.00% <0.00%> (ø)` | |
| [.../apis/messaging/v1beta1/kafka\_channel\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9rYWZrYV9jaGFubmVsX2xpZmVjeWNsZS5nbw==) | `85.71% <85.71%> (ø)` | |
| [...apis/messaging/v1beta1/kafka\_channel\_conversion.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9rYWZrYV9jaGFubmVsX2NvbnZlcnNpb24uZ28=) | `100.00% <100.00%> (ø)` | |
| [...g/apis/messaging/v1beta1/kafka\_channel\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9rYWZrYV9jaGFubmVsX2RlZmF1bHRzLmdv) | `100.00% <100.00%> (ø)` | |
| [.../pkg/apis/messaging/v1beta1/kafka\_channel\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9rYWZrYV9jaGFubmVsX3R5cGVzLmdv) | `100.00% <100.00%> (ø)` | |
| [...apis/messaging/v1beta1/kafka\_channel\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9rYWZrYV9jaGFubmVsX3ZhbGlkYXRpb24uZ28=) | `100.00% <100.00%> (ø)` | |
| [...fka/broker/core/security/CredentialsValidator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQ3JlZGVudGlhbHNWYWxpZGF0b3IuamF2YQ==) | | |
| [...enting/kafka/broker/receiver/main/ReceiverEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyRW52LmphdmE=) | | |
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | | |
| [...nting/kafka/broker/core/tracing/TracingConfig.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9UcmFjaW5nQ29uZmlnLmphdmE=) | | |
| ... and [60 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1134/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

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

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 4415366d437780c5adef156e7bed37e8802ee3dc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 4415366d437780c5adef156e7bed37e8802ee3dc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425361736105463808) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 4415366d437780c5adef156e7bed37e8802ee3dc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1425361735988023296) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | 4415366d437780c5adef156e7bed37e8802ee3dc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1425361735988023296) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 29a59c342e8d739d406e845e6d695cf9e59f53b3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425369390915784704) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1134#" title="Author self-approved">aliok</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **devguyio** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["devguyio"]} -->
aliok: There are strange issues with vendored files and license files in their modes.
Also some strange crap regards to comments.

I am investigating
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 64a59ca2fdc97177541552125a2d284993864e24 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425421835725443072) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000/artifacts/line-cov.html#file2) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 33378878196226412ebf360c6930f269f6c97399 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425425454180864000) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425709163928883200/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425709163928883200/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425709163928883200/artifacts/line-cov.html#file2) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425709163928883200/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425709163928883200/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425709163928883200/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

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

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

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240/artifacts/line-cov.html#file2) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 1dde51e220e4b0e46b4aac772b9b67577c5543dc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425710972764426240) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696/artifacts/line-cov.html#file2) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 3492cd84f9217b5205a22d69eada33b57145d85f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425726661826973696) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425808254138585088/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425808254138585088/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425808254138585088/artifacts/line-cov.html#file2) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425808254138585088/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425808254138585088/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1425808254138585088/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 

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

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

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 412d54be53bbfc531080b833a14935aefaeb504c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427888971236511744) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 6cc57663b43fe0947e48308edb67d732c71e92ff | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427893370323210240) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 424896ef25e4f6001e8c369f27b1996beacf5128 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427920769492455424) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 87f47de6edf2bffa9a0552fbde5c2a031ed81af2 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427979238501257216) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 4726c321124ac57a39eb44978ea502913603dc02 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1134#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1134#pullrequestreview-734696757" title="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-broker).

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-broker/blob/main/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":[]} -->
matzew: I'd recommend we fix those NITs upstream too, eventually we will just consume the APIs ? 
matzew: While this is all WIP etc, I'd like to at least get those things addressed that are also missing upstream.

In the shorter run, we will be reusing the API from that repo, instead of maintaining a copy here
devguyio: @matzew @aliok I'm totally fine if we unhold and have that merged on Monday, I consider my current review as an audit sort of review and will keep leaving the comments for reference of what needs to be done. 
matzew: +1

The hard coupling to versions is IMO a stupid construct anyways.

On Fri 20. Aug 2021 at 18:27, Ahmed Abdalla Abdelrehim <
***@***.***> wrote:

> ***@***.**** commented on this pull request.
> ------------------------------
>
> In control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go
> <https://github.com/knative-sandbox/eventing-kafka-broker/pull/1134#discussion_r693072420>
> :
>
> > +
> +package v1beta1
> +
> +import (
> +	"context"
> +	"fmt"
> +
> +	"knative.dev/eventing/pkg/apis/eventing"
> +	"knative.dev/pkg/apis"
> +)
> +
> +func (c *KafkaChannel) Validate(ctx context.Context) *apis.FieldError {
> +	errs := c.Spec.Validate(ctx).ViaField("spec")
> +
> +	// Validate annotations
> +	if c.Annotations != nil {
>
> we should follow the chain pattern that the FieldError was designed for.
> Basically the idea is to yank out all validation logic into functions under
> the top level, non-versioned directory (e.g.
> knative.dev/eventing-kafka/pkg/apis/messaging/ ) and use the chain
> pattern syntactic sugar to call those in the correct sequence. That way you
> decouple the validation logic from the API shape, and you reduce code
> duplication in case you want to change the validation code across all the
> API versions.
>
> An example of that can be found in serving autoscaling
> <https://github.com/knative/serving/blob/main/pkg/apis/autoscaling/v1alpha1/pa_validation.go#L34>
> for example.
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/knative-sandbox/eventing-kafka-broker/pull/1134#pullrequestreview-735196646>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AABGPTTT72NYQ3QZ4LTGPCDT5Z657ANCNFSM5B5YVCTQ>
> .
>
-- 
Sent from Gmail Mobile

matzew: > I would prefer doing this in a later PR, instead of doing everything in a single PR.

sounds good to me.

What I still would like to see is that we also address some of the comments here "upstream", with the hope we later are able to consume the API code from there 

knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 4726c321124ac57a39eb44978ea502913603dc02 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1427987598655098880) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429710523934445568) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429710523934445568) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429710523934445568) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429710523905085440) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429710523997360128) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429710523905085440) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429713289054523392) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | a4b2b7d3dc4a3d764e3ff347475ccd97aa0cde5a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429710523905085440) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429713289054523392) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429713289054523392) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429713289029357568) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429713289054523392) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429713289029357568) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429713289079689216) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429713289054523392) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429713289029357568) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429713289054523392) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429763432365690880) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 5dc86d6d85ae4065ac715e494e3b7cd442402617 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429713289104855040) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429763432365690880) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429763432390856704) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429763432365690880) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429763432390856704) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429763432445382656) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429763432365690880) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429763432445382656) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429766090069643264) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429763432365690880) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429766090069643264) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429766090069643264) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429766090048671744) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

matzew: > That's mostly independent work, that anybody can do anyway.

@aliok I can help with that! As well as the PRs to upstream?

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429766090132557824) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429766090048671744) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429771092288868352) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

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

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 3ab9a1f5631314f90c42183d8ba958dcb9f65ce1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429766090048671744) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429771092288868352) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aliok: > > That's mostly independent work, that anybody can do anyway.
> 
> @aliok I can help with that! As well as the PRs to upstream?

Thanks, would be awesome!
matzew: the PR itself gets me this:

```
=== Errors
control-plane/pkg/client/injection/client/client.go:82:5: cannot use (*wrapClient)(nil) (type *wrapClient) as type versioned.Interface in assignment:
	*wrapClient does not implement versioned.Interface (missing MessagingV1beta1 method)
```
knative-prow-robot: @aliok: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429771092288868352) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429771092343394304) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429771092255313920) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 063e873017a3ece62f51a8c8058651d831c22fa4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1429763432420216832) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1429771092288868352) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 063c39a1c0832c8faa8e2db0c2b8d4b364396cb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-build-tests/1429771092255313920) | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 936c5303e47b1f6d5c7851a8099f455c7c442004 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429774797499273216) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840/artifacts/line-cov.html#file4) | Do not exist | 84.6% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[control-plane/pkg/apis/messaging/v1beta1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | 936c5303e47b1f6d5c7851a8099f455c7c442004 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1134/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429792673962659840) | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<details>

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

Patch:
diff --git a/.github/workflows/knative-boilerplate.yaml b/.github/workflows/knative-boilerplate.yaml
index f901369ff..fb38d054f 100644
--- a/.github/workflows/knative-boilerplate.yaml
+++ b/.github/workflows/knative-boilerplate.yaml
@@ -85,7 +85,7 @@ jobs:
           boilerplate-check check \
             --boilerplate ./hack/boilerplate/boilerplate.${{ matrix.extension }}.txt \
             --file-extension ${{ matrix.extension }} \
-            --exclude "(vendor|third_party)/" |
+            --exclude "(vendor|third_party|control-plane/pkg/contract)/" |
           reviewdog -efm="%A%f:%l: %m" \
                 -efm="%C%.%#" \
                 -name="${{ matrix.language }} headers" \
diff --git a/.golangci.yaml b/.golangci.yaml
index bf3f42d95..903302d87 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -14,23 +14,8 @@ linters:
     - prealloc
     - unconvert
     - unparam
-    - govet
-    - deadcode
-    - gosimple
-    - ineffassign
-    - staticcheck
-    - structcheck
-    - typecheck
-    - unconvert
-    - unused
-    - varcheck
+  disable:
     - errcheck
-    - godot
-    - goconst
-    - gofmt
-    - goimports
-    - gomnd
-    - exhaustive
 
 issues:
   exclude-rules:
@@ -38,5 +23,3 @@ issues:
       linters:
         - gosec
         - unparam
-        - gomnd
-        - godot
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 9796a2649..a8648d8af 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package main
 
diff --git a/control-plane/pkg/apis/messaging/register.go b/control-plane/pkg/apis/messaging/register.go
new file mode 100644
index 000000000..056072ee9
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/register.go
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package messaging
+
+import "k8s.io/apimachinery/pkg/runtime/schema"
+
+const (
+	GroupName = "messaging.knative.dev"
+)
+
+var (
+	// KafkaChannelsResource represents a KafkaChannel
+	KafkaChannelsResource = schema.GroupResource{
+		Group:    GroupName,
+		Resource: "kafkachannels",
+	}
+)
diff --git a/control-plane/pkg/apis/messaging/v1beta1/doc.go b/control-plane/pkg/apis/messaging/v1beta1/doc.go
new file mode 100644
index 000000000..6c4a66d2c
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/doc.go
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package v1beta1 is the v1beta1 version of the API.
+// +k8s:deepcopy-gen=package
+// +groupName=messaging.knative.dev
+package v1beta1
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go
new file mode 100644
index 000000000..e1269c0a6
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion.go
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/pkg/apis"
+)
+
+// ConvertTo implements apis.Convertible
+func (channel *KafkaChannel) ConvertTo(_ context.Context, sink apis.Convertible) error {
+	return fmt.Errorf("v1beta1 is the highest known version, got: %T", sink)
+}
+
+// ConvertFrom implements apis.Convertible
+func (sink *KafkaChannel) ConvertFrom(_ context.Context, channel apis.Convertible) error {
+	return fmt.Errorf("v1beta1 is the highest known version, got: %T", channel)
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
new file mode 100644
index 000000000..71041c32f
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+
+	"knative.dev/eventing/pkg/apis/messaging"
+)
+
+const (
+	// DefaultNumPartitions is the KafkaChannel Spec default for the number of partitions
+	DefaultNumPartitions = 1
+	// DefaultReplicationFactor is the KafkaChannel Spec default for the replication factor
+	DefaultReplicationFactor = 1
+)
+
+func (c *KafkaChannel) SetDefaults(ctx context.Context) {
+	// Set the duck subscription to the stored version of the duck
+	// we support. Reason for this is that the stored version will
+	// not get a chance to get modified, but for newer versions
+	// conversion webhook will be able to take a crack at it and
+	// can modify it to match the duck shape.
+	if c.Annotations == nil {
+		c.Annotations = make(map[string]string)
+	}
+
+	if _, ok := c.Annotations[messaging.SubscribableDuckVersionAnnotation]; !ok {
+		c.Annotations[messaging.SubscribableDuckVersionAnnotation] = "v1"
+	}
+
+	c.Spec.SetDefaults(ctx)
+}
+
+func (cs *KafkaChannelSpec) SetDefaults(ctx context.Context) {
+	if cs.NumPartitions == 0 {
+		cs.NumPartitions = DefaultNumPartitions
+	}
+	if cs.ReplicationFactor == 0 {
+		cs.ReplicationFactor = DefaultReplicationFactor
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go
new file mode 100644
index 000000000..2eb14dde1
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+// ConditionAddressable is a reference to the Addressable condition in the base package.
+// KafkaChannel uses the common Condition set defined in the base package.
+const (
+	ConditionAddressable apis.ConditionType = "Addressable"
+)
+
+var conditionSet apis.ConditionSet
+
+func RegisterConditionSet(cs apis.ConditionSet) {
+	conditionSet = cs
+}
+
+// GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
+func (*KafkaChannel) GetConditionSet() apis.ConditionSet {
+	return conditionSet
+}
+
+// GetConditionSet retrieves the condition set for this resource.
+func (*KafkaChannelStatus) GetConditionSet() apis.ConditionSet {
+	return conditionSet
+}
+
+// GetCondition returns the condition currently associated with the given type, or nil.
+func (cs *KafkaChannelStatus) GetCondition(t apis.ConditionType) *apis.Condition {
+	return cs.GetConditionSet().Manage(cs).GetCondition(t)
+}
+
+// IsReady returns true if the resource is ready overall.
+func (cs *KafkaChannelStatus) IsReady() bool {
+	return cs.GetConditionSet().Manage(cs).IsHappy()
+}
+
+// InitializeConditions sets relevant unset conditions to Unknown state.
+func (cs *KafkaChannelStatus) InitializeConditions() {
+	cs.GetConditionSet().Manage(cs).InitializeConditions()
+}
+
+// SetAddress sets the address (as part of Addressable contract) and marks the correct condition.
+func (cs *KafkaChannelStatus) SetAddress(url *apis.URL) {
+	if cs.Address == nil {
+		cs.Address = &duckv1.Addressable{}
+	}
+	if url != nil {
+		cs.Address.URL = url
+		cs.GetConditionSet().Manage(cs).MarkTrue(ConditionAddressable)
+	} else {
+		cs.Address.URL = nil
+		cs.GetConditionSet().Manage(cs).MarkFalse(ConditionAddressable, "nil URL", "URL is nil")
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go
new file mode 100644
index 000000000..582f10d0f
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types.go
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
+)
+
+// +genclient
+// +genreconciler
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// KafkaChannel is a resource representing a Kafka Channel.
+type KafkaChannel struct {
+	metav1.TypeMeta `json:",inline"`
+	// +optional
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	// Spec defines the desired state of the Channel.
+	Spec KafkaChannelSpec `json:"spec,omitempty"`
+
+	// Status represents the current state of the KafkaChannel. This data may be out of
+	// date.
+	// +optional
+	Status KafkaChannelStatus `json:"status,omitempty"`
+}
+
+var (
+	// Check that this channel can be validated and defaulted.
+	_ apis.Validatable = (*KafkaChannel)(nil)
+	_ apis.Defaultable = (*KafkaChannel)(nil)
+
+	_ runtime.Object = (*KafkaChannel)(nil)
+
+	// Check that we can create OwnerReferences to a KafkaChannel.
+	_ kmeta.OwnerRefable = (*KafkaChannel)(nil)
+
+	// Check that the type conforms to the duck Knative Resource shape.
+	_ duckv1.KRShaped = (*KafkaChannel)(nil)
+)
+
+// KafkaChannelSpec defines the specification for a KafkaChannel.
+type KafkaChannelSpec struct {
+	// NumPartitions is the number of partitions of a Kafka topic. By default, it is set to 1.
+	NumPartitions int32 `json:"numPartitions"`
+
+	// ReplicationFactor is the replication factor of a Kafka topic. By default, it is set to 1.
+	ReplicationFactor int16 `json:"replicationFactor"`
+
+	// Channel conforms to Duck type ChannelableStatus.
+	eventingduck.ChannelableSpec `json:",inline"`
+}
+
+// KafkaChannelStatus represents the current state of a KafkaChannel.
+type KafkaChannelStatus struct {
+	// Channel conforms to Duck type ChannelableStatus.
+	eventingduck.ChannelableStatus `json:",inline"`
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// KafkaChannelList is a collection of KafkaChannels.
+type KafkaChannelList struct {
+	metav1.TypeMeta `json:",inline"`
+	// +optional
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []KafkaChannel `json:"items"`
+}
+
+// GetGroupVersionKind returns GroupVersionKind for KafkaChannels
+func (c *KafkaChannel) GetGroupVersionKind() schema.GroupVersionKind {
+	return SchemeGroupVersion.WithKind("KafkaChannel")
+}
+
+// GetStatus retrieves the duck status for this resource. Implements the KRShaped interface.
+func (k *KafkaChannel) GetStatus() *duckv1.Status {
+	return &k.Status.Status
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go
new file mode 100644
index 000000000..b15d8b0cf
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation.go
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/eventing/pkg/apis/eventing"
+	"knative.dev/pkg/apis"
+)
+
+func (c *KafkaChannel) Validate(ctx context.Context) *apis.FieldError {
+	errs := c.Spec.Validate(ctx).ViaField("spec")
+
+	// Validate annotations
+	if c.Annotations != nil {
+		if scope, ok := c.Annotations[eventing.ScopeAnnotationKey]; ok {
+			if scope != "namespace" && scope != "cluster" {
+				iv := apis.ErrInvalidValue(scope, "")
+				iv.Details = "expected either 'cluster' or 'namespace'"
+				errs = errs.Also(iv.ViaFieldKey("annotations", eventing.ScopeAnnotationKey).ViaField("metadata"))
+			}
+		}
+	}
+
+	return errs
+}
+
+func (cs *KafkaChannelSpec) Validate(ctx context.Context) *apis.FieldError {
+	var errs *apis.FieldError
+
+	if cs.NumPartitions <= 0 {
+		fe := apis.ErrInvalidValue(cs.NumPartitions, "numPartitions")
+		errs = errs.Also(fe)
+	}
+
+	if cs.ReplicationFactor <= 0 {
+		fe := apis.ErrInvalidValue(cs.ReplicationFactor, "replicationFactor")
+		errs = errs.Also(fe)
+	}
+
+	for i, subscriber := range cs.SubscribableSpec.Subscribers {
+		if subscriber.ReplyURI == nil && subscriber.SubscriberURI == nil {
+			fe := apis.ErrMissingField("replyURI", "subscriberURI")
+			fe.Details = "expected at least one of, got none"
+			errs = errs.Also(fe.ViaField(fmt.Sprintf("subscriber[%d]", i)).ViaField("subscribable"))
+		}
+	}
+	return errs
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/register.go b/control-plane/pkg/apis/messaging/v1beta1/register.go
new file mode 100644
index 000000000..ce862bbee
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/register.go
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging"
+)
+
+// SchemeGroupVersion is group version used to register these objects
+var SchemeGroupVersion = schema.GroupVersion{Group: messaging.GroupName, Version: "v1beta1"}
+
+// Kind takes an unqualified kind and returns back a Group qualified GroupKind
+func Kind(kind string) schema.GroupKind {
+	return SchemeGroupVersion.WithKind(kind).GroupKind()
+}
+
+// Resource takes an unqualified resource and returns a Group qualified GroupResource
+func Resource(resource string) schema.GroupResource {
+	return SchemeGroupVersion.WithResource(resource).GroupResource()
+}
+
+var (
+	SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
+	AddToScheme   = SchemeBuilder.AddToScheme
+)
+
+// Adds the list of known types to Scheme.
+func addKnownTypes(scheme *runtime.Scheme) error {
+	scheme.AddKnownTypes(SchemeGroupVersion,
+		&KafkaChannel{},
+		&KafkaChannelList{},
+	)
+	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
+	return nil
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go b/control-plane/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go
new file mode 100644
index 000000000..5a2bab0e4
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/zz_generated.deepcopy.go
@@ -0,0 +1,120 @@
+// +build !ignore_autogenerated
+
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by deepcopy-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	runtime "k8s.io/apimachinery/pkg/runtime"
+)
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *KafkaChannel) DeepCopyInto(out *KafkaChannel) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaChannel.
+func (in *KafkaChannel) DeepCopy() *KafkaChannel {
+	if in == nil {
+		return nil
+	}
+	out := new(KafkaChannel)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *KafkaChannel) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *KafkaChannelList) DeepCopyInto(out *KafkaChannelList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]KafkaChannel, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaChannelList.
+func (in *KafkaChannelList) DeepCopy() *KafkaChannelList {
+	if in == nil {
+		return nil
+	}
+	out := new(KafkaChannelList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *KafkaChannelList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *KafkaChannelSpec) DeepCopyInto(out *KafkaChannelSpec) {
+	*out = *in
+	in.ChannelableSpec.DeepCopyInto(&out.ChannelableSpec)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaChannelSpec.
+func (in *KafkaChannelSpec) DeepCopy() *KafkaChannelSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(KafkaChannelSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *KafkaChannelStatus) DeepCopyInto(out *KafkaChannelStatus) {
+	*out = *in
+	in.ChannelableStatus.DeepCopyInto(&out.ChannelableStatus)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaChannelStatus.
+func (in *KafkaChannelStatus) DeepCopy() *KafkaChannelStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(KafkaChannelStatus)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/control-plane/pkg/client/clientset/versioned/clientset.go b/control-plane/pkg/client/clientset/versioned/clientset.go
index 432dbc373..9afb5ac40 100644
--- a/control-plane/pkg/client/clientset/versioned/clientset.go
+++ b/control-plane/pkg/client/clientset/versioned/clientset.go
@@ -25,11 +25,13 @@ import (
 	rest "k8s.io/client-go/rest"
 	flowcontrol "k8s.io/client-go/util/flowcontrol"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1"
 )
 
 type Interface interface {
 	Discovery() discovery.DiscoveryInterface
 	EventingV1alpha1() eventingv1alpha1.EventingV1alpha1Interface
+	MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface
 }
 
 // Clientset contains the clients for groups. Each group has exactly one
@@ -37,6 +39,7 @@ type Interface interface {
 type Clientset struct {
 	*discovery.DiscoveryClient
 	eventingV1alpha1 *eventingv1alpha1.EventingV1alpha1Client
+	messagingV1beta1 *messagingv1beta1.MessagingV1beta1Client
 }
 
 // EventingV1alpha1 retrieves the EventingV1alpha1Client
@@ -44,6 +47,11 @@ func (c *Clientset) EventingV1alpha1() eventingv1alpha1.EventingV1alpha1Interfac
 	return c.eventingV1alpha1
 }
 
+// MessagingV1beta1 retrieves the MessagingV1beta1Client
+func (c *Clientset) MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface {
+	return c.messagingV1beta1
+}
+
 // Discovery retrieves the DiscoveryClient
 func (c *Clientset) Discovery() discovery.DiscoveryInterface {
 	if c == nil {
@@ -69,6 +77,10 @@ func NewForConfig(c *rest.Config) (*Clientset, error) {
 	if err != nil {
 		return nil, err
 	}
+	cs.messagingV1beta1, err = messagingv1beta1.NewForConfig(&configShallowCopy)
+	if err != nil {
+		return nil, err
+	}
 
 	cs.DiscoveryClient, err = discovery.NewDiscoveryClientForConfig(&configShallowCopy)
 	if err != nil {
@@ -82,6 +94,7 @@ func NewForConfig(c *rest.Config) (*Clientset, error) {
 func NewForConfigOrDie(c *rest.Config) *Clientset {
 	var cs Clientset
 	cs.eventingV1alpha1 = eventingv1alpha1.NewForConfigOrDie(c)
+	cs.messagingV1beta1 = messagingv1beta1.NewForConfigOrDie(c)
 
 	cs.DiscoveryClient = discovery.NewDiscoveryClientForConfigOrDie(c)
 	return &cs
@@ -91,6 +104,7 @@ func NewForConfigOrDie(c *rest.Config) *Clientset {
 func New(c rest.Interface) *Clientset {
 	var cs Clientset
 	cs.eventingV1alpha1 = eventingv1alpha1.New(c)
+	cs.messagingV1beta1 = messagingv1beta1.New(c)
 
 	cs.DiscoveryClient = discovery.NewDiscoveryClient(c)
 	return &cs
diff --git a/control-plane/pkg/client/clientset/versioned/fake/clientset_generated.go b/control-plane/pkg/client/clientset/versioned/fake/clientset_generated.go
index 9a24737f4..10aa84c0a 100644
--- a/control-plane/pkg/client/clientset/versioned/fake/clientset_generated.go
+++ b/control-plane/pkg/client/clientset/versioned/fake/clientset_generated.go
@@ -27,6 +27,8 @@ import (
 	clientset "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
 	fakeeventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1"
+	fakemessagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake"
 )
 
 // NewSimpleClientset returns a clientset that will respond with the provided objects.
@@ -80,3 +82,8 @@ var _ clientset.Interface = &Clientset{}
 func (c *Clientset) EventingV1alpha1() eventingv1alpha1.EventingV1alpha1Interface {
 	return &fakeeventingv1alpha1.FakeEventingV1alpha1{Fake: &c.Fake}
 }
+
+// MessagingV1beta1 retrieves the MessagingV1beta1Client
+func (c *Clientset) MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface {
+	return &fakemessagingv1beta1.FakeMessagingV1beta1{Fake: &c.Fake}
+}
diff --git a/control-plane/pkg/client/clientset/versioned/fake/register.go b/control-plane/pkg/client/clientset/versioned/fake/register.go
index e4475ff39..bae6056c1 100644
--- a/control-plane/pkg/client/clientset/versioned/fake/register.go
+++ b/control-plane/pkg/client/clientset/versioned/fake/register.go
@@ -25,6 +25,7 @@ import (
 	serializer "k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
 )
 
 var scheme = runtime.NewScheme()
@@ -32,6 +33,7 @@ var codecs = serializer.NewCodecFactory(scheme)
 
 var localSchemeBuilder = runtime.SchemeBuilder{
 	eventingv1alpha1.AddToScheme,
+	messagingv1beta1.AddToScheme,
 }
 
 // AddToScheme adds all types of this clientset into the given scheme. This allows composition
diff --git a/control-plane/pkg/client/clientset/versioned/scheme/register.go b/control-plane/pkg/client/clientset/versioned/scheme/register.go
index 0a193f88e..1db6ebef4 100644
--- a/control-plane/pkg/client/clientset/versioned/scheme/register.go
+++ b/control-plane/pkg/client/clientset/versioned/scheme/register.go
@@ -25,6 +25,7 @@ import (
 	serializer "k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
 )
 
 var Scheme = runtime.NewScheme()
@@ -32,6 +33,7 @@ var Codecs = serializer.NewCodecFactory(Scheme)
 var ParameterCodec = runtime.NewParameterCodec(Scheme)
 var localSchemeBuilder = runtime.SchemeBuilder{
 	eventingv1alpha1.AddToScheme,
+	messagingv1beta1.AddToScheme,
 }
 
 // AddToScheme adds all types of this clientset into the given scheme. This allows composition
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/doc.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/doc.go
new file mode 100644
index 000000000..4519a1e21
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/doc.go
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// This package has the automatically generated typed clients.
+package v1beta1
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go
new file mode 100644
index 000000000..7f850700e
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go
new file mode 100644
index 000000000..6b4898966
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go
@@ -0,0 +1,142 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+)
+
+// FakeKafkaChannels implements KafkaChannelInterface
+type FakeKafkaChannels struct {
+	Fake *FakeMessagingV1beta1
+	ns   string
+}
+
+var kafkachannelsResource = schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1beta1", Resource: "kafkachannels"}
+
+var kafkachannelsKind = schema.GroupVersionKind{Group: "messaging.knative.dev", Version: "v1beta1", Kind: "KafkaChannel"}
+
+// Get takes name of the kafkaChannel, and returns the corresponding kafkaChannel object, and an error if there is any.
+func (c *FakeKafkaChannels) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkachannelsResource, c.ns, name), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaChannels that match those selectors.
+func (c *FakeKafkaChannels) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaChannelList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkachannelsResource, kafkachannelsKind, c.ns, opts), &v1beta1.KafkaChannelList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaChannelList{ListMeta: obj.(*v1beta1.KafkaChannelList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaChannelList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaChannels.
+func (c *FakeKafkaChannels) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkachannelsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaChannel and creates it.  Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *FakeKafkaChannels) Create(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.CreateOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkachannelsResource, c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// Update takes the representation of a kafkaChannel and updates it. Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *FakeKafkaChannels) Update(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkachannelsResource, c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaChannels) UpdateStatus(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkachannelsResource, "status", c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// Delete takes name of the kafkaChannel and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaChannels) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkachannelsResource, c.ns, name), &v1beta1.KafkaChannel{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaChannels) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkachannelsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaChannelList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaChannel.
+func (c *FakeKafkaChannels) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkachannelsResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go
new file mode 100644
index 000000000..b690a9d6f
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1"
+)
+
+type FakeMessagingV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeMessagingV1beta1) KafkaChannels(namespace string) v1beta1.KafkaChannelInterface {
+	return &FakeKafkaChannels{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeMessagingV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/generated_expansion.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/generated_expansion.go
new file mode 100644
index 000000000..d77de2ebf
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/generated_expansion.go
@@ -0,0 +1,21 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1beta1
+
+type KafkaChannelExpansion interface{}
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/kafkachannel.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/kafkachannel.go
new file mode 100644
index 000000000..f1575a1cb
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/kafkachannel.go
@@ -0,0 +1,195 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	"context"
+	"time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	rest "k8s.io/client-go/rest"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	scheme "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/scheme"
+)
+
+// KafkaChannelsGetter has a method to return a KafkaChannelInterface.
+// A group's client should implement this interface.
+type KafkaChannelsGetter interface {
+	KafkaChannels(namespace string) KafkaChannelInterface
+}
+
+// KafkaChannelInterface has methods to work with KafkaChannel resources.
+type KafkaChannelInterface interface {
+	Create(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.CreateOptions) (*v1beta1.KafkaChannel, error)
+	Update(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error)
+	UpdateStatus(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error)
+	Delete(ctx context.Context, name string, opts v1.DeleteOptions) error
+	DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error
+	Get(ctx context.Context, name string, opts v1.GetOptions) (*v1beta1.KafkaChannel, error)
+	List(ctx context.Context, opts v1.ListOptions) (*v1beta1.KafkaChannelList, error)
+	Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error)
+	Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaChannel, err error)
+	KafkaChannelExpansion
+}
+
+// kafkaChannels implements KafkaChannelInterface
+type kafkaChannels struct {
+	client rest.Interface
+	ns     string
+}
+
+// newKafkaChannels returns a KafkaChannels
+func newKafkaChannels(c *MessagingV1beta1Client, namespace string) *kafkaChannels {
+	return &kafkaChannels{
+		client: c.RESTClient(),
+		ns:     namespace,
+	}
+}
+
+// Get takes name of the kafkaChannel, and returns the corresponding kafkaChannel object, and an error if there is any.
+func (c *kafkaChannels) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaChannel, err error) {
+	result = &v1beta1.KafkaChannel{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		Name(name).
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// List takes label and field selectors, and returns the list of KafkaChannels that match those selectors.
+func (c *kafkaChannels) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaChannelList, err error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	result = &v1beta1.KafkaChannelList{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaChannels.
+func (c *kafkaChannels) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	opts.Watch = true
+	return c.client.Get().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Watch(ctx)
+}
+
+// Create takes the representation of a kafkaChannel and creates it.  Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *kafkaChannels) Create(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.CreateOptions) (result *v1beta1.KafkaChannel, err error) {
+	result = &v1beta1.KafkaChannel{}
+	err = c.client.Post().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(kafkaChannel).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Update takes the representation of a kafkaChannel and updates it. Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *kafkaChannels) Update(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (result *v1beta1.KafkaChannel, err error) {
+	result = &v1beta1.KafkaChannel{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		Name(kafkaChannel.Name).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(kafkaChannel).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *kafkaChannels) UpdateStatus(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (result *v1beta1.KafkaChannel, err error) {
+	result = &v1beta1.KafkaChannel{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		Name(kafkaChannel.Name).
+		SubResource("status").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(kafkaChannel).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Delete takes name of the kafkaChannel and deletes it. Returns an error if one occurs.
+func (c *kafkaChannels) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		Name(name).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *kafkaChannels) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	var timeout time.Duration
+	if listOpts.TimeoutSeconds != nil {
+		timeout = time.Duration(*listOpts.TimeoutSeconds) * time.Second
+	}
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		VersionedParams(&listOpts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// Patch applies the patch and returns the patched kafkaChannel.
+func (c *kafkaChannels) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaChannel, err error) {
+	result = &v1beta1.KafkaChannel{}
+	err = c.client.Patch(pt).
+		Namespace(c.ns).
+		Resource("kafkachannels").
+		Name(name).
+		SubResource(subresources...).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(data).
+		Do(ctx).
+		Into(result)
+	return
+}
diff --git a/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/messaging_client.go b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/messaging_client.go
new file mode 100644
index 000000000..7f7e3d329
--- /dev/null
+++ b/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1/messaging_client.go
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	rest "k8s.io/client-go/rest"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/scheme"
+)
+
+type MessagingV1beta1Interface interface {
+	RESTClient() rest.Interface
+	KafkaChannelsGetter
+}
+
+// MessagingV1beta1Client is used to interact with features provided by the messaging.knative.dev group.
+type MessagingV1beta1Client struct {
+	restClient rest.Interface
+}
+
+func (c *MessagingV1beta1Client) KafkaChannels(namespace string) KafkaChannelInterface {
+	return newKafkaChannels(c, namespace)
+}
+
+// NewForConfig creates a new MessagingV1beta1Client for the given config.
+func NewForConfig(c *rest.Config) (*MessagingV1beta1Client, error) {
+	config := *c
+	if err := setConfigDefaults(&config); err != nil {
+		return nil, err
+	}
+	client, err := rest.RESTClientFor(&config)
+	if err != nil {
+		return nil, err
+	}
+	return &MessagingV1beta1Client{client}, nil
+}
+
+// NewForConfigOrDie creates a new MessagingV1beta1Client for the given config and
+// panics if there is an error in the config.
+func NewForConfigOrDie(c *rest.Config) *MessagingV1beta1Client {
+	client, err := NewForConfig(c)
+	if err != nil {
+		panic(err)
+	}
+	return client
+}
+
+// New creates a new MessagingV1beta1Client for the given RESTClient.
+func New(c rest.Interface) *MessagingV1beta1Client {
+	return &MessagingV1beta1Client{c}
+}
+
+func setConfigDefaults(config *rest.Config) error {
+	gv := v1beta1.SchemeGroupVersion
+	config.GroupVersion = &gv
+	config.APIPath = "/apis"
+	config.NegotiatedSerializer = scheme.Codecs.WithoutConversion()
+
+	if config.UserAgent == "" {
+		config.UserAgent = rest.DefaultKubernetesUserAgent()
+	}
+
+	return nil
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *MessagingV1beta1Client) RESTClient() rest.Interface {
+	if c == nil {
+		return nil
+	}
+	return c.restClient
+}
diff --git a/control-plane/pkg/client/informers/externalversions/factory.go b/control-plane/pkg/client/informers/externalversions/factory.go
index b1c2ce372..630aaae3b 100644
--- a/control-plane/pkg/client/informers/externalversions/factory.go
+++ b/control-plane/pkg/client/informers/externalversions/factory.go
@@ -30,6 +30,7 @@ import (
 	versioned "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/eventing"
 	internalinterfaces "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/internalinterfaces"
+	messaging "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/messaging"
 )
 
 // SharedInformerOption defines the functional option type for SharedInformerFactory.
@@ -173,8 +174,13 @@ type SharedInformerFactory interface {
 	WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
 
 	Eventing() eventing.Interface
+	Messaging() messaging.Interface
 }
 
 func (f *sharedInformerFactory) Eventing() eventing.Interface {
 	return eventing.New(f, f.namespace, f.tweakListOptions)
 }
+
+func (f *sharedInformerFactory) Messaging() messaging.Interface {
+	return messaging.New(f, f.namespace, f.tweakListOptions)
+}
diff --git a/control-plane/pkg/client/informers/externalversions/generic.go b/control-plane/pkg/client/informers/externalversions/generic.go
index 93f81b44a..09d8e44db 100644
--- a/control-plane/pkg/client/informers/externalversions/generic.go
+++ b/control-plane/pkg/client/informers/externalversions/generic.go
@@ -24,6 +24,7 @@ import (
 	schema "k8s.io/apimachinery/pkg/runtime/schema"
 	cache "k8s.io/client-go/tools/cache"
 	v1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
 )
 
 // GenericInformer is type of SharedIndexInformer which will locate and delegate to other
@@ -56,6 +57,10 @@ func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource
 	case v1alpha1.SchemeGroupVersion.WithResource("kafkasinks"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Eventing().V1alpha1().KafkaSinks().Informer()}, nil
 
+		// Group=messaging.knative.dev, Version=v1beta1
+	case v1beta1.SchemeGroupVersion.WithResource("kafkachannels"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Messaging().V1beta1().KafkaChannels().Informer()}, nil
+
 	}
 
 	return nil, fmt.Errorf("no informer found for %v", resource)
diff --git a/control-plane/pkg/client/informers/externalversions/messaging/interface.go b/control-plane/pkg/client/informers/externalversions/messaging/interface.go
new file mode 100644
index 000000000..3f0be0d11
--- /dev/null
+++ b/control-plane/pkg/client/informers/externalversions/messaging/interface.go
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package messaging
+
+import (
+	internalinterfaces "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/internalinterfaces"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/messaging/v1beta1"
+)
+
+// Interface provides access to each of this group's versions.
+type Interface interface {
+	// V1beta1 provides access to shared informers for resources in V1beta1.
+	V1beta1() v1beta1.Interface
+}
+
+type group struct {
+	factory          internalinterfaces.SharedInformerFactory
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// New returns a new Interface.
+func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
+	return &group{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
+}
+
+// V1beta1 returns a new v1beta1.Interface.
+func (g *group) V1beta1() v1beta1.Interface {
+	return v1beta1.New(g.factory, g.namespace, g.tweakListOptions)
+}
diff --git a/control-plane/pkg/client/informers/externalversions/messaging/v1beta1/interface.go b/control-plane/pkg/client/informers/externalversions/messaging/v1beta1/interface.go
new file mode 100644
index 000000000..6f1679d7e
--- /dev/null
+++ b/control-plane/pkg/client/informers/externalversions/messaging/v1beta1/interface.go
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	internalinterfaces "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/internalinterfaces"
+)
+
+// Interface provides access to all the informers in this group version.
+type Interface interface {
+	// KafkaChannels returns a KafkaChannelInformer.
+	KafkaChannels() KafkaChannelInformer
+}
+
+type version struct {
+	factory          internalinterfaces.SharedInformerFactory
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// New returns a new Interface.
+func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
+	return &version{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
+}
+
+// KafkaChannels returns a KafkaChannelInformer.
+func (v *version) KafkaChannels() KafkaChannelInformer {
+	return &kafkaChannelInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
diff --git a/control-plane/pkg/client/informers/externalversions/messaging/v1beta1/kafkachannel.go b/control-plane/pkg/client/informers/externalversions/messaging/v1beta1/kafkachannel.go
new file mode 100644
index 000000000..17c30b138
--- /dev/null
+++ b/control-plane/pkg/client/informers/externalversions/messaging/v1beta1/kafkachannel.go
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by informer-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	"context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	versioned "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/internalinterfaces"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/messaging/v1beta1"
+)
+
+// KafkaChannelInformer provides access to a shared informer and lister for
+// KafkaChannels.
+type KafkaChannelInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1beta1.KafkaChannelLister
+}
+
+type kafkaChannelInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewKafkaChannelInformer constructs a new informer for KafkaChannel type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewKafkaChannelInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredKafkaChannelInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredKafkaChannelInformer constructs a new informer for KafkaChannel type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredKafkaChannelInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.MessagingV1beta1().KafkaChannels(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.MessagingV1beta1().KafkaChannels(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&messagingv1beta1.KafkaChannel{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *kafkaChannelInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredKafkaChannelInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *kafkaChannelInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&messagingv1beta1.KafkaChannel{}, f.defaultInformer)
+}
+
+func (f *kafkaChannelInformer) Lister() v1beta1.KafkaChannelLister {
+	return v1beta1.NewKafkaChannelLister(f.Informer().GetIndexer())
+}
diff --git a/control-plane/pkg/client/injection/client/client.go b/control-plane/pkg/client/injection/client/client.go
index 43d16be83..d90fefc41 100644
--- a/control-plane/pkg/client/injection/client/client.go
+++ b/control-plane/pkg/client/injection/client/client.go
@@ -34,8 +34,10 @@ import (
 	dynamic "k8s.io/client-go/dynamic"
 	rest "k8s.io/client-go/rest"
 	v1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
 	versioned "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
 	typedeventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/eventing/v1alpha1"
+	typedmessagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/typed/messaging/v1beta1"
 	injection "knative.dev/pkg/injection"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient"
 	logging "knative.dev/pkg/logging"
@@ -241,3 +243,149 @@ func (w *wrapEventingV1alpha1KafkaSinkImpl) UpdateStatus(ctx context.Context, in
 func (w *wrapEventingV1alpha1KafkaSinkImpl) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
 	return nil, errors.New("NYI: Watch")
 }
+
+// MessagingV1beta1 retrieves the MessagingV1beta1Client
+func (w *wrapClient) MessagingV1beta1() typedmessagingv1beta1.MessagingV1beta1Interface {
+	return &wrapMessagingV1beta1{
+		dyn: w.dyn,
+	}
+}
+
+type wrapMessagingV1beta1 struct {
+	dyn dynamic.Interface
+}
+
+func (w *wrapMessagingV1beta1) RESTClient() rest.Interface {
+	panic("RESTClient called on dynamic client!")
+}
+
+func (w *wrapMessagingV1beta1) KafkaChannels(namespace string) typedmessagingv1beta1.KafkaChannelInterface {
+	return &wrapMessagingV1beta1KafkaChannelImpl{
+		dyn: w.dyn.Resource(schema.GroupVersionResource{
+			Group:    "messaging.knative.dev",
+			Version:  "v1beta1",
+			Resource: "kafkachannels",
+		}),
+
+		namespace: namespace,
+	}
+}
+
+type wrapMessagingV1beta1KafkaChannelImpl struct {
+	dyn dynamic.NamespaceableResourceInterface
+
+	namespace string
+}
+
+var _ typedmessagingv1beta1.KafkaChannelInterface = (*wrapMessagingV1beta1KafkaChannelImpl)(nil)
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) Create(ctx context.Context, in *v1beta1.KafkaChannel, opts v1.CreateOptions) (*v1beta1.KafkaChannel, error) {
+	in.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   "messaging.knative.dev",
+		Version: "v1beta1",
+		Kind:    "KafkaChannel",
+	})
+	uo := &unstructured.Unstructured{}
+	if err := convert(in, uo); err != nil {
+		return nil, err
+	}
+	uo, err := w.dyn.Namespace(w.namespace).Create(ctx, uo, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := &v1beta1.KafkaChannel{}
+	if err := convert(uo, out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	return w.dyn.Namespace(w.namespace).Delete(ctx, name, opts)
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	return w.dyn.Namespace(w.namespace).DeleteCollection(ctx, opts, listOpts)
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) Get(ctx context.Context, name string, opts v1.GetOptions) (*v1beta1.KafkaChannel, error) {
+	uo, err := w.dyn.Namespace(w.namespace).Get(ctx, name, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := &v1beta1.KafkaChannel{}
+	if err := convert(uo, out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) List(ctx context.Context, opts v1.ListOptions) (*v1beta1.KafkaChannelList, error) {
+	uo, err := w.dyn.Namespace(w.namespace).List(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := &v1beta1.KafkaChannelList{}
+	if err := convert(uo, out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaChannel, err error) {
+	uo, err := w.dyn.Namespace(w.namespace).Patch(ctx, name, pt, data, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := &v1beta1.KafkaChannel{}
+	if err := convert(uo, out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) Update(ctx context.Context, in *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error) {
+	in.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   "messaging.knative.dev",
+		Version: "v1beta1",
+		Kind:    "KafkaChannel",
+	})
+	uo := &unstructured.Unstructured{}
+	if err := convert(in, uo); err != nil {
+		return nil, err
+	}
+	uo, err := w.dyn.Namespace(w.namespace).Update(ctx, uo, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := &v1beta1.KafkaChannel{}
+	if err := convert(uo, out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) UpdateStatus(ctx context.Context, in *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error) {
+	in.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   "messaging.knative.dev",
+		Version: "v1beta1",
+		Kind:    "KafkaChannel",
+	})
+	uo := &unstructured.Unstructured{}
+	if err := convert(in, uo); err != nil {
+		return nil, err
+	}
+	uo, err := w.dyn.Namespace(w.namespace).UpdateStatus(ctx, uo, opts)
+	if err != nil {
+		return nil, err
+	}
+	out := &v1beta1.KafkaChannel{}
+	if err := convert(uo, out); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (w *wrapMessagingV1beta1KafkaChannelImpl) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return nil, errors.New("NYI: Watch")
+}
diff --git a/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake/fake.go b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake/fake.go
new file mode 100644
index 000000000..53f6d02df
--- /dev/null
+++ b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/factory/fake"
+	kafkachannel "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = kafkachannel.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Messaging().V1beta1().KafkaChannels()
+	return context.WithValue(ctx, kafkachannel.Key{}, inf), inf.Informer()
+}
diff --git a/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/fake/fake.go b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/fake/fake.go
new file mode 100644
index 000000000..ca6dffbc9
--- /dev/null
+++ b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/fake/fake.go
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	factoryfiltered "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/factory/filtered"
+	filtered "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Messaging().V1beta1().KafkaChannels()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/kafkachannel.go b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/kafkachannel.go
new file mode 100644
index 000000000..f77095e27
--- /dev/null
+++ b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/filtered/kafkachannel.go
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package filtered
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apismessagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	versioned "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/messaging/v1beta1"
+	client "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
+	filtered "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/factory/filtered"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/messaging/v1beta1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Messaging().V1beta1().KafkaChannels()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	for _, selector := range labelSelectors {
+		inf := &wrapper{client: client.Get(ctx), selector: selector}
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+	}
+	return ctx
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1beta1.KafkaChannelInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/messaging/v1beta1.KafkaChannelInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1beta1.KafkaChannelInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+
+	selector string
+}
+
+var _ v1beta1.KafkaChannelInformer = (*wrapper)(nil)
+var _ messagingv1beta1.KafkaChannelLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apismessagingv1beta1.KafkaChannel{}, 0, nil)
+}
+
+func (w *wrapper) Lister() messagingv1beta1.KafkaChannelLister {
+	return w
+}
+
+func (w *wrapper) KafkaChannels(namespace string) messagingv1beta1.KafkaChannelNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, selector: w.selector}
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1beta1.KafkaChannel, err error) {
+	reqs, err := labels.ParseToRequirements(w.selector)
+	if err != nil {
+		return nil, err
+	}
+	selector = selector.Add(reqs...)
+	lo, err := w.client.MessagingV1beta1().KafkaChannels(w.namespace).List(context.TODO(), v1.ListOptions{
+		LabelSelector: selector.String(),
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apismessagingv1beta1.KafkaChannel, error) {
+	// TODO(mattmoor): Check that the fetched object matches the selector.
+	return w.client.MessagingV1beta1().KafkaChannels(w.namespace).Get(context.TODO(), name, v1.GetOptions{
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+}
diff --git a/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/kafkachannel.go b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/kafkachannel.go
new file mode 100644
index 000000000..e6a86da7c
--- /dev/null
+++ b/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/kafkachannel.go
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkachannel
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apismessagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	versioned "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/messaging/v1beta1"
+	client "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
+	factory "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/factory"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/messaging/v1beta1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Messaging().V1beta1().KafkaChannels()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1beta1.KafkaChannelInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-kafka-broker/control-plane/pkg/client/informers/externalversions/messaging/v1beta1.KafkaChannelInformer from context.")
+	}
+	return untyped.(v1beta1.KafkaChannelInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+}
+
+var _ v1beta1.KafkaChannelInformer = (*wrapper)(nil)
+var _ messagingv1beta1.KafkaChannelLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apismessagingv1beta1.KafkaChannel{}, 0, nil)
+}
+
+func (w *wrapper) Lister() messagingv1beta1.KafkaChannelLister {
+	return w
+}
+
+func (w *wrapper) KafkaChannels(namespace string) messagingv1beta1.KafkaChannelNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace}
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1beta1.KafkaChannel, err error) {
+	lo, err := w.client.MessagingV1beta1().KafkaChannels(w.namespace).List(context.TODO(), v1.ListOptions{
+		LabelSelector: selector.String(),
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apismessagingv1beta1.KafkaChannel, error) {
+	return w.client.MessagingV1beta1().KafkaChannels(w.namespace).Get(context.TODO(), name, v1.GetOptions{
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+}
diff --git a/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go b/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
new file mode 100644
index 000000000..bca4fab16
--- /dev/null
+++ b/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/controller.go
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkachannel
+
+import (
+	context "context"
+	fmt "fmt"
+	reflect "reflect"
+	strings "strings"
+
+	zap "go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	scheme "k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	record "k8s.io/client-go/tools/record"
+	versionedscheme "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned/scheme"
+	client "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
+	kafkachannel "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	controller "knative.dev/pkg/controller"
+	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+const (
+	defaultControllerAgentName = "kafkachannel-controller"
+	defaultFinalizerName       = "kafkachannels.messaging.knative.dev"
+)
+
+// NewImpl returns a controller.Impl that handles queuing and feeding work from
+// the queue through an implementation of controller.Reconciler, delegating to
+// the provided Interface and optional Finalizer methods. OptionsFn is used to return
+// controller.ControllerOptions to be used by the internal reconciler.
+func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsFn) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	// Check the options function input. It should be 0 or 1.
+	if len(optionsFns) > 1 {
+		logger.Fatal("Up to one options function is supported, found: ", len(optionsFns))
+	}
+
+	kafkachannelInformer := kafkachannel.Get(ctx)
+
+	lister := kafkachannelInformer.Lister()
+
+	var promoteFilterFunc func(obj interface{}) bool
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					if promoteFilterFunc != nil {
+						if ok := promoteFilterFunc(elt); !ok {
+							continue
+						}
+					}
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client.Get(ctx),
+		Lister:        lister,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
+
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "messaging.knative.dev.KafkaChannel"),
+	)
+
+	impl := controller.NewContext(ctx, rec, controller.ControllerOptions{WorkQueueName: ctrTypeName, Logger: logger})
+	agentName := defaultControllerAgentName
+
+	// Pass impl to the options. Save any optional results.
+	for _, fn := range optionsFns {
+		opts := fn(impl)
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.AgentName != "" {
+			agentName = opts.AgentName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+		if opts.PromoteFilterFunc != nil {
+			promoteFilterFunc = opts.PromoteFilterFunc
+		}
+	}
+
+	rec.Recorder = createRecorder(ctx, agentName)
+
+	return impl
+}
+
+func createRecorder(ctx context.Context, agentName string) record.EventRecorder {
+	logger := logging.FromContext(ctx)
+
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcaster()
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&v1.EventSinkImpl{Interface: kubeclient.Get(ctx).CoreV1().Events("")}),
+		}
+		recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: agentName})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+		}()
+	}
+
+	return recorder
+}
+
+func init() {
+	versionedscheme.AddToScheme(scheme.Scheme)
+}
diff --git a/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go b/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go
new file mode 100644
index 000000000..846a47fd0
--- /dev/null
+++ b/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/reconciler.go
@@ -0,0 +1,464 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkachannel
+
+import (
+	context "context"
+	json "encoding/json"
+	fmt "fmt"
+
+	zap "go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
+	equality "k8s.io/apimachinery/pkg/api/equality"
+	errors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	sets "k8s.io/apimachinery/pkg/util/sets"
+	record "k8s.io/client-go/tools/record"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	versioned "knative.dev/eventing-kafka-broker/control-plane/pkg/client/clientset/versioned"
+	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/listers/messaging/v1beta1"
+	controller "knative.dev/pkg/controller"
+	kmp "knative.dev/pkg/kmp"
+	logging "knative.dev/pkg/logging"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// Interface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1beta1.KafkaChannel.
+type Interface interface {
+	// ReconcileKind implements custom logic to reconcile v1beta1.KafkaChannel. Any changes
+	// to the objects .Status or .Finalizers will be propagated to the stored
+	// object. It is recommended that implementors do not call any update calls
+	// for the Kind inside of ReconcileKind, it is the responsibility of the calling
+	// controller to propagate those properties. The resource passed to ReconcileKind
+	// will always have an empty deletion timestamp.
+	ReconcileKind(ctx context.Context, o *v1beta1.KafkaChannel) reconciler.Event
+}
+
+// Finalizer defines the strongly typed interfaces to be implemented by a
+// controller finalizing v1beta1.KafkaChannel.
+type Finalizer interface {
+	// FinalizeKind implements custom logic to finalize v1beta1.KafkaChannel. Any changes
+	// to the objects .Status or .Finalizers will be ignored. Returning a nil or
+	// Normal type reconciler.Event will allow the finalizer to be deleted on
+	// the resource. The resource passed to FinalizeKind will always have a set
+	// deletion timestamp.
+	FinalizeKind(ctx context.Context, o *v1beta1.KafkaChannel) reconciler.Event
+}
+
+// ReadOnlyInterface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1beta1.KafkaChannel if they want to process resources for which
+// they are not the leader.
+type ReadOnlyInterface interface {
+	// ObserveKind implements logic to observe v1beta1.KafkaChannel.
+	// This method should not write to the API.
+	ObserveKind(ctx context.Context, o *v1beta1.KafkaChannel) reconciler.Event
+}
+
+// ReadOnlyFinalizer defines the strongly typed interfaces to be implemented by a
+// controller finalizing v1beta1.KafkaChannel if they want to process tombstoned resources
+// even when they are not the leader.  Due to the nature of how finalizers are handled
+// there are no guarantees that this will be called.
+//
+// Deprecated: Use reconciler.OnDeletionInterface instead.
+type ReadOnlyFinalizer interface {
+	// ObserveFinalizeKind implements custom logic to observe the final state of v1beta1.KafkaChannel.
+	// This method should not write to the API.
+	//
+	// Deprecated: Use reconciler.ObserveDeletion instead.
+	ObserveFinalizeKind(ctx context.Context, o *v1beta1.KafkaChannel) reconciler.Event
+}
+
+type doReconcile func(ctx context.Context, o *v1beta1.KafkaChannel) reconciler.Event
+
+// reconcilerImpl implements controller.Reconciler for v1beta1.KafkaChannel resources.
+type reconcilerImpl struct {
+	// LeaderAwareFuncs is inlined to help us implement reconciler.LeaderAware.
+	reconciler.LeaderAwareFuncs
+
+	// Client is used to write back status updates.
+	Client versioned.Interface
+
+	// Listers index properties about resources.
+	Lister messagingv1beta1.KafkaChannelLister
+
+	// Recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	Recorder record.EventRecorder
+
+	// configStore allows for decorating a context with config maps.
+	// +optional
+	configStore reconciler.ConfigStore
+
+	// reconciler is the implementation of the business logic of the resource.
+	reconciler Interface
+
+	// finalizerName is the name of the finalizer to reconcile.
+	finalizerName string
+
+	// skipStatusUpdates configures whether or not this reconciler automatically updates
+	// the status of the reconciled resource.
+	skipStatusUpdates bool
+}
+
+// Check that our Reconciler implements controller.Reconciler.
+var _ controller.Reconciler = (*reconcilerImpl)(nil)
+
+// Check that our generated Reconciler is always LeaderAware.
+var _ reconciler.LeaderAware = (*reconcilerImpl)(nil)
+
+func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versioned.Interface, lister messagingv1beta1.KafkaChannelLister, recorder record.EventRecorder, r Interface, options ...controller.Options) controller.Reconciler {
+	// Check the options function input. It should be 0 or 1.
+	if len(options) > 1 {
+		logger.Fatal("Up to one options struct is supported, found: ", len(options))
+	}
+
+	// Fail fast when users inadvertently implement the other LeaderAware interface.
+	// For the typed reconcilers, Promote shouldn't take any arguments.
+	if _, ok := r.(reconciler.LeaderAware); ok {
+		logger.Fatalf("%T implements the incorrect LeaderAware interface. Promote() should not take an argument as genreconciler handles the enqueuing automatically.", r)
+	}
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					// TODO: Consider letting users specify a filter in options.
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client,
+		Lister:        lister,
+		Recorder:      recorder,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	for _, opts := range options {
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+	}
+
+	return rec
+}
+
+// Reconcile implements controller.Reconciler
+func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
+	logger := logging.FromContext(ctx)
+
+	// Initialize the reconciler state. This will convert the namespace/name
+	// string into a distinct namespace and name, determine if this instance of
+	// the reconciler is the leader, and any additional interfaces implemented
+	// by the reconciler. Returns an error is the resource key is invalid.
+	s, err := newState(key, r)
+	if err != nil {
+		logger.Error("Invalid resource key: ", key)
+		return nil
+	}
+
+	// If we are not the leader, and we don't implement either ReadOnly
+	// observer interfaces, then take a fast-path out.
+	if s.isNotLeaderNorObserver() {
+		return controller.NewSkipKey(key)
+	}
+
+	// If configStore is set, attach the frozen configuration to the context.
+	if r.configStore != nil {
+		ctx = r.configStore.ToContext(ctx)
+	}
+
+	// Add the recorder to context.
+	ctx = controller.WithEventRecorder(ctx, r.Recorder)
+
+	// Get the resource with this namespace/name.
+
+	getter := r.Lister.KafkaChannels(s.namespace)
+
+	original, err := getter.Get(s.name)
+
+	if errors.IsNotFound(err) {
+		// The resource may no longer exist, in which case we stop processing and call
+		// the ObserveDeletion handler if appropriate.
+		logger.Debugf("Resource %q no longer exists", key)
+		if del, ok := r.reconciler.(reconciler.OnDeletionInterface); ok {
+			return del.ObserveDeletion(ctx, types.NamespacedName{
+				Namespace: s.namespace,
+				Name:      s.name,
+			})
+		}
+		return nil
+	} else if err != nil {
+		return err
+	}
+
+	// Don't modify the informers copy.
+	resource := original.DeepCopy()
+
+	var reconcileEvent reconciler.Event
+
+	name, do := s.reconcileMethodFor(resource)
+	// Append the target method to the logger.
+	logger = logger.With(zap.String("targetMethod", name))
+	switch name {
+	case reconciler.DoReconcileKind:
+		// Set and update the finalizer on resource if r.reconciler
+		// implements Finalizer.
+		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
+			return fmt.Errorf("failed to set finalizers: %w", err)
+		}
+
+		if !r.skipStatusUpdates {
+			reconciler.PreProcessReconcile(ctx, resource)
+		}
+
+		// Reconcile this copy of the resource and then write back any status
+		// updates regardless of whether the reconciliation errored out.
+		reconcileEvent = do(ctx, resource)
+
+		if !r.skipStatusUpdates {
+			reconciler.PostProcessReconcile(ctx, resource, original)
+		}
+
+	case reconciler.DoFinalizeKind:
+		// For finalizing reconcilers, if this resource being marked for deletion
+		// and reconciled cleanly (nil or normal event), remove the finalizer.
+		reconcileEvent = do(ctx, resource)
+
+		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
+			return fmt.Errorf("failed to clear finalizers: %w", err)
+		}
+
+	case reconciler.DoObserveKind, reconciler.DoObserveFinalizeKind:
+		// Observe any changes to this resource, since we are not the leader.
+		reconcileEvent = do(ctx, resource)
+
+	}
+
+	// Synchronize the status.
+	switch {
+	case r.skipStatusUpdates:
+		// This reconciler implementation is configured to skip resource updates.
+		// This may mean this reconciler does not observe spec, but reconciles external changes.
+	case equality.Semantic.DeepEqual(original.Status, resource.Status):
+		// If we didn't change anything then don't call updateStatus.
+		// This is important because the copy we loaded from the injectionInformer's
+		// cache may be stale and we don't want to overwrite a prior update
+		// to status with this stale state.
+	case !s.isLeader:
+		// High-availability reconcilers may have many replicas watching the resource, but only
+		// the elected leader is expected to write modifications.
+		logger.Warn("Saw status changes when we aren't the leader!")
+	default:
+		if err = r.updateStatus(ctx, original, resource); err != nil {
+			logger.Warnw("Failed to update resource status", zap.Error(err))
+			r.Recorder.Eventf(resource, v1.EventTypeWarning, "UpdateFailed",
+				"Failed to update status for %q: %v", resource.Name, err)
+			return err
+		}
+	}
+
+	// Report the reconciler event, if any.
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			logger.Infow("Returned an event", zap.Any("event", reconcileEvent))
+			r.Recorder.Event(resource, event.EventType, event.Reason, event.Error())
+
+			// the event was wrapped inside an error, consider the reconciliation as failed
+			if _, isEvent := reconcileEvent.(*reconciler.ReconcilerEvent); !isEvent {
+				return reconcileEvent
+			}
+			return nil
+		}
+
+		if controller.IsSkipKey(reconcileEvent) {
+			// This is a wrapped error, don't emit an event.
+		} else if ok, _ := controller.IsRequeueKey(reconcileEvent); ok {
+			// This is a wrapped error, don't emit an event.
+		} else {
+			logger.Errorw("Returned an error", zap.Error(reconcileEvent))
+			r.Recorder.Event(resource, v1.EventTypeWarning, "InternalError", reconcileEvent.Error())
+		}
+		return reconcileEvent
+	}
+
+	return nil
+}
+
+func (r *reconcilerImpl) updateStatus(ctx context.Context, existing *v1beta1.KafkaChannel, desired *v1beta1.KafkaChannel) error {
+	existing = existing.DeepCopy()
+	return reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
+		// The first iteration tries to use the injectionInformer's state, subsequent attempts fetch the latest state via API.
+		if attempts > 0 {
+
+			getter := r.Client.MessagingV1beta1().KafkaChannels(desired.Namespace)
+
+			existing, err = getter.Get(ctx, desired.Name, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+		}
+
+		// If there's nothing to update, just return.
+		if equality.Semantic.DeepEqual(existing.Status, desired.Status) {
+			return nil
+		}
+
+		if diff, err := kmp.SafeDiff(existing.Status, desired.Status); err == nil && diff != "" {
+			logging.FromContext(ctx).Debug("Updating status with: ", diff)
+		}
+
+		existing.Status = desired.Status
+
+		updater := r.Client.MessagingV1beta1().KafkaChannels(existing.Namespace)
+
+		_, err = updater.UpdateStatus(ctx, existing, metav1.UpdateOptions{})
+		return err
+	})
+}
+
+// updateFinalizersFiltered will update the Finalizers of the resource.
+// TODO: this method could be generic and sync all finalizers. For now it only
+// updates defaultFinalizerName or its override.
+func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource *v1beta1.KafkaChannel) (*v1beta1.KafkaChannel, error) {
+
+	getter := r.Lister.KafkaChannels(resource.Namespace)
+
+	actual, err := getter.Get(resource.Name)
+	if err != nil {
+		return resource, err
+	}
+
+	// Don't modify the informers copy.
+	existing := actual.DeepCopy()
+
+	var finalizers []string
+
+	// If there's nothing to update, just return.
+	existingFinalizers := sets.NewString(existing.Finalizers...)
+	desiredFinalizers := sets.NewString(resource.Finalizers...)
+
+	if desiredFinalizers.Has(r.finalizerName) {
+		if existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Add the finalizer.
+		finalizers = append(existing.Finalizers, r.finalizerName)
+	} else {
+		if !existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Remove the finalizer.
+		existingFinalizers.Delete(r.finalizerName)
+		finalizers = existingFinalizers.List()
+	}
+
+	mergePatch := map[string]interface{}{
+		"metadata": map[string]interface{}{
+			"finalizers":      finalizers,
+			"resourceVersion": existing.ResourceVersion,
+		},
+	}
+
+	patch, err := json.Marshal(mergePatch)
+	if err != nil {
+		return resource, err
+	}
+
+	patcher := r.Client.MessagingV1beta1().KafkaChannels(resource.Namespace)
+
+	resourceName := resource.Name
+	updated, err := patcher.Patch(ctx, resourceName, types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		r.Recorder.Eventf(existing, v1.EventTypeWarning, "FinalizerUpdateFailed",
+			"Failed to update finalizers for %q: %v", resourceName, err)
+	} else {
+		r.Recorder.Eventf(updated, v1.EventTypeNormal, "FinalizerUpdate",
+			"Updated %q finalizers", resource.GetName())
+	}
+	return updated, err
+}
+
+func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *v1beta1.KafkaChannel) (*v1beta1.KafkaChannel, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	// If this resource is not being deleted, mark the finalizer.
+	if resource.GetDeletionTimestamp().IsZero() {
+		finalizers.Insert(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
+
+func (r *reconcilerImpl) clearFinalizer(ctx context.Context, resource *v1beta1.KafkaChannel, reconcileEvent reconciler.Event) (*v1beta1.KafkaChannel, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+	if resource.GetDeletionTimestamp().IsZero() {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			if event.EventType == v1.EventTypeNormal {
+				finalizers.Delete(r.finalizerName)
+			}
+		}
+	} else {
+		finalizers.Delete(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
diff --git a/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/state.go b/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/state.go
new file mode 100644
index 000000000..1b65d8cc1
--- /dev/null
+++ b/control-plane/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel/state.go
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkachannel
+
+import (
+	fmt "fmt"
+
+	types "k8s.io/apimachinery/pkg/types"
+	cache "k8s.io/client-go/tools/cache"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// state is used to track the state of a reconciler in a single run.
+type state struct {
+	// key is the original reconciliation key from the queue.
+	key string
+	// namespace is the namespace split from the reconciliation key.
+	namespace string
+	// name is the name split from the reconciliation key.
+	name string
+	// reconciler is the reconciler.
+	reconciler Interface
+	// roi is the read only interface cast of the reconciler.
+	roi ReadOnlyInterface
+	// isROI (Read Only Interface) the reconciler only observes reconciliation.
+	isROI bool
+	// rof is the read only finalizer cast of the reconciler.
+	rof ReadOnlyFinalizer
+	// isROF (Read Only Finalizer) the reconciler only observes finalize.
+	isROF bool
+	// isLeader the instance of the reconciler is the elected leader.
+	isLeader bool
+}
+
+func newState(key string, r *reconcilerImpl) (*state, error) {
+	// Convert the namespace/name string into a distinct namespace and name.
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return nil, fmt.Errorf("invalid resource key: %s", key)
+	}
+
+	roi, isROI := r.reconciler.(ReadOnlyInterface)
+	rof, isROF := r.reconciler.(ReadOnlyFinalizer)
+
+	isLeader := r.IsLeaderFor(types.NamespacedName{
+		Namespace: namespace,
+		Name:      name,
+	})
+
+	return &state{
+		key:        key,
+		namespace:  namespace,
+		name:       name,
+		reconciler: r.reconciler,
+		roi:        roi,
+		isROI:      isROI,
+		rof:        rof,
+		isROF:      isROF,
+		isLeader:   isLeader,
+	}, nil
+}
+
+// isNotLeaderNorObserver checks to see if this reconciler with the current
+// state is enabled to do any work or not.
+// isNotLeaderNorObserver returns true when there is no work possible for the
+// reconciler.
+func (s *state) isNotLeaderNorObserver() bool {
+	if !s.isLeader && !s.isROI && !s.isROF {
+		// If we are not the leader, and we don't implement either ReadOnly
+		// interface, then take a fast-path out.
+		return true
+	}
+	return false
+}
+
+func (s *state) reconcileMethodFor(o *v1beta1.KafkaChannel) (string, doReconcile) {
+	if o.GetDeletionTimestamp().IsZero() {
+		if s.isLeader {
+			return reconciler.DoReconcileKind, s.reconciler.ReconcileKind
+		} else if s.isROI {
+			return reconciler.DoObserveKind, s.roi.ObserveKind
+		}
+	} else if fin, ok := s.reconciler.(Finalizer); s.isLeader && ok {
+		return reconciler.DoFinalizeKind, fin.FinalizeKind
+	} else if !s.isLeader && s.isROF {
+		return reconciler.DoObserveFinalizeKind, s.rof.ObserveFinalizeKind
+	}
+	return "unknown", nil
+}
diff --git a/control-plane/pkg/client/listers/messaging/v1beta1/expansion_generated.go b/control-plane/pkg/client/listers/messaging/v1beta1/expansion_generated.go
new file mode 100644
index 000000000..730951458
--- /dev/null
+++ b/control-plane/pkg/client/listers/messaging/v1beta1/expansion_generated.go
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1beta1
+
+// KafkaChannelListerExpansion allows custom methods to be added to
+// KafkaChannelLister.
+type KafkaChannelListerExpansion interface{}
+
+// KafkaChannelNamespaceListerExpansion allows custom methods to be added to
+// KafkaChannelNamespaceLister.
+type KafkaChannelNamespaceListerExpansion interface{}
diff --git a/control-plane/pkg/client/listers/messaging/v1beta1/kafkachannel.go b/control-plane/pkg/client/listers/messaging/v1beta1/kafkachannel.go
new file mode 100644
index 000000000..0bf135c64
--- /dev/null
+++ b/control-plane/pkg/client/listers/messaging/v1beta1/kafkachannel.go
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Code generated by lister-gen. DO NOT EDIT.
+
+package v1beta1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
+)
+
+// KafkaChannelLister helps list KafkaChannels.
+// All objects returned here must be treated as read-only.
+type KafkaChannelLister interface {
+	// List lists all KafkaChannels in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1beta1.KafkaChannel, err error)
+	// KafkaChannels returns an object that can list and get KafkaChannels.
+	KafkaChannels(namespace string) KafkaChannelNamespaceLister
+	KafkaChannelListerExpansion
+}
+
+// kafkaChannelLister implements the KafkaChannelLister interface.
+type kafkaChannelLister struct {
+	indexer cache.Indexer
+}
+
+// NewKafkaChannelLister returns a new KafkaChannelLister.
+func NewKafkaChannelLister(indexer cache.Indexer) KafkaChannelLister {
+	return &kafkaChannelLister{indexer: indexer}
+}
+
+// List lists all KafkaChannels in the indexer.
+func (s *kafkaChannelLister) List(selector labels.Selector) (ret []*v1beta1.KafkaChannel, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1beta1.KafkaChannel))
+	})
+	return ret, err
+}
+
+// KafkaChannels returns an object that can list and get KafkaChannels.
+func (s *kafkaChannelLister) KafkaChannels(namespace string) KafkaChannelNamespaceLister {
+	return kafkaChannelNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// KafkaChannelNamespaceLister helps list and get KafkaChannels.
+// All objects returned here must be treated as read-only.
+type KafkaChannelNamespaceLister interface {
+	// List lists all KafkaChannels in the indexer for a given namespace.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1beta1.KafkaChannel, err error)
+	// Get retrieves the KafkaChannel from the indexer for a given namespace and name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*v1beta1.KafkaChannel, error)
+	KafkaChannelNamespaceListerExpansion
+}
+
+// kafkaChannelNamespaceLister implements the KafkaChannelNamespaceLister
+// interface.
+type kafkaChannelNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all KafkaChannels in the indexer for a given namespace.
+func (s kafkaChannelNamespaceLister) List(selector labels.Selector) (ret []*v1beta1.KafkaChannel, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1beta1.KafkaChannel))
+	})
+	return ret, err
+}
+
+// Get retrieves the KafkaChannel from the indexer for a given namespace and name.
+func (s kafkaChannelNamespaceLister) Get(name string) (*v1beta1.KafkaChannel, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1beta1.Resource("kafkachannel"), name)
+	}
+	return obj.(*v1beta1.KafkaChannel), nil
+}
diff --git a/control-plane/pkg/receiver/path.go b/control-plane/pkg/receiver/path.go
index cf0caa830..3139b1c26 100644
--- a/control-plane/pkg/receiver/path.go
+++ b/control-plane/pkg/receiver/path.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package receiver
 
 import (
diff --git a/hack/tools.go b/hack/tools.go
index b7371c57d..5166fbdfe 100644
--- a/hack/tools.go
+++ b/hack/tools.go
@@ -1,20 +1,20 @@
 // +build tools
 
 /*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package tools
 
diff --git a/hack/update-codegen.sh b/hack/update-codegen.sh
index 43dd78b09..c84296a87 100755
--- a/hack/update-codegen.sh
+++ b/hack/update-codegen.sh
@@ -38,7 +38,7 @@ group "Kubernetes Codegen"
 #                  instead of the $GOPATH directly. For normal projects this can be dropped.
 ${CODEGEN_PKG}/generate-groups.sh "deepcopy,client,informer,lister" \
   knative.dev/eventing-kafka-broker/control-plane/pkg/client knative.dev/eventing-kafka-broker/control-plane/pkg/apis \
-  "eventing:v1alpha1" \
+  "eventing:v1alpha1 messaging:v1beta1" \
   --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate/boilerplate.go.txt
 
 group "Knative Codegen"
@@ -46,7 +46,7 @@ group "Knative Codegen"
 # Knative Injection
 ${KNATIVE_CODEGEN_PKG}/hack/generate-knative.sh "injection" \
   knative.dev/eventing-kafka-broker/control-plane/pkg/client knative.dev/eventing-kafka-broker/control-plane/pkg/apis \
-  "eventing:v1alpha1" \
+  "eventing:v1alpha1 messaging:v1beta1" \
   --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate/boilerplate.go.txt
 
 group "Update deps post-codegen"


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
index 4cdd7f5a3..2a331a678 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package v1alpha1
 
 import (
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
index 377b8177b..22784c098 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package v1alpha1
 
 import (
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion_test.go
new file mode 100644
index 000000000..6fd4b8fda
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion_test.go
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"testing"
+)
+
+func TestKafkaChannelConversionBadType(t *testing.T) {
+	good, bad := &KafkaChannel{}, &KafkaChannel{}
+
+	if err := good.ConvertTo(context.Background(), bad); err == nil {
+		t.Errorf("ConvertTo() = %#v, wanted error", bad)
+	}
+
+	if err := good.ConvertFrom(context.Background(), bad); err == nil {
+		t.Errorf("ConvertFrom() = %#v, wanted error", good)
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
new file mode 100644
index 000000000..05c872b1b
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	testNumPartitions     = 10
+	testReplicationFactor = 5
+)
+
+func TestKafkaChannelDefaults(t *testing.T) {
+	testCases := map[string]struct {
+		initial  KafkaChannel
+		expected KafkaChannel
+	}{
+		"nil spec": {
+			initial: KafkaChannel{},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     DefaultNumPartitions,
+					ReplicationFactor: DefaultReplicationFactor,
+				},
+			},
+		},
+		"numPartitions not set": {
+			initial: KafkaChannel{
+				Spec: KafkaChannelSpec{
+					ReplicationFactor: testReplicationFactor,
+				},
+			},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     DefaultNumPartitions,
+					ReplicationFactor: testReplicationFactor,
+				},
+			},
+		},
+		"replicationFactor not set": {
+			initial: KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions: testNumPartitions,
+				},
+			},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     testNumPartitions,
+					ReplicationFactor: DefaultReplicationFactor,
+				},
+			},
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			tc.initial.SetDefaults(context.TODO())
+			if diff := cmp.Diff(tc.expected, tc.initial); diff != "" {
+				t.Fatalf("Unexpected defaults (-want, +got): %s", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go
new file mode 100644
index 000000000..57b1dca22
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"knative.dev/pkg/apis"
+)
+
+func TestKafkaChannelStatusSetAddress(t *testing.T) {
+	status := KafkaChannelStatus{}
+	status.SetAddress(apis.HTTP("localhost"))
+
+	addressable := status.GetCondition(ConditionAddressable).IsTrue()
+
+	assert.True(t, addressable)
+
+	status.SetAddress(nil)
+
+	notAddressable := status.GetCondition(ConditionAddressable).IsFalse()
+
+	assert.True(t, notAddressable)
+}
+
+func TestKafkaChannelGetConditionSet(t *testing.T) {
+
+	cs := (&KafkaChannel{}).GetConditionSet()
+
+	assert.Equal(t, conditionSet, cs)
+}
+
+func TestKafkaChannelStatusGetConditionSet(t *testing.T) {
+
+	cs := (&KafkaChannelStatus{}).GetConditionSet()
+
+	assert.Equal(t, conditionSet, cs)
+}
+
+func TestRegisterAlternateKafkaChannelConditionSet(t *testing.T) {
+
+	cs := apis.NewLivingConditionSet(apis.ConditionReady, "hello")
+
+	RegisterConditionSet(cs)
+
+	kc := KafkaChannel{}
+
+	assert.Equal(t, cs, kc.GetConditionSet())
+	assert.Equal(t, cs, kc.Status.GetConditionSet())
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
new file mode 100644
index 000000000..934a30c20
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+func TestKafkaChannelGetGroupVersionKind(t *testing.T) {
+	ch := KafkaChannel{}
+	gvk := ch.GetGroupVersionKind()
+
+	if gvk.Kind != "KafkaChannel" {
+		t.Errorf("Should be 'KafkaChannel'.")
+	}
+}
+
+func TestKafkaChannelGetStatus(t *testing.T) {
+	status := &duckv1.Status{}
+	config := KafkaChannel{
+		Status: KafkaChannelStatus{
+			ChannelableStatus: eventingduck.ChannelableStatus{
+				Status: *status,
+			},
+		},
+	}
+
+	if !cmp.Equal(config.GetStatus(), status) {
+		t.Errorf("GetStatus did not retrieve status. Got=%v Want=%v", config.GetStatus(), status)
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
new file mode 100644
index 000000000..f8adaef02
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/webhook/resourcesemantics"
+
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/apis/eventing"
+	"knative.dev/pkg/apis"
+)
+
+func TestKafkaChannelValidation(t *testing.T) {
+
+	testCases := map[string]struct {
+		cr   resourcesemantics.GenericCRD
+		want *apis.FieldError
+	}{
+		"empty spec": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrInvalidValue(0, "spec.numPartitions")
+				errs = errs.Also(fe)
+				fe = apis.ErrInvalidValue(0, "spec.replicationFactor")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		"negative numPartitions": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     -10,
+					ReplicationFactor: 1,
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue(-10, "spec.numPartitions")
+				return fe
+			}(),
+		},
+		"negative replicationFactor": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: -10,
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue(-10, "spec.replicationFactor")
+				return fe
+			}(),
+		},
+		"valid subscribers array": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					ChannelableSpec: eventingduck.ChannelableSpec{
+						SubscribableSpec: eventingduck.SubscribableSpec{
+							Subscribers: []eventingduck.SubscriberSpec{{
+								SubscriberURI: apis.HTTP("subscriberendpoint"),
+								ReplyURI:      apis.HTTP("resultendpoint"),
+							}},
+						}},
+				},
+			},
+			want: nil,
+		},
+		"empty subscriber at index 1": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					ChannelableSpec: eventingduck.ChannelableSpec{
+						SubscribableSpec: eventingduck.SubscribableSpec{
+							Subscribers: []eventingduck.SubscriberSpec{{
+								SubscriberURI: apis.HTTP("subscriberendpoint"),
+								ReplyURI:      apis.HTTP("replyendpoint"),
+							}, {}},
+						}},
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrMissingField("spec.subscribable.subscriber[1].replyURI", "spec.subscribable.subscriber[1].subscriberURI")
+				fe.Details = "expected at least one of, got none"
+				return fe
+			}(),
+		},
+		"two empty subscribers": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					ChannelableSpec: eventingduck.ChannelableSpec{
+						SubscribableSpec: eventingduck.SubscribableSpec{
+							Subscribers: []eventingduck.SubscriberSpec{{}, {}},
+						},
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrMissingField("spec.subscribable.subscriber[0].replyURI", "spec.subscribable.subscriber[0].subscriberURI")
+				fe.Details = "expected at least one of, got none"
+				errs = errs.Also(fe)
+				fe = apis.ErrMissingField("spec.subscribable.subscriber[1].replyURI", "spec.subscribable.subscriber[1].subscriberURI")
+				fe.Details = "expected at least one of, got none"
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		"invalid scope annotation": {
+			cr: &KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						eventing.ScopeAnnotationKey: "notvalid",
+					},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue("notvalid", "metadata.annotations.[eventing.knative.dev/scope]")
+				fe.Details = "expected either 'cluster' or 'namespace'"
+				return fe
+			}(),
+		},
+	}
+
+	for n, test := range testCases {
+		t.Run(n, func(t *testing.T) {
+			got := test.cr.Validate(context.Background())
+			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+				t.Errorf("%s: validate (-want, +got) = %v", n, diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/receiver/path_test.go b/control-plane/pkg/receiver/path_test.go
index 09f564377..b6df44cf9 100644
--- a/control-plane/pkg/receiver/path_test.go
+++ b/control-plane/pkg/receiver/path_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package receiver
 
 import (
diff --git a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
index 9ee01a2ac..5a3cb6bdc 100644
--- a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
+++ b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package testing
 
 import (
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/reconciler/kafka/topic_test.go
index b2884fa7c..804d12492 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/reconciler/kafka/topic_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package kafka
 
 import (
diff --git a/test/e2e_new/resources/kafkasink/kafkasink.go b/test/e2e_new/resources/kafkasink/kafkasink.go
index 410ef010c..0fa409894 100644
--- a/test/e2e_new/resources/kafkasink/kafkasink.go
+++ b/test/e2e_new/resources/kafkasink/kafkasink.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package kafkasink
 
diff --git a/test/e2e_new/resources/kafkasink/kafkasink_test.go b/test/e2e_new/resources/kafkasink/kafkasink_test.go
index f1f86e6f1..54427418a 100644
--- a/test/e2e_new/resources/kafkasink/kafkasink_test.go
+++ b/test/e2e_new/resources/kafkasink/kafkasink_test.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package kafkasink_test