Repository: knative-sandbox/eventing-kafka
Pull request #798 (2021-08-09 06:00:10)

[Consolidated KafkaChannel] Set KafkaChannel consolidated dispatcher OwenrRef
## Proposed Changes

- Set the consolidated KafkaChannel controller deployment as the owner of the dispatcher deployment. This means that the way to uninstall the KafkaChannes is by deleting the controller deployment only. 

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- 🧽 The consolidated KafkaChannel dispatcher is now owned by the controller. 
```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/798#" title="Author self-approved">devguyio</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @devguyio: 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-unit-tests | df7324dcfafdf34c1452e70acc209a301758d016 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-unit-tests/1424611448591486976) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<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: @devguyio: 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-integration-test-channel-consolidated | e47d87ab6d54e22e42336ee3aa3edd563435e717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1424614009302159360) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

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

<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: @devguyio: 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-integration-test-channel-consolidated | e47d87ab6d54e22e42336ee3aa3edd563435e717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1424614009302159360) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | e47d87ab6d54e22e42336ee3aa3edd563435e717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1424614009377656832) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

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

<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: @devguyio: 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-integration-test-channel-consolidated | e47d87ab6d54e22e42336ee3aa3edd563435e717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1424614009302159360) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | e47d87ab6d54e22e42336ee3aa3edd563435e717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1424614009377656832) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | e47d87ab6d54e22e42336ee3aa3edd563435e717 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/798/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1424614009344102400) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

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

<details>

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798?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     #798      +/-   ##
==========================================
- Coverage   75.41%   74.69%   -0.73%     
==========================================
  Files         144      139       -5     
  Lines        6497     6370     -127     
==========================================
- Hits         4900     4758     -142     
- Misses       1376     1391      +15     
  Partials      221      221              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...l/consolidated/reconciler/controller/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9jb250cm9sbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [...consolidated/reconciler/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9rYWZrYWNoYW5uZWwuZ28=) | `49.20% <100.00%> (+0.16%)` | :arrow_up: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `96.77% <0.00%> (-0.04%)` | :arrow_down: |
| [test/rekt/resources/kafkatopic/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-dGVzdC9yZWt0L3Jlc291cmNlcy9rYWZrYXRvcGljL3RvcGljLmdv) | `31.57% <0.00%> (ø)` | |
| [test/rekt/resources/kafkacat/kafkacat.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-dGVzdC9yZWt0L3Jlc291cmNlcy9rYWZrYWNhdC9rYWZrYWNhdC5nbw==) | `66.66% <0.00%> (ø)` | |
| [test/rekt/resources/kafkasource/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-dGVzdC9yZWt0L3Jlc291cmNlcy9rYWZrYXNvdXJjZS9rYWZrYXNvdXJjZS5nbw==) | `74.19% <0.00%> (ø)` | |
| [test/rekt/resources/resetoffset/resetoffset.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-dGVzdC9yZWt0L3Jlc291cmNlcy9yZXNldG9mZnNldC9yZXNldG9mZnNldC5nbw==) | `42.30% <0.00%> (ø)` | |
| [test/rekt/resources/kafkachannel/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-dGVzdC9yZWt0L3Jlc291cmNlcy9rYWZrYWNoYW5uZWwva2Fma2FjaGFubmVsLmdv) | `26.66% <0.00%> (ø)` | |
| [pkg/channel/distributed/common/kafka/util/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL3V0aWwvdXRpbC5nbw==) | `100.00% <0.00%> (ø)` | |
| [...onciler/controller/resources/dispatcher\_service.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9yZXNvdXJjZXMvZGlzcGF0Y2hlcl9zZXJ2aWNlLmdv) | | |
| ... and [5 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798/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/pull/798?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [09a3149...e47d87a](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/798?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).


Patch:
diff --git a/config/channel/consolidated/500-dispatcher.yaml b/config/channel/consolidated/500-dispatcher.yaml
deleted file mode 120000
index 4baa3fded..000000000
--- a/config/channel/consolidated/500-dispatcher.yaml
+++ /dev/null
@@ -1,1 +0,0 @@
-deployments/dispatcher.yaml
\ No newline at end of file
diff --git a/config/channel/consolidated/deployments/controller.yaml b/config/channel/consolidated/deployments/controller.yaml
index 823d5cae4..acd482d54 100644
--- a/config/channel/consolidated/deployments/controller.yaml
+++ b/config/channel/consolidated/deployments/controller.yaml
@@ -18,6 +18,8 @@ metadata:
   name: kafka-ch-controller
   namespace: knative-eventing
   labels:
+    messaging.knative.dev/channel: kafka-channel
+    messaging.knative.dev/role: controller
     kafka.eventing.knative.dev/release: devel
 spec:
   replicas: 1
diff --git a/config/channel/consolidated/deployments/dispatcher.yaml b/config/channel/consolidated/deployments/dispatcher.yaml
deleted file mode 100644
index 90c462cfc..000000000
--- a/config/channel/consolidated/deployments/dispatcher.yaml
+++ /dev/null
@@ -1,103 +0,0 @@
-# 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.
-
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: kafka-ch-dispatcher
-  namespace: knative-eventing
-  labels:
-    kafka.eventing.knative.dev/release: devel
-spec:
-  # this deployment is going to be scaled up by the
-  # controller when the very first KafkaChannel is created
-  replicas: 0
-  selector:
-    matchLabels:
-      messaging.knative.dev/channel: kafka-channel
-      messaging.knative.dev/role: dispatcher
-  template:
-    metadata:
-      annotations:
-        # This annotation is used by the controller to track updates
-        # to config-kafka and apply them in the dispatcher deployment
-        kafka.eventing.knative.dev/configmap-hash: ''
-      labels:
-        # Do not change. Used by the controller for probing.
-        messaging.knative.dev/channel: kafka-channel
-        # Do not change. Used by the controller for probing.
-        messaging.knative.dev/role: dispatcher
-        kafka.eventing.knative.dev/release: devel
-    spec:
-      containers:
-      - name: dispatcher
-        image: ko://knative.dev/eventing-kafka/cmd/channel/consolidated/dispatcher
-        env:
-        - name: SYSTEM_NAMESPACE
-          value: ''
-          valueFrom:
-            fieldRef:
-              apiVersion: v1
-              fieldPath: metadata.namespace
-        - name: POD_NAME
-          valueFrom:
-            fieldRef:
-              fieldPath: metadata.name
-        - name: METRICS_DOMAIN
-          value: knative.dev/eventing
-        - name: CONFIG_LOGGING_NAME
-          value: config-logging
-        - name: CONFIG_LEADERELECTION_NAME
-          value: config-leader-election
-        - name: CONTAINER_NAME
-          value: dispatcher
-        ports:
-        - containerPort: 9090
-          name: metrics
-          protocol: TCP
-        - containerPort: 8081
-          name: sub-status
-          protocol: TCP
-        volumeMounts:
-        - name: config-kafka
-          mountPath: /etc/config-kafka
-      serviceAccountName: kafka-ch-dispatcher
-      volumes:
-      - name: config-kafka
-        configMap:
-          name: config-kafka
-
----
-
-apiVersion: v1
-kind: Service
-metadata:
-  labels:
-    messaging.knative.dev/channel: kafka-channel
-    messaging.knative.dev/role: dispatcher
-  name: kafka-ch-dispatcher
-  namespace: knative-eventing
-spec:
-  ports:
-  - name: http-dispatcher
-    port: 80
-    protocol: TCP
-    targetPort: 8080
-  - name: http-sub-status
-    port: 8081
-    protocol: TCP
-    targetPort: 8081
-  selector:
-    messaging.knative.dev/channel: kafka-channel
-    messaging.knative.dev/role: dispatcher
diff --git a/go.mod b/go.mod
index 5a03bf207..19e53d045 100644
--- a/go.mod
+++ b/go.mod
@@ -35,7 +35,7 @@ require (
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20210803191615-72cde96b9f5a
-	knative.dev/eventing v0.24.1-0.20210806092320-e7445c5583e7
+	knative.dev/eventing v0.25.0
 	knative.dev/hack v0.0.0-20210622141627-e28525d8d260
 	knative.dev/networking v0.0.0-20210803181815-acdfd41c575c
 	knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5
diff --git a/go.sum b/go.sum
index fd6df39f6..a83669c07 100644
--- a/go.sum
+++ b/go.sum
@@ -38,7 +38,6 @@ contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d h
 contrib.go.opencensus.io/exporter/ocagent v0.7.1-0.20200907061046-05415f1de66d/go.mod h1:IshRmMJBhDfFj5Y67nVhMYTTIze91RUeT73ipWKs/GY=
 contrib.go.opencensus.io/exporter/prometheus v0.3.0 h1:08FMdJYpItzsknogU6PiiNo7XQZg/25GjH236+YCwD0=
 contrib.go.opencensus.io/exporter/prometheus v0.3.0/go.mod h1:rpCPVQKhiyH8oomWgm34ZmgIdZa8OVYO5WAIygPbBBE=
-contrib.go.opencensus.io/exporter/stackdriver v0.13.5/go.mod h1:aXENhDJ1Y4lIg4EUaVTwzvYETVNZk10Pu26tevFKLUc=
 contrib.go.opencensus.io/exporter/zipkin v0.1.2 h1:YqE293IZrKtqPnpwDPH/lOqTWD/s3Iwabycam74JV3g=
 contrib.go.opencensus.io/exporter/zipkin v0.1.2/go.mod h1:mP5xM3rrgOjpn79MM8fZbj3gsxcuytSqtH0dxSWW1RE=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
@@ -136,9 +135,7 @@ github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj
 github.com/aryann/difflib v0.0.0-20170710044230-e206f873d14a/go.mod h1:DAHtR1m6lCRdSC2Tm3DSWRPvIPr6xNKyeHdqDQSQT+A=
 github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/aws/aws-lambda-go v1.13.3/go.mod h1:4UKl9IzQMoD+QF79YdCuzCwp8VbmG4VAQwij/eHl5CU=
-github.com/aws/aws-sdk-go v1.23.20/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.27.0/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
-github.com/aws/aws-sdk-go v1.31.12/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
 github.com/aws/aws-sdk-go-v2 v0.18.0/go.mod h1:JWVYvqSMppoMJC0x5wdwiImzgXTI9FuZwxzkQq9wy+g=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
@@ -235,7 +232,6 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
@@ -294,7 +290,6 @@ github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh
 github.com/go-openapi/swag v0.19.15 h1:D2NRCBzS9/pEY3gP9Nl8aDqGUcPFrwG2p+CNFrLyrCM=
 github.com/go-openapi/swag v0.19.15/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
-github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gobuffalo/flect v0.2.3 h1:f/ZukRnSNA/DUpSNDadko7Qc0PhGvsew35p/2tu+CRY=
 github.com/gobuffalo/flect v0.2.3/go.mod h1:vmkQwuZYhN5Pc4ljYQZzP+1sq+NEkK+lh20jmEmX3jc=
@@ -392,7 +387,6 @@ github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm4
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
@@ -417,7 +411,6 @@ github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgf
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/grpc-ecosystem/grpc-gateway v1.14.6/go.mod h1:zdiPV4Yse/1gnckTHtghG4GkDEdKCRJduHpTxT3/jcw=
-github.com/grpc-ecosystem/grpc-gateway v1.14.8/go.mod h1:NZE8t6vs6TnwLL/ITkaK8W3ecMLGAbh2jXTclvpiwYo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
@@ -482,7 +475,6 @@ github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZ
 github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
-github.com/jmespath/go-jmespath v0.3.0/go.mod h1:9QtRXoHjLGCJ5IBSaohpXITPlowMeeYCZ7fLUTSywik=
 github.com/joho/godotenv v1.3.0 h1:Zjp+RcGpHhGlrMbJzXTrZZPrWj+1vfm90La1wgB6Bhc=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
@@ -793,7 +785,6 @@ go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
-go.uber.org/atomic v1.8.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/automaxprocs v1.4.0 h1:CpDZl6aOlLhReez+8S3eEotD7Jx0Os++lemPlMULQP0=
@@ -885,7 +876,6 @@ golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191002035440-2ec189313ef0/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -1036,7 +1026,6 @@ golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgw
 golang.org/x/tools v0.0.0-20190816200558-6889da9d5479/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20190907020128-2ca718005c18/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20191010075000-0337d82405ff/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -1076,7 +1065,6 @@ golang.org/x/tools v0.0.0-20201110124207-079ba7bd75cd/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.0.0-20201201161351-ac6f37ff4c2a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
-golang.org/x/tools v0.1.4/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.5 h1:ouewzE6p+/VEB31YYnTbEJdi8pFqKp4P4n85vwo3DHA=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -1094,7 +1082,6 @@ google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEt
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
 google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
 google.golang.org/api v0.9.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
-google.golang.org/api v0.10.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
 google.golang.org/api v0.13.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsbkAI=
 google.golang.org/api v0.14.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsbkAI=
 google.golang.org/api v0.15.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsbkAI=
@@ -1116,7 +1103,6 @@ google.golang.org/appengine v1.2.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0=
-google.golang.org/appengine v1.6.2/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0=
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/appengine v1.6.6/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6c=
@@ -1181,7 +1167,6 @@ google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTp
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
 google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
 google.golang.org/grpc v1.39.0 h1:Klz8I9kdtkIN6EpHHUOMLCYhTn/2WAe5a0s1hcBkdTI=
 google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
@@ -1282,18 +1267,15 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210803191615-72cde96b9f5a h1:VbJj20YhSx3j+OE8DhncTt+TYQrb+rAS1BGPoeo3GF0=
 knative.dev/control-protocol v0.0.0-20210803191615-72cde96b9f5a/go.mod h1:9gYmAq6zVKUA+p1+DXslH91jtga5q2z+YJWcH+z88EE=
-knative.dev/eventing v0.24.1-0.20210806092320-e7445c5583e7 h1:qdFn+2z9djrKb4KkpAzB6CCzMcWKeOJ3MqugIL3Jgg0=
-knative.dev/eventing v0.24.1-0.20210806092320-e7445c5583e7/go.mod h1:mYN9AeU95zB9QT9ATmYiq0i7k41u69xD0jW1udHdPrc=
+knative.dev/eventing v0.25.0 h1:lBKgQFGvyeUyvf+HOyuxFd5cXx+SMqnzqtPi2hXiCi4=
+knative.dev/eventing v0.25.0/go.mod h1:8jIsrnSONPgv+m63OTzpwZQJiQASYl77C3llCyYlBMU=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260 h1:f2eMtOubAOc/Q7JlvFPDKXiPlJVK+VpX2Cot8hRzCgQ=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210622141627-e28525d8d260/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/networking v0.0.0-20210803181815-acdfd41c575c h1:7G6TQr7ZyIHx35Dn5zuNKUDhlly3KkFxgrKLXeKmjj8=
 knative.dev/networking v0.0.0-20210803181815-acdfd41c575c/go.mod h1:UA9m1M3rGssy63gVwjSh7CYoWTKZNO8cnY9QsIu7tyo=
-knative.dev/pkg v0.0.0-20210713175550-7d1b0f19ef71/go.mod h1:d06VsGUT2grnlhA8GcUjsoePok8P859AYcIqWISMGtY=
-knative.dev/pkg v0.0.0-20210803032247-552bbc106170/go.mod h1:NYZRIPU+Pv39VfbZV1BtMIe4kCavNle1udsPrvOLm+Y=
 knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5 h1:jpOTmAXg1oLS8u5HPBrFP1XsOSFCQIvlTRxP8TDGg2E=
 knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5/go.mod h1:RPk5txNA3apR9X40D4MpUOP9/VqOG8CrtABWfOwGVS4=
-knative.dev/reconciler-test v0.0.0-20210714231231-91e5b206e1d5/go.mod h1:Zp2WHfJuh6+rOpOAq78My82drT4Bxjm/K4etmKcA7ko=
 knative.dev/reconciler-test v0.0.0-20210803183715-b61cc77c06f6 h1:jSz98FX9JfAMJX3qVeRF7RU7A3XLJJxBNz8GYU5z0bM=
 knative.dev/reconciler-test v0.0.0-20210803183715-b61cc77c06f6/go.mod h1:+Kovy+G5zXZNcuO/uB+zfo37vFKZzsLIlWezt/nKMz0=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
diff --git a/hack/release.sh b/hack/release.sh
index 829620238..167978542 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -26,8 +26,10 @@ declare -A COMPONENTS
 COMPONENTS=(
   ["channel-consolidated.yaml"]="config/channel/consolidated"
   ["channel-distributed.yaml"]="config/channel/distributed"
+  ["channel-crds.yaml"]="config/channel/resources"
   ["source.yaml"]="config/source/single"
   ["mt-source.yaml"]="config/source/multi"
+  ["source-crds.yaml"]="config/source/common/resources"
 )
 readonly COMPONENTS
 
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go b/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
index ff2416f0f..a51986ce1 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
@@ -33,13 +33,9 @@ func (c *KafkaChannel) SetDefaults(ctx context.Context) {
 		c.Annotations = make(map[string]string)
 	}
 
-	// TODO this should be wrapped in a if to avoid conversion of ducks and let some other duck controller do that?!
-	//   BUT we need this hack to properly update from v1alpha1 to v1beta1, so:
-	//   * KEEP THIS FOR THE WHOLE 0.22 LIFECYCLE
-	//   * REMOVE THIS BEFORE 0.23 RELEASE
-	//if _, ok := c.Annotations[messaging.SubscribableDuckVersionAnnotation]; !ok {
-	c.Annotations[messaging.SubscribableDuckVersionAnnotation] = "v1"
-	//}
+	if _, ok := c.Annotations[messaging.SubscribableDuckVersionAnnotation]; !ok {
+		c.Annotations[messaging.SubscribableDuckVersionAnnotation] = "v1"
+	}
 
 	c.Spec.SetDefaults(ctx)
 }
diff --git a/pkg/channel/consolidated/reconciler/controller/controller.go b/pkg/channel/consolidated/reconciler/controller/controller.go
index 2aa823051..f96afaf1f 100644
--- a/pkg/channel/consolidated/reconciler/controller/controller.go
+++ b/pkg/channel/consolidated/reconciler/controller/controller.go
@@ -18,24 +18,22 @@ package controller
 
 import (
 	"context"
+	"fmt"
+	"time"
 
 	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
-
-	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	"knative.dev/eventing-kafka/pkg/channel/consolidated/status"
-	kafkamessagingv1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1beta1"
-	kafkaChannelClient "knative.dev/eventing-kafka/pkg/client/injection/client"
-	"knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
-	kafkaChannelReconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
-	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+	"k8s.io/utils/pointer"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingClient "knative.dev/eventing/pkg/client/injection/client"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
+	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
 	endpointsinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/service"
@@ -46,13 +44,24 @@ import (
 	"knative.dev/pkg/logging"
 	knativeReconciler "knative.dev/pkg/reconciler"
 	"knative.dev/pkg/system"
+
+	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/status"
+	kafkamessagingv1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1beta1"
+	kafkaChannelClient "knative.dev/eventing-kafka/pkg/client/injection/client"
+	"knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
+	kafkaChannelReconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 )
 
 const (
-	channelLabelKey   = "messaging.knative.dev/channel"
-	channelLabelValue = "kafka-channel"
-	roleLabelKey      = "messaging.knative.dev/role"
-	roleLabelValue    = "dispatcher"
+	channelLabelKey          = "messaging.knative.dev/channel"
+	channelLabelValue        = "kafka-channel"
+	roleLabelKey             = "messaging.knative.dev/role"
+	dispatcherRoleLabelValue = "dispatcher"
+	controllerRoleLabelValue = "controller"
+	interval                 = 100 * time.Millisecond
+	timeout                  = 5 * time.Minute
 )
 
 // NewController initializes the controller and is called by the generated code.
@@ -63,7 +72,7 @@ func NewController(
 ) *controller.Impl {
 	logger := logging.FromContext(ctx)
 	kafkaChannelInformer := kafkachannel.Get(ctx)
-	deploymentInformer := deployment.Get(ctx)
+	deploymentInformer := deploymentinformer.Get(ctx)
 	endpointsInformer := endpointsinformer.Get(ctx)
 	serviceAccountInformer := serviceaccount.Get(ctx)
 	roleBindingInformer := rolebinding.Get(ctx)
@@ -92,6 +101,13 @@ func NewController(
 	r.dispatcherImage = env.Image
 	r.dispatcherServiceAccount = env.DispatcherServiceAccount
 
+	// get the ref of the controller deployment
+	ownerRef, err := getControllerOwnerRef(ctx)
+	if err != nil {
+		logger.Fatalw("Could not determine the proper owner reference for the dispatcher deployment.", zap.Error(err))
+	}
+	r.controllerRef = *ownerRef
+
 	impl := kafkaChannelReconciler.NewImpl(ctx, r)
 
 	statusProber := status.NewProber(
@@ -118,7 +134,7 @@ func NewController(
 	}
 
 	// Get and Watch the Kakfa config map and dynamically update Kafka configuration.
-	err := commonconfig.InitializeKafkaConfigMapWatcher(ctx, cmw, logger, handleKafkaConfigMapChange, system.Namespace())
+	err = commonconfig.InitializeKafkaConfigMapWatcher(ctx, cmw, logger, handleKafkaConfigMapChange, system.Namespace())
 	if err != nil {
 		logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
 	}
@@ -155,7 +171,7 @@ func NewController(
 	podInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: knativeReconciler.ChainFilterFuncs(
 			knativeReconciler.LabelFilterFunc(channelLabelKey, channelLabelValue, false),
-			knativeReconciler.LabelFilterFunc(roleLabelKey, roleLabelValue, false),
+			knativeReconciler.LabelFilterFunc(roleLabelKey, dispatcherRoleLabelValue, false),
 		),
 		Handler: cache.ResourceEventHandlerFuncs{
 			// Cancel probing when a Pod is deleted
@@ -178,3 +194,38 @@ func getPodInformerEventHandler(ctx context.Context, logger *zap.SugaredLogger,
 		}
 	}
 }
+
+func getControllerOwnerRef(ctx context.Context) (*metav1.OwnerReference, error) {
+	logger := logging.FromContext(ctx)
+	ctrlDeploymentLabels := labels.Set{
+		channelLabelKey: channelLabelValue,
+		roleLabelKey:    controllerRoleLabelValue,
+	}
+
+	ownerRef := metav1.OwnerReference{
+		APIVersion: "apps/v1",
+		Kind:       "Deployment",
+		Controller: pointer.BoolPtr(true),
+	}
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		k8sClient := kubeclient.Get(ctx)
+		deploymentList, err := k8sClient.AppsV1().Deployments(system.Namespace()).List(ctx, metav1.ListOptions{LabelSelector: ctrlDeploymentLabels.String()})
+		if err != nil {
+			return true, fmt.Errorf("error listing KafkaChannel controller deployment labels %w", err)
+		} else if len(deploymentList.Items) == 0 {
+			// Simple exponential backoff
+			logger.Debugw("found zero KafkaChannel controller deployment matching labels. Retrying.", zap.String("namespace", system.Namespace()), zap.Any("selectors", ctrlDeploymentLabels.AsSelector()))
+			return false, nil
+		} else if len(deploymentList.Items) > 1 {
+			return true, fmt.Errorf("found an unexpected number of KafkaChannel controller deployment matching labels. Got: %d, Want: 1", len(deploymentList.Items))
+		}
+		d := deploymentList.Items[0]
+		ownerRef.Name = d.Name
+		ownerRef.UID = d.UID
+		return true, nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to determine the deployment of the KafkaChannel controller based on labels. %w", err)
+	}
+	return &ownerRef, nil
+}
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index b346bba85..aaa0f2060 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -22,6 +22,7 @@ import (
 	"strconv"
 
 	"github.com/Shopify/sarama"
+	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -35,9 +36,9 @@ import (
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	rbacv1listers "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/utils/pointer"
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/apis/eventing"
+	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/apis/duck"
 	"knative.dev/pkg/controller"
@@ -56,9 +57,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/common/client"
 	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	"knative.dev/eventing-kafka/pkg/common/constants"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
-	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
 )
 
 const (
@@ -135,6 +134,7 @@ type Reconciler struct {
 	serviceAccountLister corev1listers.ServiceAccountLister
 	roleBindingLister    rbacv1listers.RoleBindingLister
 	statusManager        status.Manager
+	controllerRef        metav1.OwnerReference
 }
 
 type envConfig struct {
@@ -286,6 +286,7 @@ func (r *Reconciler) reconcileSubscribers(ctx context.Context, ch *v1beta1.Kafka
 }
 
 func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope string, dispatcherNamespace string, kc *v1beta1.KafkaChannel) error {
+	logger := logging.FromContext(ctx)
 	if scope == scopeNamespace {
 		// Configure RBAC in namespace to access the configmaps
 		sa, err := r.reconcileServiceAccount(ctx, dispatcherNamespace, kc)
@@ -315,93 +316,50 @@ func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope string, disp
 		Replicas:            1,
 		ServiceAccount:      r.dispatcherServiceAccount,
 		ConfigMapHash:       r.kafkaConfigMapHash,
+		OwnerRef:            r.controllerRef,
 	}
 
-	expected := resources.MakeDispatcher(args)
+	want := resources.NewDispatcherBuilder().WithArgs(&args).Build()
 	d, err := r.deploymentLister.Deployments(dispatcherNamespace).Get(dispatcherName)
 	if err != nil {
 		if apierrs.IsNotFound(err) {
-			d, err := r.KubeClientSet.AppsV1().Deployments(dispatcherNamespace).Create(ctx, expected, metav1.CreateOptions{})
+			d, err := r.KubeClientSet.AppsV1().Deployments(dispatcherNamespace).Create(ctx, want, metav1.CreateOptions{})
 			if err == nil {
 				controller.GetEventRecorder(ctx).Event(kc, corev1.EventTypeNormal, dispatcherDeploymentCreated, "Dispatcher deployment created")
 				kc.Status.PropagateDispatcherStatus(&d.Status)
 				return err
 			} else {
+				logger.Errorw("error while creating dispatcher deployment", zap.Error(err), zap.String("namespace", dispatcherNamespace), zap.Any("deployment", want))
 				kc.Status.MarkDispatcherFailed(dispatcherDeploymentFailed, "Failed to create the dispatcher deployment: %v", err)
 				return newDeploymentWarn(err)
 			}
 		}
-
-		logging.FromContext(ctx).Errorw("Unable to get the dispatcher deployment", zap.Error(err))
+		logger.Errorw("can't get dispatcher deployment", zap.Error(err), zap.String("namespace", dispatcherNamespace),
+			zap.String("dispatcher-name", dispatcherName))
 		kc.Status.MarkDispatcherUnknown("DispatcherDeploymentFailed", "Failed to get dispatcher deployment: %v", err)
 		return err
 	} else {
-		existing := utils.FindContainer(d, resources.DispatcherContainerName)
-		if existing == nil {
-			logging.FromContext(ctx).Errorw("Container %s does not exist in existing dispatcher deployment. Updating the deployment", resources.DispatcherContainerName)
-			d, err := r.KubeClientSet.AppsV1().Deployments(dispatcherNamespace).Update(ctx, expected, metav1.UpdateOptions{})
-			if err == nil {
-				controller.GetEventRecorder(ctx).Event(kc, corev1.EventTypeNormal, dispatcherDeploymentUpdated, "Dispatcher deployment updated")
-				kc.Status.PropagateDispatcherStatus(&d.Status)
-				return nil
-			} else {
-				kc.Status.MarkServiceFailed("DispatcherDeploymentUpdateFailed", "Failed to update the dispatcher deployment: %v", err)
-			}
-			return newDeploymentWarn(err)
-		}
-
-		expectedContainer := utils.FindContainer(expected, resources.DispatcherContainerName)
-		if expectedContainer == nil {
-			return fmt.Errorf("container %s does not exist in expected dispatcher deployment. Cannot check if the deployment needs an update", resources.DispatcherContainerName)
-		}
-
-		expectedConfigMapHash := r.kafkaConfigMapHash
-
-		needsUpdate := false
-		// do not touch the original deployment, deepcopy it
-		deploymentCopy := d.DeepCopy()
-		existingCopy := utils.FindContainer(deploymentCopy, resources.DispatcherContainerName)
-
-		if existingCopy.Image != expectedContainer.Image {
-			logging.FromContext(ctx).Infof("Dispatcher deployment image is not what we expect it to be, updating Deployment Got: %q Expect: %q", expected.Spec.Template.Spec.Containers[0].Image, d.Spec.Template.Spec.Containers[0].Image)
-			existingCopy.Image = expectedContainer.Image
-			needsUpdate = true
-		}
-
-		if *deploymentCopy.Spec.Replicas == 0 {
-			logging.FromContext(ctx).Infof("Dispatcher deployment has 0 replica. Scaling up deployment to 1 replica")
-			deploymentCopy.Spec.Replicas = pointer.Int32Ptr(1)
-			needsUpdate = true
-		}
-
-		if deploymentCopy.Spec.Template.Annotations == nil {
-			logging.FromContext(ctx).Infof("Configmap hash is not set. Updating the dispatcher deployment.")
-			deploymentCopy.Spec.Template.Annotations = map[string]string{
-				commonconstants.ConfigMapHashAnnotationKey: expectedConfigMapHash,
-			}
-			needsUpdate = true
-		}
-
-		if deploymentCopy.Spec.Template.Annotations[commonconstants.ConfigMapHashAnnotationKey] != expectedConfigMapHash {
-			logging.FromContext(ctx).Infof("Configmap hash is changed. Updating the dispatcher deployment.")
-			deploymentCopy.Spec.Template.Annotations[commonconstants.ConfigMapHashAnnotationKey] = expectedConfigMapHash
-			needsUpdate = true
+		// scale up the dispatcher to 1, otherwise keep the existing number in case the user has scaled it up.
+		if *d.Spec.Replicas == 0 {
+			logger.Infof("Dispatcher deployment has 0 replica. Scaling up deployment to 1 replica")
+			args.Replicas = 1
+		} else {
+			args.Replicas = *d.Spec.Replicas
 		}
+		want = resources.NewDispatcherBuilderFromDeployment(d.DeepCopy()).WithArgs(&args).Build()
 
-		if needsUpdate {
-			deploymentCopy, err = r.KubeClientSet.AppsV1().Deployments(dispatcherNamespace).Update(ctx, deploymentCopy, metav1.UpdateOptions{})
-			if err == nil {
-				controller.GetEventRecorder(ctx).Event(kc, corev1.EventTypeNormal, dispatcherDeploymentUpdated, "Dispatcher deployment updated")
-				kc.Status.PropagateDispatcherStatus(&deploymentCopy.Status)
-				return nil
-			} else {
+		if !equality.Semantic.DeepEqual(want.Spec, d.Spec) {
+			logger.Infof("Dispatcher deployment changed; reconciling:\n%s", cmp.Diff(want.Spec, d.Spec))
+			if d, err = r.KubeClientSet.AppsV1().Deployments(dispatcherNamespace).Update(ctx, want, metav1.UpdateOptions{}); err != nil {
+				logger.Errorw("error while updating dispatcher deployment", zap.Error(err), zap.String("namespace", dispatcherNamespace), zap.Any("deployment", want))
 				kc.Status.MarkServiceFailed("DispatcherDeploymentUpdateFailed", "Failed to update the dispatcher deployment: %v", err)
 				return newDeploymentWarn(err)
+			} else {
+				controller.GetEventRecorder(ctx).Event(kc, corev1.EventTypeNormal, dispatcherDeploymentUpdated, "Dispatcher deployment updated")
 			}
-		} else {
-			kc.Status.PropagateDispatcherStatus(&d.Status)
-			return nil
 		}
+		kc.Status.PropagateDispatcherStatus(&d.Status)
+		return nil
 	}
 }
 
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
index 9dcb60989..5686d9183 100644
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
+++ b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
@@ -20,9 +20,10 @@ import (
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/system"
+
 	"knative.dev/eventing-kafka/pkg/common/constants"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
-	"knative.dev/pkg/system"
 )
 
 const (
@@ -37,6 +38,11 @@ var (
 	}
 )
 
+type DispatcherBuilder struct {
+	deployment *v1.Deployment
+	args       *DispatcherArgs
+}
+
 type DispatcherArgs struct {
 	DispatcherScope     string
 	DispatcherNamespace string
@@ -44,11 +50,53 @@ type DispatcherArgs struct {
 	Replicas            int32
 	ServiceAccount      string
 	ConfigMapHash       string
+	OwnerRef            metav1.OwnerReference
+}
+
+// NewDispatcherBuilder returns a builder which builds from scratch a dispatcher deployment.
+// Intended to be used when creating the dispatcher deployment for the first time.
+func NewDispatcherBuilder() *DispatcherBuilder {
+	b := &DispatcherBuilder{}
+	b.deployment = dispatcherTemplate()
+	return b
+}
+
+// NewDispatcherBuilderFromDeployment returns a builder which builds a dispatcher deployment from the given deployment.
+// Intended to be used when updating an existing dispatcher deployment.
+func NewDispatcherBuilderFromDeployment(d *v1.Deployment) *DispatcherBuilder {
+	b := &DispatcherBuilder{}
+	b.deployment = d
+	return b
 }
 
-// MakeDispatcher generates the dispatcher deployment for the KafKa channel
-func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
-	replicas := args.Replicas
+func (b *DispatcherBuilder) WithArgs(args *DispatcherArgs) *DispatcherBuilder {
+	b.args = args
+	return b
+}
+
+func (b *DispatcherBuilder) Build() *v1.Deployment {
+	replicas := b.args.Replicas
+	b.deployment.ObjectMeta.Namespace = b.args.DispatcherNamespace
+	b.deployment.ObjectMeta.OwnerReferences = []metav1.OwnerReference{b.args.OwnerRef}
+	b.deployment.Spec.Replicas = &replicas
+	b.deployment.Spec.Template.ObjectMeta.Annotations = map[string]string{
+		commonconstants.ConfigMapHashAnnotationKey: b.args.ConfigMapHash,
+	}
+	b.deployment.Spec.Template.Spec.ServiceAccountName = b.args.ServiceAccount
+
+	for i, c := range b.deployment.Spec.Template.Spec.Containers {
+		if c.Name == DispatcherContainerName {
+			container := &b.deployment.Spec.Template.Spec.Containers[i]
+			container.Image = b.args.Image
+			if container.Env == nil {
+				container.Env = makeEnv(b.args)
+			}
+		}
+	}
+	return b.deployment
+}
+
+func dispatcherTemplate() *v1.Deployment {
 
 	return &v1.Deployment{
 		TypeMeta: metav1.TypeMeta{
@@ -56,28 +104,20 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 			Kind:       "Deployments",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      dispatcherName,
-			Namespace: args.DispatcherNamespace,
+			Name: dispatcherName,
 		},
 		Spec: v1.DeploymentSpec{
-			Replicas: &replicas,
 			Selector: &metav1.LabelSelector{
 				MatchLabels: dispatcherLabels,
 			},
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: dispatcherLabels,
-					Annotations: map[string]string{
-						commonconstants.ConfigMapHashAnnotationKey: args.ConfigMapHash,
-					},
 				},
 				Spec: corev1.PodSpec{
-					ServiceAccountName: args.ServiceAccount,
 					Containers: []corev1.Container{
 						{
-							Name:  DispatcherContainerName,
-							Image: args.Image,
-							Env:   makeEnv(args),
+							Name: DispatcherContainerName,
 							Ports: []corev1.ContainerPort{{
 								Name:          "metrics",
 								ContainerPort: 9090,
@@ -108,7 +148,7 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 	}
 }
 
-func makeEnv(args DispatcherArgs) []corev1.EnvVar {
+func makeEnv(args *DispatcherArgs) []corev1.EnvVar {
 	vars := []corev1.EnvVar{{
 		Name:  system.NamespaceEnvKey,
 		Value: system.Namespace(),
@@ -131,18 +171,20 @@ func makeEnv(args DispatcherArgs) []corev1.EnvVar {
 					FieldPath: "metadata.namespace",
 				},
 			},
-		}, corev1.EnvVar{
-			Name: "POD_NAME",
-			ValueFrom: &corev1.EnvVarSource{
-				FieldRef: &corev1.ObjectFieldSelector{
-					FieldPath: "metadata.name",
-				},
-			},
-		}, corev1.EnvVar{
-			Name:  "CONTAINER_NAME",
-			Value: "dispatcher",
 		})
 	}
 
+	vars = append(vars, corev1.EnvVar{
+		Name: "POD_NAME",
+		ValueFrom: &corev1.EnvVarSource{
+			FieldRef: &corev1.ObjectFieldSelector{
+				APIVersion: "",
+				FieldPath:  "metadata.name",
+			},
+		},
+	}, corev1.EnvVar{
+		Name:  "CONTAINER_NAME",
+		Value: "dispatcher",
+	})
 	return vars
 }
diff --git a/vendor/modules.txt b/vendor/modules.txt
index b8a150cf5..7e59c460a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1046,7 +1046,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.24.1-0.20210806092320-e7445c5583e7
+# knative.dev/eventing v0.25.0
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test


Test patch:
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
index 02da8abdc..b25d11263 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
@@ -612,7 +612,6 @@ func TestDeploymentMoreThanOneReplicas(t *testing.T) {
 				reconcilertesting.WithKafkaFinalizer(finalizerName),
 				reconcilertesting.WithKafkaChannelConfigReady(),
 				reconcilertesting.WithKafkaChannelTopicReady(),
-				//				reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
 				reconcilertesting.WithKafkaChannelServiceReady(),
 				reconcilertesting.WithKafkaChannelEndpointsReady(),
 				reconcilertesting.WithKafkaChannelChannelServiceReady(),
@@ -853,13 +852,14 @@ func (ca *mockClusterAdmin) DeleteConsumerGroup(group string) error {
 var _ sarama.ClusterAdmin = (*mockClusterAdmin)(nil)
 
 func makeDeploymentWithParams(image string, replicas int32, configMapHash string) *appsv1.Deployment {
-	return resources.MakeDispatcher(resources.DispatcherArgs{
+	args := resources.DispatcherArgs{
 		DispatcherNamespace: testNS,
 		Image:               image,
 		Replicas:            replicas,
 		ServiceAccount:      testDispatcherserviceAccount,
 		ConfigMapHash:       configMapHash,
-	})
+	}
+	return resources.NewDispatcherBuilder().WithArgs(&args).Build()
 }
 
 func makeDeploymentWithImageAndReplicas(image string, replicas int32) *appsv1.Deployment {
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service_test.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service_test.go
deleted file mode 100644
index 729e27f8f..000000000
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service_test.go
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
-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 resources
-
-import (
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/intstr"
-)
-
-func TestNewDispatcherService(t *testing.T) {
-	want := &corev1.Service{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "v1",
-			Kind:       "Service",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      dispatcherName,
-			Namespace: testNS,
-			Labels:    dispatcherLabels,
-		},
-		Spec: corev1.ServiceSpec{
-			Selector: dispatcherLabels,
-			Ports: []corev1.ServicePort{
-				{
-					Name:       "http-dispatcher",
-					Protocol:   corev1.ProtocolTCP,
-					Port:       80,
-					TargetPort: intstr.IntOrString{IntVal: 8080},
-				},
-			},
-		},
-	}
-
-	got := MakeDispatcherService(testNS)
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
deleted file mode 100644
index 7cb872d30..000000000
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
-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 resources
-
-import (
-	"os"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	v1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
-	"knative.dev/pkg/system"
-)
-
-const (
-	imageName      = "my-test-image"
-	serviceAccount = "kafka-ch-dispatcher"
-)
-
-func TestNewDispatcher(t *testing.T) {
-	os.Setenv(system.NamespaceEnvKey, "knative-testing")
-
-	args := DispatcherArgs{
-		DispatcherScope:     "cluster",
-		DispatcherNamespace: testNS,
-		Image:               imageName,
-		Replicas:            1,
-		ServiceAccount:      serviceAccount,
-		ConfigMapHash:       testConfigMapHash,
-	}
-
-	replicas := int32(1)
-	want := &v1.Deployment{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "apps/v1",
-			Kind:       "Deployments",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: testNS,
-			Name:      dispatcherName,
-		},
-		Spec: v1.DeploymentSpec{
-			Replicas: &replicas,
-			Selector: &metav1.LabelSelector{
-				MatchLabels: dispatcherLabels,
-			},
-			Template: corev1.PodTemplateSpec{
-				ObjectMeta: metav1.ObjectMeta{
-					Labels: dispatcherLabels,
-					Annotations: map[string]string{
-						commonconstants.ConfigMapHashAnnotationKey: testConfigMapHash,
-					},
-				},
-				Spec: corev1.PodSpec{
-					ServiceAccountName: serviceAccount,
-					Containers: []corev1.Container{
-						{
-							Name:  "dispatcher",
-							Image: imageName,
-							Env: []corev1.EnvVar{{
-								Name:  system.NamespaceEnvKey,
-								Value: "knative-testing",
-							}, {
-								Name:  "METRICS_DOMAIN",
-								Value: "knative.dev/eventing",
-							}, {
-								Name:  "CONFIG_LOGGING_NAME",
-								Value: "config-logging",
-							}, {
-								Name:  "CONFIG_LEADERELECTION_NAME",
-								Value: "config-leader-election",
-							}},
-							Ports: []corev1.ContainerPort{{
-								Name:          "metrics",
-								ContainerPort: 9090,
-							}},
-							VolumeMounts: []corev1.VolumeMount{
-								{
-									Name:      "config-kafka",
-									MountPath: "/etc/config-kafka",
-								},
-							},
-						}},
-					Volumes: []corev1.Volume{
-						{
-							Name: "config-kafka",
-							VolumeSource: corev1.VolumeSource{
-								ConfigMap: &corev1.ConfigMapVolumeSource{
-									LocalObjectReference: corev1.LocalObjectReference{
-										Name: "config-kafka",
-									},
-								},
-							},
-						}},
-				},
-			},
-		},
-	}
-
-	got := MakeDispatcher(args)
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
-
-func TestNewNamespaceDispatcher(t *testing.T) {
-	os.Setenv(system.NamespaceEnvKey, "knative-testing")
-
-	args := DispatcherArgs{
-		DispatcherScope:     "namespace",
-		DispatcherNamespace: testNS,
-		Image:               imageName,
-		Replicas:            1,
-		ServiceAccount:      serviceAccount,
-		ConfigMapHash:       testConfigMapHash,
-	}
-
-	replicas := int32(1)
-	want := &v1.Deployment{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "apps/v1",
-			Kind:       "Deployments",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: testNS,
-			Name:      dispatcherName,
-		},
-		Spec: v1.DeploymentSpec{
-			Replicas: &replicas,
-			Selector: &metav1.LabelSelector{
-				MatchLabels: dispatcherLabels,
-			},
-			Template: corev1.PodTemplateSpec{
-				ObjectMeta: metav1.ObjectMeta{
-					Labels: dispatcherLabels,
-					Annotations: map[string]string{
-						commonconstants.ConfigMapHashAnnotationKey: testConfigMapHash,
-					},
-				},
-				Spec: corev1.PodSpec{
-					ServiceAccountName: serviceAccount,
-					Containers: []corev1.Container{
-						{
-							Name:  "dispatcher",
-							Image: imageName,
-							Env: []corev1.EnvVar{{
-								Name:  system.NamespaceEnvKey,
-								Value: "knative-testing",
-							}, {
-								Name:  "METRICS_DOMAIN",
-								Value: "knative.dev/eventing",
-							}, {
-								Name:  "CONFIG_LOGGING_NAME",
-								Value: "config-logging",
-							}, {
-								Name:  "CONFIG_LEADERELECTION_NAME",
-								Value: "config-leader-election",
-							}, {
-								Name: "NAMESPACE",
-								ValueFrom: &corev1.EnvVarSource{
-									FieldRef: &corev1.ObjectFieldSelector{
-										FieldPath: "metadata.namespace",
-									},
-								},
-							}, {
-								Name: "POD_NAME",
-								ValueFrom: &corev1.EnvVarSource{
-									FieldRef: &corev1.ObjectFieldSelector{
-										FieldPath: "metadata.name",
-									},
-								},
-							}, {
-								Name:  "CONTAINER_NAME",
-								Value: "dispatcher",
-							}},
-							Ports: []corev1.ContainerPort{{
-								Name:          "metrics",
-								ContainerPort: 9090,
-							}},
-							VolumeMounts: []corev1.VolumeMount{
-								{
-									Name:      "config-kafka",
-									MountPath: "/etc/config-kafka",
-								},
-							},
-						}},
-					Volumes: []corev1.Volume{
-						{
-							Name: "config-kafka",
-							VolumeSource: corev1.VolumeSource{
-								ConfigMap: &corev1.ConfigMapVolumeSource{
-									LocalObjectReference: corev1.LocalObjectReference{
-										Name: "config-kafka",
-									},
-								},
-							},
-						}},
-				},
-			},
-		},
-	}
-
-	got := MakeDispatcher(args)
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/role_binding_test.go b/pkg/channel/consolidated/reconciler/controller/resources/role_binding_test.go
deleted file mode 100644
index 792fe9b3c..000000000
--- a/pkg/channel/consolidated/reconciler/controller/resources/role_binding_test.go
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
-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 resources
-
-import (
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	rbacv1 "k8s.io/api/rbac/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-)
-
-const (
-	rbName = "my-test-role-binding"
-	crName = "my-test-cluster-role"
-)
-
-func TestNewRoleBinding(t *testing.T) {
-	want := &rbacv1.RoleBinding{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      rbName,
-			Namespace: testNS,
-		},
-		RoleRef: rbacv1.RoleRef{
-			APIGroup: "rbac.authorization.k8s.io",
-			Kind:     "ClusterRole",
-			Name:     crName,
-		},
-		Subjects: []rbacv1.Subject{
-			{
-				Kind:      "ServiceAccount",
-				Namespace: testNS,
-				Name:      serviceAccount,
-			},
-		},
-	}
-
-	sa := MakeServiceAccount(testNS, serviceAccount)
-	got := MakeRoleBinding(testNS, rbName, sa, crName)
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/service_account_test.go b/pkg/channel/consolidated/reconciler/controller/resources/service_account_test.go
deleted file mode 100644
index 84b4968fb..000000000
--- a/pkg/channel/consolidated/reconciler/controller/resources/service_account_test.go
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-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 resources
-
-import (
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-)
-
-func TestNewServiceAccount(t *testing.T) {
-	want := &corev1.ServiceAccount{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: testNS,
-			Name:      serviceAccount,
-		},
-	}
-
-	got := MakeServiceAccount(testNS, serviceAccount)
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/service_test.go b/pkg/channel/consolidated/reconciler/controller/resources/service_test.go
deleted file mode 100644
index 940329a16..000000000
--- a/pkg/channel/consolidated/reconciler/controller/resources/service_test.go
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
-Copyright 2019 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package resources
-
-import (
-	"errors"
-	"fmt"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	"knative.dev/pkg/kmeta"
-)
-
-const (
-	kcName             = "my-test-kc"
-	testNS             = "my-test-ns"
-	testDispatcherNS   = "dispatcher-namespace"
-	testDispatcherName = "dispatcher-name"
-	testConfigMapHash  = "deadbeef"
-)
-
-func TestMakeChannelServiceAddress(t *testing.T) {
-	if want, got := "my-test-kc-kn-channel", MakeChannelServiceName(kcName); want != got {
-		t.Errorf("Want: %q got %q", want, got)
-	}
-}
-
-func TestMakeService(t *testing.T) {
-	imc := &v1beta1.KafkaChannel{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      kcName,
-			Namespace: testNS,
-		},
-	}
-	want := &corev1.Service{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "v1",
-			Kind:       "Service",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%s-kn-channel", kcName),
-			Namespace: testNS,
-			Labels: map[string]string{
-				MessagingRoleLabel: MessagingRole,
-			},
-			OwnerReferences: []metav1.OwnerReference{
-				*kmeta.NewControllerRef(imc),
-			},
-		},
-		Spec: corev1.ServiceSpec{
-			Ports: []corev1.ServicePort{
-				{
-					Name:     portName,
-					Protocol: corev1.ProtocolTCP,
-					Port:     portNumber,
-				},
-			},
-		},
-	}
-
-	got, err := MakeK8sService(imc)
-	if err != nil {
-		t.Fatalf("Failed to create new service: %s", err)
-	}
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
-
-func TestMakeServiceWithExternal(t *testing.T) {
-	imc := &v1beta1.KafkaChannel{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      kcName,
-			Namespace: testNS,
-		},
-	}
-	want := &corev1.Service{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "v1",
-			Kind:       "Service",
-		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%s-kn-channel", kcName),
-			Namespace: testNS,
-			Labels: map[string]string{
-				MessagingRoleLabel: MessagingRole,
-			},
-			OwnerReferences: []metav1.OwnerReference{
-				*kmeta.NewControllerRef(imc),
-			},
-		},
-		Spec: corev1.ServiceSpec{
-			Type:         corev1.ServiceTypeExternalName,
-			ExternalName: "dispatcher-name.dispatcher-namespace.svc.cluster.local",
-		},
-	}
-
-	got, err := MakeK8sService(imc, ExternalService(testDispatcherNS, testDispatcherName))
-	if err != nil {
-		t.Fatalf("Failed to create new service: %s", err)
-	}
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("unexpected condition (-want, +got) = %v", diff)
-	}
-}
-
-func TestMakeServiceWithFailingOption(t *testing.T) {
-	imc := &v1beta1.KafkaChannel{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      kcName,
-			Namespace: testNS,
-		},
-	}
-	_, err := MakeK8sService(imc, func(svc *corev1.Service) error { return errors.New("test-induced failure") })
-	if err == nil {
-		t.Fatalf("Expcted error from new service but got none")
-	}
-}