Repository: knative-sandbox/eventing-kafka
Pull request #74 (2020-09-30 19:01:16)

Refactor contrib channel pkgs
This PR is another step in the eventing-contrib/kafka repo-consolidation saga...

The goal of the PR is to move the KafkaChannel implementation from the temporary contrib/ directory into it's intended location at pkg/channel/consolidated.  Some common and webhook code was moved as well.  The cmd/main.go and config YAML was also moved.  

> Note: While the config/ YAML was updated so that the ko:// references are correct, there are a bunch of labels of the form `contrib.eventing.knative.dev/release: devel` which were not changed.

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/74#" title="Author self-approved">travis-minke-sap</a>*

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

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

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

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file3) | Do not exist | 63.1% | 
[pkg/channel/consolidated/dispatcher/tracing.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file4) | Do not exist | 91.7% | 
[pkg/channel/consolidated/reconciler/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file6) | Do not exist | 54.2% | 
[pkg/channel/consolidated/reconciler/controller/resources/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file8) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file9) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/role_binding.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file10) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/service.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file11) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/service_account.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file12) | Do not exist | 100.0% | 
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file13) | Do not exist | 90.5% | 
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file14) | Do not exist | 83.9% | 
[pkg/common/consumer/consumer_handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311380756806766594/artifacts/line-cov.html#file15) | Do not exist | 89.5% | 

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

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

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

<details>

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file3) | Do not exist | 63.1% | 
[pkg/channel/consolidated/dispatcher/tracing.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file4) | Do not exist | 91.7% | 
[pkg/channel/consolidated/reconciler/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file5) | Do not exist | 0.0% | 
[pkg/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file6) | Do not exist | 54.2% | 
[pkg/channel/consolidated/reconciler/controller/resources/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file7) | Do not exist | 0.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file8) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file9) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/role_binding.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file10) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/service.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file11) | Do not exist | 100.0% | 
[pkg/channel/consolidated/reconciler/controller/resources/service_account.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file12) | Do not exist | 100.0% | 
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file13) | Do not exist | 90.5% | 
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file14) | Do not exist | 83.9% | 
[pkg/common/consumer/consumer_handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/74/pull-knative-sandbox-eventing-kafka-go-coverage/1311406928894627842/artifacts/line-cov.html#file15) | Do not exist | 89.5% | 

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

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

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

<details>

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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/contrib/kafka/channel/cmd/channel_controller/main.go b/cmd/channel/consolidated/controller/main.go
similarity index 90%
rename from contrib/kafka/channel/cmd/channel_controller/main.go
rename to cmd/channel/consolidated/controller/main.go
index 8926f3013..162d62d37 100644
--- a/contrib/kafka/channel/cmd/channel_controller/main.go
+++ b/cmd/channel/consolidated/controller/main.go
@@ -19,7 +19,7 @@ package main
 import (
 	"knative.dev/pkg/injection/sharedmain"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/reconciler/controller"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller"
 )
 
 const component = "kafkachannel-controller"
diff --git a/contrib/kafka/channel/cmd/channel_dispatcher/main.go b/cmd/channel/consolidated/dispatcher/main.go
similarity index 91%
rename from contrib/kafka/channel/cmd/channel_dispatcher/main.go
rename to cmd/channel/consolidated/dispatcher/main.go
index 7852b7c8c..476dd8d0e 100644
--- a/contrib/kafka/channel/cmd/channel_dispatcher/main.go
+++ b/cmd/channel/consolidated/dispatcher/main.go
@@ -23,7 +23,7 @@ import (
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/signals"
 
-	controller "knative.dev/eventing-kafka/contrib/kafka/channel/pkg/reconciler/dispatcher"
+	controller "knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/dispatcher"
 )
 
 const component = "kafkachannel-dispatcher"
diff --git a/cmd/webhook/README.md b/cmd/webhook/README.md
deleted file mode 100644
index 56dd8e6d2..000000000
--- a/cmd/webhook/README.md
+++ /dev/null
@@ -1,1 +0,0 @@
-# TODO - eventing-contrib/kafka/cmd/webhook content here
diff --git a/contrib/kafka/channel/cmd/webhook/main.go b/cmd/webhook/main.go
similarity index 100%
rename from contrib/kafka/channel/cmd/webhook/main.go
rename to cmd/webhook/main.go
diff --git a/contrib/kafka/channel/config/200-addressable-resolver-clusterrole.yaml b/config/channel/consolidated/200-addressable-resolver-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-addressable-resolver-clusterrole.yaml
rename to config/channel/consolidated/200-addressable-resolver-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/200-channelable-manipulator-clusterrole.yaml b/config/channel/consolidated/200-channelable-manipulator-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-channelable-manipulator-clusterrole.yaml
rename to config/channel/consolidated/200-channelable-manipulator-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/200-controller-clusterrole.yaml b/config/channel/consolidated/200-controller-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-controller-clusterrole.yaml
rename to config/channel/consolidated/200-controller-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/200-controller-serviceaccount.yaml b/config/channel/consolidated/200-controller-serviceaccount.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-controller-serviceaccount.yaml
rename to config/channel/consolidated/200-controller-serviceaccount.yaml
diff --git a/contrib/kafka/channel/config/200-dispatcher-clusterrole.yaml b/config/channel/consolidated/200-dispatcher-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-dispatcher-clusterrole.yaml
rename to config/channel/consolidated/200-dispatcher-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/200-dispatcher-serviceaccount.yaml b/config/channel/consolidated/200-dispatcher-serviceaccount.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-dispatcher-serviceaccount.yaml
rename to config/channel/consolidated/200-dispatcher-serviceaccount.yaml
diff --git a/contrib/kafka/channel/config/200-webhook-clusterrole.yaml b/config/channel/consolidated/200-webhook-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-webhook-clusterrole.yaml
rename to config/channel/consolidated/200-webhook-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/200-webhook-serviceaccount.yaml b/config/channel/consolidated/200-webhook-serviceaccount.yaml
similarity index 100%
rename from contrib/kafka/channel/config/200-webhook-serviceaccount.yaml
rename to config/channel/consolidated/200-webhook-serviceaccount.yaml
diff --git a/contrib/kafka/channel/config/201-clusterrolebinding.yaml b/config/channel/consolidated/201-clusterrolebinding.yaml
similarity index 100%
rename from contrib/kafka/channel/config/201-clusterrolebinding.yaml
rename to config/channel/consolidated/201-clusterrolebinding.yaml
diff --git a/contrib/kafka/channel/config/300-kafka-channel.yaml b/config/channel/consolidated/300-kafka-channel.yaml
similarity index 100%
rename from contrib/kafka/channel/config/300-kafka-channel.yaml
rename to config/channel/consolidated/300-kafka-channel.yaml
diff --git a/contrib/kafka/channel/config/400-kafka-config.yaml b/config/channel/consolidated/400-kafka-config.yaml
similarity index 100%
rename from contrib/kafka/channel/config/400-kafka-config.yaml
rename to config/channel/consolidated/400-kafka-config.yaml
diff --git a/contrib/kafka/channel/config/400-leader-election-config.yaml b/config/channel/consolidated/400-leader-election-config.yaml
similarity index 100%
rename from contrib/kafka/channel/config/400-leader-election-config.yaml
rename to config/channel/consolidated/400-leader-election-config.yaml
diff --git a/contrib/kafka/channel/config/400-webhook-service.yaml b/config/channel/consolidated/400-webhook-service.yaml
similarity index 100%
rename from contrib/kafka/channel/config/400-webhook-service.yaml
rename to config/channel/consolidated/400-webhook-service.yaml
diff --git a/contrib/kafka/channel/config/500-controller.yaml b/config/channel/consolidated/500-controller.yaml
similarity index 100%
rename from contrib/kafka/channel/config/500-controller.yaml
rename to config/channel/consolidated/500-controller.yaml
diff --git a/contrib/kafka/channel/config/500-dispatcher.yaml b/config/channel/consolidated/500-dispatcher.yaml
similarity index 100%
rename from contrib/kafka/channel/config/500-dispatcher.yaml
rename to config/channel/consolidated/500-dispatcher.yaml
diff --git a/contrib/kafka/channel/config/500-webhook-configuration.yaml b/config/channel/consolidated/500-webhook-configuration.yaml
similarity index 100%
rename from contrib/kafka/channel/config/500-webhook-configuration.yaml
rename to config/channel/consolidated/500-webhook-configuration.yaml
diff --git a/contrib/kafka/channel/config/500-webhook.yaml b/config/channel/consolidated/500-webhook.yaml
similarity index 100%
rename from contrib/kafka/channel/config/500-webhook.yaml
rename to config/channel/consolidated/500-webhook.yaml
diff --git a/config/channel/consolidated/README.md b/config/channel/consolidated/README.md
deleted file mode 100644
index 800755345..000000000
--- a/config/channel/consolidated/README.md
+++ /dev/null
@@ -1,1 +0,0 @@
-# TODO - eventing-contrib/kafka config content here
diff --git a/contrib/kafka/channel/config/configmaps/kafka-config.yaml b/config/channel/consolidated/configmaps/kafka-config.yaml
similarity index 100%
rename from contrib/kafka/channel/config/configmaps/kafka-config.yaml
rename to config/channel/consolidated/configmaps/kafka-config.yaml
diff --git a/contrib/kafka/channel/config/configmaps/leader-election.yaml b/config/channel/consolidated/configmaps/leader-election.yaml
similarity index 100%
rename from contrib/kafka/channel/config/configmaps/leader-election.yaml
rename to config/channel/consolidated/configmaps/leader-election.yaml
diff --git a/contrib/kafka/channel/config/deployments/controller.yaml b/config/channel/consolidated/deployments/controller.yaml
similarity index 90%
rename from contrib/kafka/channel/config/deployments/controller.yaml
rename to config/channel/consolidated/deployments/controller.yaml
index 0ade554d5..2430cdb29 100644
--- a/contrib/kafka/channel/config/deployments/controller.yaml
+++ b/config/channel/consolidated/deployments/controller.yaml
@@ -32,7 +32,7 @@ spec:
       serviceAccountName: kafka-ch-controller
       containers:
       - name: controller
-        image: ko://knative.dev/eventing-kafka/contrib/kafka/channel/cmd/channel_controller
+        image: ko://knative.dev/eventing-kafka/cmd/channel/consolidated/controller
         env:
         - name: CONFIG_LOGGING_NAME
           value: config-logging
@@ -45,7 +45,7 @@ spec:
         - name: CONFIG_LEADERELECTION_NAME
           value: config-leader-election-kafka
         - name: DISPATCHER_IMAGE
-          value: ko://knative.dev/eventing-kafka/contrib/kafka/channel/cmd/channel_dispatcher
+          value: ko://knative.dev/eventing-kafka/cmd/channel/consolidated/dispatcher
         ports:
         - containerPort: 9090
           name: metrics
diff --git a/contrib/kafka/channel/config/deployments/dispatcher.yaml b/config/channel/consolidated/deployments/dispatcher.yaml
similarity index 96%
rename from contrib/kafka/channel/config/deployments/dispatcher.yaml
rename to config/channel/consolidated/deployments/dispatcher.yaml
index 9015ec3f1..f5b907c19 100644
--- a/contrib/kafka/channel/config/deployments/dispatcher.yaml
+++ b/config/channel/consolidated/deployments/dispatcher.yaml
@@ -36,7 +36,7 @@ spec:
     spec:
       containers:
       - name: dispatcher
-        image: ko://knative.dev/eventing-kafka/contrib/kafka/channel/cmd/channel_dispatcher
+        image: ko://knative.dev/eventing-kafka/cmd/channel/consolidated/dispatcher
         env:
         - name: SYSTEM_NAMESPACE
           valueFrom:
diff --git a/contrib/kafka/channel/config/deployments/webhook-service.yaml b/config/channel/consolidated/deployments/webhook-service.yaml
similarity index 100%
rename from contrib/kafka/channel/config/deployments/webhook-service.yaml
rename to config/channel/consolidated/deployments/webhook-service.yaml
diff --git a/contrib/kafka/channel/config/deployments/webhook.yaml b/config/channel/consolidated/deployments/webhook.yaml
similarity index 96%
rename from contrib/kafka/channel/config/deployments/webhook.yaml
rename to config/channel/consolidated/deployments/webhook.yaml
index 14098f6d2..46b06882c 100644
--- a/contrib/kafka/channel/config/deployments/webhook.yaml
+++ b/config/channel/consolidated/deployments/webhook.yaml
@@ -35,7 +35,7 @@ spec:
       containers:
       - name: kafka-webhook
         terminationMessagePolicy: FallbackToLogsOnError
-        image: ko://knative.dev/eventing-kafka/contrib/kafka/channel/cmd/webhook
+        image: ko://knative.dev/eventing-kafka/cmd/webhook
         env:
         - name: SYSTEM_NAMESPACE
           valueFrom:
diff --git a/contrib/kafka/channel/config/dummy.go b/config/channel/consolidated/dummy.go
similarity index 86%
rename from contrib/kafka/channel/config/dummy.go
rename to config/channel/consolidated/dummy.go
index ec36cc370..58fd4ad95 100644
--- a/contrib/kafka/channel/config/dummy.go
+++ b/config/channel/consolidated/dummy.go
@@ -14,6 +14,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package config is a placeholder that allows us to pull in config files
+// This package is a placeholder that allows us to pull in config files
 // via go mod vendor.
-package config
+package consolidated
diff --git a/contrib/kafka/channel/config/resources/kafka-channel.yaml b/config/channel/consolidated/resources/kafka-channel.yaml
similarity index 100%
rename from contrib/kafka/channel/config/resources/kafka-channel.yaml
rename to config/channel/consolidated/resources/kafka-channel.yaml
diff --git a/contrib/kafka/channel/config/roles/addressable-resolver-clusterrole.yaml b/config/channel/consolidated/roles/addressable-resolver-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/roles/addressable-resolver-clusterrole.yaml
rename to config/channel/consolidated/roles/addressable-resolver-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/roles/channelable-manipulator-clusterrole.yaml b/config/channel/consolidated/roles/channelable-manipulator-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/roles/channelable-manipulator-clusterrole.yaml
rename to config/channel/consolidated/roles/channelable-manipulator-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/roles/controller-clusterrole.yaml b/config/channel/consolidated/roles/controller-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/roles/controller-clusterrole.yaml
rename to config/channel/consolidated/roles/controller-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/roles/dispatcher-clusterrole.yaml b/config/channel/consolidated/roles/dispatcher-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/roles/dispatcher-clusterrole.yaml
rename to config/channel/consolidated/roles/dispatcher-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/roles/webhook-clusterrole.yaml b/config/channel/consolidated/roles/webhook-clusterrole.yaml
similarity index 100%
rename from contrib/kafka/channel/config/roles/webhook-clusterrole.yaml
rename to config/channel/consolidated/roles/webhook-clusterrole.yaml
diff --git a/contrib/kafka/channel/config/webhooks/webhook-configuration.yaml b/config/channel/consolidated/webhooks/webhook-configuration.yaml
similarity index 100%
rename from contrib/kafka/channel/config/webhooks/webhook-configuration.yaml
rename to config/channel/consolidated/webhooks/webhook-configuration.yaml
diff --git a/contrib/kafka/channel/README.md b/contrib/kafka/channel/README.md
deleted file mode 100644
index 073c6954a..000000000
--- a/contrib/kafka/channel/README.md
+++ /dev/null
@@ -1,141 +0,0 @@
-# Apache Kafka Channels
-
-Kafka channels are those backed by [Apache Kafka](http://kafka.apache.org/)
-topics.
-
-## Deployment steps
-
-1. Setup [Knative Eventing](../../DEVELOPMENT.md)
-1. Install an Apache Kafka cluster, if you have not done so already.
-
-   For Kubernetes a simple installation is done using the
-   [Strimzi Kafka Operator](http://strimzi.io). Its installation
-   [guides](http://strimzi.io/quickstarts/) provide content for Kubernetes and
-   Openshift.
-
-   > Note: This `KafkaChannel` is not limited to Apache Kafka installations on
-   > Kubernetes. It is also possible to use an off-cluster Apache Kafka
-   > installation.
-
-1. Now that Apache Kafka is installed, you need to configure the
-   `bootstrapServers` value in the `config-kafka` ConfigMap, located inside the
-   `config/400-kafka-config.yaml` file.
-
-   ```yaml
-   apiVersion: v1
-   kind: ConfigMap
-   metadata:
-     name: config-kafka
-     namespace: knative-eventing
-   data:
-     # Broker URL. Replace this with the URLs for your kafka cluster,
-     # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
-     bootstrapServers: REPLACE_WITH_CLUSTER_URL
-   ```
-
-1. Apply the Kafka config:
-
-   ```sh
-   ko apply -f config
-   ```
-
-1. Create the `KafkaChannel` custom objects:
-
-   ```yaml
-   apiVersion: messaging.knative.dev/v1beta1
-   kind: KafkaChannel
-   metadata:
-     name: my-kafka-channel
-   spec:
-     numPartitions: 1
-     replicationFactor: 1
-   ```
-
-   You can configure the number of partitions with `numPartitions`, as well as
-   the replication factor with `replicationFactor`. If not set, both will
-   default to `1`.
-
-## Components
-
-The major components are:
-
-- Kafka Channel Controller
-- Kafka Channel Dispatcher
-- Kafka Webhook
-- Kafka Config Map
-
-The Kafka Channel Controller is located in one Pod:
-
-```shell
-kubectl get deployment -n knative-eventing kafka-ch-controller
-```
-
-The Kafka Channel Dispatcher receives and distributes all events to the
-appropriate consumers:
-
-```shell
-kubectl get deployment -n knative-eventing kafka-ch-dispatcher
-```
-
-The Kafka Webhook is used to validate and set defaults to `KafkaChannel` custom
-objects:
-
-```shell
-kubectl get deployment -n knative-eventing kafka-webhook
-```
-
-The Kafka Config Map is used to configure the `bootstrapServers` of your Apache
-Kafka installation:
-
-```shell
-kubectl get configmap -n knative-eventing config-kafka
-```
-
-### Namespace Dispatchers
-
-By default events are received and dispatched by a single cluster-scoped
-dispatcher components. You can also specify whether events should be received
-and dispatched by the dispatcher in the same namespace as the channel definition
-by adding the `eventing.knative.dev/scope: namespace` annotation.
-
-First, you need to create the configMap `config-kafka` in the same namespace as
-the KafkaChannel.
-
-```yaml
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: config-kafka
-  namespace: <YOUR_NAMESPACE>
-data:
-  # Broker URL. Replace this with the URLs for your kafka cluster,
-  # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
-  bootstrapServers: REPLACE_WITH_CLUSTER_URL
-```
-
-> Note: the `bootstrapServers` value does not have to be the same as the one
-> specified in `knative-eventing/config-kafka`.
-
-Then create a KafkaChannel:
-
-```yaml
-apiVersion: messaging.knative.dev/v1beta1
-kind: KafkaChannel
-metadata:
-  name: my-kafka-channel
-  namespace: <YOUR_NAMESPACE>
-  annotations:
-    eventing.knative.dev/scope: namespace
-spec:
-  numPartitions: 1
-  replicationFactor: 1
-```
-
-The dispatcher is created in `<YOUR_NAMESPACE>`:
-
-```sh
-kubectl get deployment -n <YOUR_NAMESPACE> kafka-ch-dispatcher
-```
-
-Both cluster-scoped and namespace-scoped dispatcher can coexist. However once
-the annotation is set (or not set), its value is immutable.
diff --git a/contrib/kafka/source/pkg/adapter/adapter.go b/contrib/kafka/source/pkg/adapter/adapter.go
index f102cace6..5fb6f2f3a 100644
--- a/contrib/kafka/source/pkg/adapter/adapter.go
+++ b/contrib/kafka/source/pkg/adapter/adapter.go
@@ -28,11 +28,10 @@ import (
 
 	"context"
 
-	kafkabinding "knative.dev/eventing-kafka/contrib/kafka"
-	"knative.dev/eventing-kafka/contrib/kafka/common/pkg/kafka"
-
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
+	kafkabinding "knative.dev/eventing-kafka/contrib/kafka"
+	"knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/pkg/logging"
 )
 
@@ -97,7 +96,7 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 	}
 	config.Consumer.Offsets.AutoCommit.Enable = false
 
-	consumerGroupFactory := kafka.NewConsumerGroupFactory(addrs, config)
+	consumerGroupFactory := consumer.NewConsumerGroupFactory(addrs, config)
 	group, err := consumerGroupFactory.StartConsumerGroup(a.config.ConsumerGroup, a.config.Topics, a.logger, a)
 	if err != nil {
 		panic(err)
diff --git a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go
index f1350e3f6..38559eef0 100644
--- a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go
+++ b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go
@@ -21,7 +21,7 @@ import (
 
 	"knative.dev/eventing/pkg/apis/messaging"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 )
 
 func (c *KafkaChannel) SetDefaults(ctx context.Context) {
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go b/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
index 43c564195..7438adb08 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
@@ -21,7 +21,7 @@ import (
 
 	"knative.dev/eventing/pkg/apis/messaging"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 )
 
 func (c *KafkaChannel) SetDefaults(ctx context.Context) {
diff --git a/pkg/channel/consolidated/README.md b/pkg/channel/consolidated/README.md
index 9262d82c7..6e4626c56 100644
--- a/pkg/channel/consolidated/README.md
+++ b/pkg/channel/consolidated/README.md
@@ -1,1 +1,141 @@
-TODO - eventing-contrib/kafka channel here !
+# Apache Kafka Channels
+
+Kafka channels are those backed by [Apache Kafka](http://kafka.apache.org/)
+topics.
+
+## Deployment steps
+
+1. Setup [Knative Eventing](../../../contrib/DEVELOPMENT.md)
+1. Install an Apache Kafka cluster, if you have not done so already.
+
+   For Kubernetes a simple installation is done using the
+   [Strimzi Kafka Operator](http://strimzi.io). Its installation
+   [guides](http://strimzi.io/quickstarts/) provide content for Kubernetes and
+   Openshift.
+
+   > Note: This `KafkaChannel` is not limited to Apache Kafka installations on
+   > Kubernetes. It is also possible to use an off-cluster Apache Kafka
+   > installation.
+
+1. Now that Apache Kafka is installed, you need to configure the
+   `bootstrapServers` value in the `config-kafka` ConfigMap, located inside the
+   `config/400-kafka-config.yaml` file.
+
+   ```yaml
+   apiVersion: v1
+   kind: ConfigMap
+   metadata:
+     name: config-kafka
+     namespace: knative-eventing
+   data:
+     # Broker URL. Replace this with the URLs for your kafka cluster,
+     # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
+     bootstrapServers: REPLACE_WITH_CLUSTER_URL
+   ```
+
+1. Apply the Kafka config:
+
+   ```sh
+   ko apply -f config
+   ```
+
+1. Create the `KafkaChannel` custom objects:
+
+   ```yaml
+   apiVersion: messaging.knative.dev/v1beta1
+   kind: KafkaChannel
+   metadata:
+     name: my-kafka-channel
+   spec:
+     numPartitions: 1
+     replicationFactor: 1
+   ```
+
+   You can configure the number of partitions with `numPartitions`, as well as
+   the replication factor with `replicationFactor`. If not set, both will
+   default to `1`.
+
+## Components
+
+The major components are:
+
+- Kafka Channel Controller
+- Kafka Channel Dispatcher
+- Kafka Webhook
+- Kafka Config Map
+
+The Kafka Channel Controller is located in one Pod:
+
+```shell
+kubectl get deployment -n knative-eventing kafka-ch-controller
+```
+
+The Kafka Channel Dispatcher receives and distributes all events to the
+appropriate consumers:
+
+```shell
+kubectl get deployment -n knative-eventing kafka-ch-dispatcher
+```
+
+The Kafka Webhook is used to validate and set defaults to `KafkaChannel` custom
+objects:
+
+```shell
+kubectl get deployment -n knative-eventing kafka-webhook
+```
+
+The Kafka Config Map is used to configure the `bootstrapServers` of your Apache
+Kafka installation:
+
+```shell
+kubectl get configmap -n knative-eventing config-kafka
+```
+
+### Namespace Dispatchers
+
+By default events are received and dispatched by a single cluster-scoped
+dispatcher components. You can also specify whether events should be received
+and dispatched by the dispatcher in the same namespace as the channel definition
+by adding the `eventing.knative.dev/scope: namespace` annotation.
+
+First, you need to create the configMap `config-kafka` in the same namespace as
+the KafkaChannel.
+
+```yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka
+  namespace: <YOUR_NAMESPACE>
+data:
+  # Broker URL. Replace this with the URLs for your kafka cluster,
+  # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
+  bootstrapServers: REPLACE_WITH_CLUSTER_URL
+```
+
+> Note: the `bootstrapServers` value does not have to be the same as the one
+> specified in `knative-eventing/config-kafka`.
+
+Then create a KafkaChannel:
+
+```yaml
+apiVersion: messaging.knative.dev/v1beta1
+kind: KafkaChannel
+metadata:
+  name: my-kafka-channel
+  namespace: <YOUR_NAMESPACE>
+  annotations:
+    eventing.knative.dev/scope: namespace
+spec:
+  numPartitions: 1
+  replicationFactor: 1
+```
+
+The dispatcher is created in `<YOUR_NAMESPACE>`:
+
+```sh
+kubectl get deployment -n <YOUR_NAMESPACE> kafka-ch-dispatcher
+```
+
+Both cluster-scoped and namespace-scoped dispatcher can coexist. However once
+the annotation is set (or not set), its value is immutable.
diff --git a/contrib/kafka/channel/pkg/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
similarity index 97%
rename from contrib/kafka/channel/pkg/dispatcher/dispatcher.go
rename to pkg/channel/consolidated/dispatcher/dispatcher.go
index 2caad7756..c3c9b2940 100644
--- a/contrib/kafka/channel/pkg/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -30,12 +30,11 @@ import (
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+	"knative.dev/eventing-kafka/pkg/common/consumer"
 	eventingchannels "knative.dev/eventing/pkg/channel"
 	"knative.dev/eventing/pkg/channel/fanout"
 	"knative.dev/eventing/pkg/kncloudevents"
-
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
-	"knative.dev/eventing-kafka/contrib/kafka/common/pkg/kafka"
 )
 
 type KafkaDispatcher struct {
@@ -52,7 +51,7 @@ type KafkaDispatcher struct {
 	subscriptions        map[types.UID]Subscription
 	// consumerUpdateLock must be used to update kafkaConsumers
 	consumerUpdateLock   sync.Mutex
-	kafkaConsumerFactory kafka.KafkaConsumerGroupFactory
+	kafkaConsumerFactory consumer.KafkaConsumerGroupFactory
 
 	topicFunc TopicFunc
 	logger    *zap.SugaredLogger
@@ -95,7 +94,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 
 	dispatcher := &KafkaDispatcher{
 		dispatcher:           eventingchannels.NewMessageDispatcher(args.Logger.Desugar()),
-		kafkaConsumerFactory: kafka.NewConsumerGroupFactory(args.Brokers, conf),
+		kafkaConsumerFactory: consumer.NewConsumerGroupFactory(args.Brokers, conf),
 		channelSubscriptions: make(map[eventingchannels.ChannelReference][]types.UID),
 		subsConsumerGroups:   make(map[types.UID]sarama.ConsumerGroup),
 		subscriptions:        make(map[types.UID]Subscription),
@@ -183,7 +182,7 @@ func (c consumerMessageHandler) Handle(ctx context.Context, consumerMessage *sar
 	return err == nil, err
 }
 
-var _ kafka.KafkaConsumerHandler = (*consumerMessageHandler)(nil)
+var _ consumer.KafkaConsumerHandler = (*consumerMessageHandler)(nil)
 
 type Config struct {
 	// The configuration of each channel in this handler.
diff --git a/contrib/kafka/channel/pkg/dispatcher/tracing.go b/pkg/channel/consolidated/dispatcher/tracing.go
similarity index 100%
rename from contrib/kafka/channel/pkg/dispatcher/tracing.go
rename to pkg/channel/consolidated/dispatcher/tracing.go
diff --git a/contrib/kafka/channel/pkg/dispatcher/tracing_test.go b/pkg/channel/consolidated/dispatcher/tracing_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/dispatcher/tracing_test.go
rename to pkg/channel/consolidated/dispatcher/tracing_test.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/controller.go b/pkg/channel/consolidated/reconciler/controller/controller.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/controller.go
rename to pkg/channel/consolidated/reconciler/controller/controller.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
similarity index 99%
rename from contrib/kafka/channel/pkg/reconciler/controller/kafkachannel.go
rename to pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index 1beacb9dc..52a59af76 100644
--- a/contrib/kafka/channel/pkg/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -49,9 +49,9 @@ import (
 	"knative.dev/pkg/controller"
 	pkgreconciler "knative.dev/pkg/reconciler"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/reconciler/controller/resources"
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
 	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	kafkaScheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
 	kafkaChannelReconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/client.go b/pkg/channel/consolidated/reconciler/controller/resources/client.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/client.go
rename to pkg/channel/consolidated/reconciler/controller/resources/client.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher.go
rename to pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher_service.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher_service.go
rename to pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher_service_test.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher_service_test.go
rename to pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service_test.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher_test.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/dispatcher_test.go
rename to pkg/channel/consolidated/reconciler/controller/resources/dispatcher_test.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/role_binding.go b/pkg/channel/consolidated/reconciler/controller/resources/role_binding.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/role_binding.go
rename to pkg/channel/consolidated/reconciler/controller/resources/role_binding.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/role_binding_test.go b/pkg/channel/consolidated/reconciler/controller/resources/role_binding_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/role_binding_test.go
rename to pkg/channel/consolidated/reconciler/controller/resources/role_binding_test.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/service.go b/pkg/channel/consolidated/reconciler/controller/resources/service.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/service.go
rename to pkg/channel/consolidated/reconciler/controller/resources/service.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/service_account.go b/pkg/channel/consolidated/reconciler/controller/resources/service_account.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/service_account.go
rename to pkg/channel/consolidated/reconciler/controller/resources/service_account.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/service_account_test.go b/pkg/channel/consolidated/reconciler/controller/resources/service_account_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/service_account_test.go
rename to pkg/channel/consolidated/reconciler/controller/resources/service_account_test.go
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/resources/service_test.go b/pkg/channel/consolidated/reconciler/controller/resources/service_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/controller/resources/service_test.go
rename to pkg/channel/consolidated/reconciler/controller/resources/service_test.go
diff --git a/contrib/kafka/channel/pkg/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
similarity index 98%
rename from contrib/kafka/channel/pkg/reconciler/dispatcher/kafkachannel.go
rename to pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index 0e5c36fef..89b90fefe 100644
--- a/contrib/kafka/channel/pkg/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -38,9 +38,9 @@ import (
 	"knative.dev/pkg/logging"
 	pkgreconciler "knative.dev/pkg/reconciler"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/dispatcher"
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
 	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/dispatcher"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	kafkaScheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
 	kafkaclientsetinjection "knative.dev/eventing-kafka/pkg/client/injection/client"
diff --git a/contrib/kafka/channel/pkg/reconciler/stats_reporter.go b/pkg/channel/consolidated/reconciler/stats_reporter.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/stats_reporter.go
rename to pkg/channel/consolidated/reconciler/stats_reporter.go
diff --git a/contrib/kafka/channel/pkg/reconciler/testing/factory.go b/pkg/channel/consolidated/reconciler/testing/factory.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/testing/factory.go
rename to pkg/channel/consolidated/reconciler/testing/factory.go
diff --git a/contrib/kafka/channel/pkg/reconciler/testing/kafkachannel.go b/pkg/channel/consolidated/reconciler/testing/kafkachannel.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/testing/kafkachannel.go
rename to pkg/channel/consolidated/reconciler/testing/kafkachannel.go
diff --git a/contrib/kafka/channel/pkg/reconciler/testing/listers.go b/pkg/channel/consolidated/reconciler/testing/listers.go
similarity index 100%
rename from contrib/kafka/channel/pkg/reconciler/testing/listers.go
rename to pkg/channel/consolidated/reconciler/testing/listers.go
diff --git a/contrib/kafka/channel/pkg/utils/util.go b/pkg/channel/consolidated/utils/util.go
similarity index 100%
rename from contrib/kafka/channel/pkg/utils/util.go
rename to pkg/channel/consolidated/utils/util.go
diff --git a/contrib/kafka/channel/pkg/utils/util_test.go b/pkg/channel/consolidated/utils/util_test.go
similarity index 100%
rename from contrib/kafka/channel/pkg/utils/util_test.go
rename to pkg/channel/consolidated/utils/util_test.go
diff --git a/contrib/kafka/common/pkg/kafka/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
similarity index 99%
rename from contrib/kafka/common/pkg/kafka/consumer_factory.go
rename to pkg/common/consumer/consumer_factory.go
index 9aa63dcae..ccc5dd412 100644
--- a/contrib/kafka/common/pkg/kafka/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -13,7 +13,7 @@ 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
+package consumer
 
 import (
 	"context"
diff --git a/contrib/kafka/common/pkg/kafka/consumer_handler.go b/pkg/common/consumer/consumer_handler.go
similarity index 99%
rename from contrib/kafka/common/pkg/kafka/consumer_handler.go
rename to pkg/common/consumer/consumer_handler.go
index 321fd0f7f..5e1eea1a1 100644
--- a/contrib/kafka/common/pkg/kafka/consumer_handler.go
+++ b/pkg/common/consumer/consumer_handler.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kafka
+package consumer
 
 import (
 	"context"


Test patch:
diff --git a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go
index e55896f7a..c48693f3d 100644
--- a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go
+++ b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go
@@ -19,7 +19,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 
 	"github.com/google/go-cmp/cmp"
 )
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go b/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
index 557b607a2..cac1ca396 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
@@ -19,7 +19,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 
 	"github.com/google/go-cmp/cmp"
 )
diff --git a/contrib/kafka/channel/pkg/dispatcher/dispatcher_it_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
similarity index 99%
rename from contrib/kafka/channel/pkg/dispatcher/dispatcher_it_test.go
rename to pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
index 002121bd3..48c7c64b4 100644
--- a/contrib/kafka/channel/pkg/dispatcher/dispatcher_it_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
@@ -38,7 +38,7 @@ import (
 	"knative.dev/pkg/apis"
 	tracingconfig "knative.dev/pkg/tracing/config"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 )
 
 // This dispatcher tests the full integration of the dispatcher code with Kafka.
diff --git a/contrib/kafka/channel/pkg/dispatcher/dispatcher_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
similarity index 97%
rename from contrib/kafka/channel/pkg/dispatcher/dispatcher_test.go
rename to pkg/channel/consolidated/dispatcher/dispatcher_test.go
index f1d06609f..3410efd8a 100644
--- a/contrib/kafka/channel/pkg/dispatcher/dispatcher_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_test.go
@@ -23,8 +23,6 @@ import (
 	"net/url"
 	"testing"
 
-	"knative.dev/eventing/pkg/channel/fanout"
-
 	"github.com/Shopify/sarama"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"github.com/google/go-cmp/cmp"
@@ -33,12 +31,11 @@ import (
 	"go.uber.org/zap/zaptest"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
-
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+	"knative.dev/eventing-kafka/pkg/common/consumer"
 	eventingchannels "knative.dev/eventing/pkg/channel"
+	"knative.dev/eventing/pkg/channel/fanout"
 	_ "knative.dev/pkg/system/testing"
-
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
-	"knative.dev/eventing-kafka/contrib/kafka/common/pkg/kafka"
 )
 
 // ----- Mocks
@@ -48,7 +45,7 @@ type mockKafkaConsumerFactory struct {
 	createErr bool
 }
 
-func (c mockKafkaConsumerFactory) StartConsumerGroup(groupID string, topics []string, logger *zap.SugaredLogger, handler kafka.KafkaConsumerHandler) (sarama.ConsumerGroup, error) {
+func (c mockKafkaConsumerFactory) StartConsumerGroup(groupID string, topics []string, logger *zap.SugaredLogger, handler consumer.KafkaConsumerHandler) (sarama.ConsumerGroup, error) {
 	if c.createErr {
 		return nil, errors.New("error creating consumer")
 	}
@@ -56,7 +53,7 @@ func (c mockKafkaConsumerFactory) StartConsumerGroup(groupID string, topics []st
 	return mockConsumerGroup{}, nil
 }
 
-var _ kafka.KafkaConsumerGroupFactory = (*mockKafkaConsumerFactory)(nil)
+var _ consumer.KafkaConsumerGroupFactory = (*mockKafkaConsumerFactory)(nil)
 
 type mockConsumerGroup struct{}
 
diff --git a/contrib/kafka/channel/pkg/reconciler/controller/kafkachannel_test.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
similarity index 66%
rename from contrib/kafka/channel/pkg/reconciler/controller/kafkachannel_test.go
rename to pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
index 9671f4c4c..4f8d5614c 100644
--- a/contrib/kafka/channel/pkg/reconciler/controller/kafkachannel_test.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel_test.go
@@ -43,11 +43,10 @@ import (
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
 
-	"knative.dev/eventing-kafka/contrib/kafka/channel/pkg/reconciler/controller/resources"
-	reconcilekafkatesting "knative.dev/eventing-kafka/contrib/kafka/channel/pkg/reconciler/testing"
-	reconcilertesting "knative.dev/eventing-kafka/contrib/kafka/channel/pkg/reconciler/testing"
-	. "knative.dev/eventing-kafka/contrib/kafka/channel/pkg/utils"
 	"knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
+	reconcilertesting "knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/testing"
+	. "knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
 	fakekafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
 )
@@ -86,9 +85,9 @@ func TestAllCases(t *testing.T) {
 			Name: "deleting",
 			Key:  kcKey,
 			Objects: []runtime.Object{
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaChannelDeleted)},
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaChannelDeleted)},
 			WantErr: false,
 			WantEvents: []string{
 				Eventf(corev1.EventTypeNormal, "KafkaChannelReconciled", `KafkaChannel reconciled: "test-namespace/test-kc"`),
@@ -97,9 +96,9 @@ func TestAllCases(t *testing.T) {
 			Name: "deployment does not exist, automatically created and patching finalizers",
 			Key:  kcKey,
 			Objects: []runtime.Object{
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaChannelTopicReady()),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaChannelTopicReady()),
 			},
 			WantErr: true,
 			WantCreates: []runtime.Object{
@@ -107,12 +106,12 @@ func TestAllCases(t *testing.T) {
 				makeService(),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsDoesNotExist", "Dispatcher Endpoints does not exist")),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsDoesNotExist", "Dispatcher Endpoints does not exist")),
 			}},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(testNS, kcName),
@@ -128,22 +127,22 @@ func TestAllCases(t *testing.T) {
 			Key:  kcKey,
 			Objects: []runtime.Object{
 				makeReadyDeployment(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 			},
 			WantErr: true,
 			WantCreates: []runtime.Object{
 				makeService(),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsDoesNotExist", "Dispatcher Endpoints does not exist")),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsDoesNotExist", "Dispatcher Endpoints does not exist")),
 			}},
 			WantEvents: []string{
 				Eventf(corev1.EventTypeNormal, dispatcherServiceCreated, "Dispatcher service created"),
@@ -155,19 +154,19 @@ func TestAllCases(t *testing.T) {
 			Objects: []runtime.Object{
 				makeReadyDeployment(),
 				makeService(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 			},
 			WantErr: true,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsDoesNotExist", "Dispatcher Endpoints does not exist"),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsDoesNotExist", "Dispatcher Endpoints does not exist"),
 				),
 			}},
 			WantEvents: []string{
@@ -180,19 +179,19 @@ func TestAllCases(t *testing.T) {
 				makeReadyDeployment(),
 				makeService(),
 				makeEmptyEndpoints(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 			},
 			WantErr: true,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsNotReady", "There are no endpoints ready for Dispatcher service"),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsNotReady("DispatcherEndpointsNotReady", "There are no endpoints ready for Dispatcher service"),
 				),
 			}},
 			WantEvents: []string{
@@ -205,24 +204,24 @@ func TestAllCases(t *testing.T) {
 				makeReadyDeployment(),
 				makeService(),
 				makeReadyEndpoints(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 			},
 			WantErr: false,
 			WantCreates: []runtime.Object{
-				makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+				makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-					reconcilekafkatesting.WithKafkaChannelChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelAddress(channelServiceAddress),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsReady(),
+					reconcilertesting.WithKafkaChannelChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
 				),
 			}},
 			WantEvents: []string{
@@ -235,22 +234,22 @@ func TestAllCases(t *testing.T) {
 				makeReadyDeployment(),
 				makeService(),
 				makeReadyEndpoints(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
-				makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
+				makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 			},
 			WantErr: false,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-					reconcilekafkatesting.WithKafkaChannelChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelAddress(channelServiceAddress),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsReady(),
+					reconcilertesting.WithKafkaChannelChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
 				),
 			}},
 			WantEvents: []string{
@@ -263,21 +262,21 @@ func TestAllCases(t *testing.T) {
 				makeReadyDeployment(),
 				makeService(),
 				makeReadyEndpoints(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 				makeChannelServiceNotOwnedByUs(),
 			},
 			WantErr: true,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-					reconcilekafkatesting.WithKafkaChannelChannelServicetNotReady("ChannelServiceFailed", "Channel Service failed: kafkachannel: test-namespace/test-kc does not own Service: \"test-kc-kn-channel\""),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsReady(),
+					reconcilertesting.WithKafkaChannelChannelServicetNotReady("ChannelServiceFailed", "Channel Service failed: kafkachannel: test-namespace/test-kc does not own Service: \"test-kc-kn-channel\""),
 				),
 			}},
 			WantEvents: []string{
@@ -290,27 +289,27 @@ func TestAllCases(t *testing.T) {
 				makeReadyDeployment(),
 				makeService(),
 				makeReadyEndpoints(),
-				reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+				reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithKafkaFinalizer(finalizerName)),
 			},
 			WantErr: true,
 			WithReactors: []clientgotesting.ReactionFunc{
 				InduceFailure("create", "Services"),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-				Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-					reconcilekafkatesting.WithInitKafkaChannelConditions,
-					reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-					reconcilekafkatesting.WithKafkaChannelConfigReady(),
-					reconcilekafkatesting.WithKafkaChannelTopicReady(),
-					reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-					reconcilekafkatesting.WithKafkaChannelServiceReady(),
-					reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-					reconcilekafkatesting.WithKafkaChannelChannelServicetNotReady("ChannelServiceFailed", "Channel Service failed: inducing failure for create services"),
+				Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+					reconcilertesting.WithInitKafkaChannelConditions,
+					reconcilertesting.WithKafkaFinalizer(finalizerName),
+					reconcilertesting.WithKafkaChannelConfigReady(),
+					reconcilertesting.WithKafkaChannelTopicReady(),
+					reconcilertesting.WithKafkaChannelDeploymentReady(),
+					reconcilertesting.WithKafkaChannelServiceReady(),
+					reconcilertesting.WithKafkaChannelEndpointsReady(),
+					reconcilertesting.WithKafkaChannelChannelServicetNotReady("ChannelServiceFailed", "Channel Service failed: inducing failure for create services"),
 				),
 			}},
 			WantCreates: []runtime.Object{
-				makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+				makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 			},
 			WantEvents: []string{
 				Eventf(corev1.EventTypeWarning, "InternalError", "inducing failure for create services"),
@@ -319,7 +318,7 @@ func TestAllCases(t *testing.T) {
 		},
 	}
 
-	table.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilekafkatesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	table.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
 
 		r := &Reconciler{
 			systemNamespace: testNS,
@@ -351,24 +350,24 @@ func TestTopicExists(t *testing.T) {
 			makeReadyDeployment(),
 			makeService(),
 			makeReadyEndpoints(),
-			reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+			reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithKafkaFinalizer(finalizerName)),
 		},
 		WantErr: false,
 		WantCreates: []runtime.Object{
-			makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+			makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithInitKafkaChannelConditions,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-				reconcilekafkatesting.WithKafkaChannelConfigReady(),
-				reconcilekafkatesting.WithKafkaChannelTopicReady(),
-				reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-				reconcilekafkatesting.WithKafkaChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-				reconcilekafkatesting.WithKafkaChannelChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelAddress(channelServiceAddress),
+			Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithInitKafkaChannelConditions,
+				reconcilertesting.WithKafkaFinalizer(finalizerName),
+				reconcilertesting.WithKafkaChannelConfigReady(),
+				reconcilertesting.WithKafkaChannelTopicReady(),
+				reconcilertesting.WithKafkaChannelDeploymentReady(),
+				reconcilertesting.WithKafkaChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelEndpointsReady(),
+				reconcilertesting.WithKafkaChannelChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
 			),
 		}},
 		WantEvents: []string{
@@ -376,7 +375,7 @@ func TestTopicExists(t *testing.T) {
 		},
 	}
 
-	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilekafkatesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
 
 		r := &Reconciler{
 			systemNamespace: testNS,
@@ -416,27 +415,27 @@ func TestDeploymentUpdatedOnImageChange(t *testing.T) {
 			makeDeploymentWithImageAndReplicas("differentimage", 1),
 			makeService(),
 			makeReadyEndpoints(),
-			reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+			reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithKafkaFinalizer(finalizerName)),
 		},
 		WantErr: false,
 		WantCreates: []runtime.Object{
-			makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+			makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 		},
 		WantUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: makeDeployment(),
 		}},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithInitKafkaChannelConditions,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-				reconcilekafkatesting.WithKafkaChannelConfigReady(),
-				reconcilekafkatesting.WithKafkaChannelTopicReady(),
+			Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithInitKafkaChannelConditions,
+				reconcilertesting.WithKafkaFinalizer(finalizerName),
+				reconcilertesting.WithKafkaChannelConfigReady(),
+				reconcilertesting.WithKafkaChannelTopicReady(),
 				//				reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-				reconcilekafkatesting.WithKafkaChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-				reconcilekafkatesting.WithKafkaChannelChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelAddress(channelServiceAddress),
+				reconcilertesting.WithKafkaChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelEndpointsReady(),
+				reconcilertesting.WithKafkaChannelChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
 			),
 		}},
 		WantEvents: []string{
@@ -445,7 +444,7 @@ func TestDeploymentUpdatedOnImageChange(t *testing.T) {
 		},
 	}
 
-	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilekafkatesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
 
 		r := &Reconciler{
 			systemNamespace: testNS,
@@ -485,27 +484,27 @@ func TestDeploymentZeroReplicas(t *testing.T) {
 			makeDeploymentWithImageAndReplicas(testDispatcherImage, 0),
 			makeService(),
 			makeReadyEndpoints(),
-			reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+			reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithKafkaFinalizer(finalizerName)),
 		},
 		WantErr: false,
 		WantCreates: []runtime.Object{
-			makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+			makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 		},
 		WantUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: makeDeployment(),
 		}},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithInitKafkaChannelConditions,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-				reconcilekafkatesting.WithKafkaChannelConfigReady(),
-				reconcilekafkatesting.WithKafkaChannelTopicReady(),
+			Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithInitKafkaChannelConditions,
+				reconcilertesting.WithKafkaFinalizer(finalizerName),
+				reconcilertesting.WithKafkaChannelConfigReady(),
+				reconcilertesting.WithKafkaChannelTopicReady(),
 				//				reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-				reconcilekafkatesting.WithKafkaChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-				reconcilekafkatesting.WithKafkaChannelChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelAddress(channelServiceAddress),
+				reconcilertesting.WithKafkaChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelEndpointsReady(),
+				reconcilertesting.WithKafkaChannelChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
 			),
 		}},
 		WantEvents: []string{
@@ -514,7 +513,7 @@ func TestDeploymentZeroReplicas(t *testing.T) {
 		},
 	}
 
-	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilekafkatesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
 
 		r := &Reconciler{
 			systemNamespace: testNS,
@@ -554,25 +553,25 @@ func TestDeploymentMoreThanOneReplicas(t *testing.T) {
 			makeDeploymentWithImageAndReplicas(testDispatcherImage, 3),
 			makeService(),
 			makeReadyEndpoints(),
-			reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName)),
+			reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithKafkaFinalizer(finalizerName)),
 		},
 		WantErr: false,
 		WantCreates: []runtime.Object{
-			makeChannelService(reconcilekafkatesting.NewKafkaChannel(kcName, testNS)),
+			makeChannelService(reconcilertesting.NewKafkaChannel(kcName, testNS)),
 		},
 		WantUpdates: []clientgotesting.UpdateActionImpl{},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: reconcilekafkatesting.NewKafkaChannel(kcName, testNS,
-				reconcilekafkatesting.WithInitKafkaChannelConditions,
-				reconcilekafkatesting.WithKafkaFinalizer(finalizerName),
-				reconcilekafkatesting.WithKafkaChannelConfigReady(),
-				reconcilekafkatesting.WithKafkaChannelTopicReady(),
+			Object: reconcilertesting.NewKafkaChannel(kcName, testNS,
+				reconcilertesting.WithInitKafkaChannelConditions,
+				reconcilertesting.WithKafkaFinalizer(finalizerName),
+				reconcilertesting.WithKafkaChannelConfigReady(),
+				reconcilertesting.WithKafkaChannelTopicReady(),
 				//				reconcilekafkatesting.WithKafkaChannelDeploymentReady(),
-				reconcilekafkatesting.WithKafkaChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelEndpointsReady(),
-				reconcilekafkatesting.WithKafkaChannelChannelServiceReady(),
-				reconcilekafkatesting.WithKafkaChannelAddress(channelServiceAddress),
+				reconcilertesting.WithKafkaChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelEndpointsReady(),
+				reconcilertesting.WithKafkaChannelChannelServiceReady(),
+				reconcilertesting.WithKafkaChannelAddress(channelServiceAddress),
 			),
 		}},
 		WantEvents: []string{
@@ -580,7 +579,7 @@ func TestDeploymentMoreThanOneReplicas(t *testing.T) {
 		},
 	}
 
-	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilekafkatesting.Listers, cmw configmap.Watcher) controller.Reconciler {
+	row.Test(t, reconcilertesting.MakeFactory(func(ctx context.Context, listers *reconcilertesting.Listers, cmw configmap.Watcher) controller.Reconciler {
 
 		r := &Reconciler{
 			systemNamespace: testNS,
diff --git a/contrib/kafka/common/pkg/kafka/consumer_factory_test.go b/pkg/common/consumer/consumer_factory_test.go
similarity index 99%
rename from contrib/kafka/common/pkg/kafka/consumer_factory_test.go
rename to pkg/common/consumer/consumer_factory_test.go
index cb9407a91..e975a5cf9 100644
--- a/contrib/kafka/common/pkg/kafka/consumer_factory_test.go
+++ b/pkg/common/consumer/consumer_factory_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kafka
+package consumer
 
 import (
 	"context"
diff --git a/contrib/kafka/common/pkg/kafka/consumer_handler_test.go b/pkg/common/consumer/consumer_handler_test.go
similarity index 99%
rename from contrib/kafka/common/pkg/kafka/consumer_handler_test.go
rename to pkg/common/consumer/consumer_handler_test.go
index 1c3594250..a755bfb2a 100644
--- a/contrib/kafka/common/pkg/kafka/consumer_handler_test.go
+++ b/pkg/common/consumer/consumer_handler_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kafka
+package consumer
 
 import (
 	"context"