Repository: knative-sandbox/eventing-kafka-broker
Pull request #2039 (2022-03-29 20:02:04)

[data-plane]: Bump jackson-bom from 2.13.2.20220324 to 2.13.2.20220328 in /data-plane
Bumps [jackson-bom](https://github.com/FasterXML/jackson-bom) from 2.13.2.20220324 to 2.13.2.20220328.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/4d5d57bd57588fbe4a1c155330fa9ed523391cd4"><code>4d5d57b</code></a> [maven-release-plugin] prepare release jackson-bom-2.13.2.20220328</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/6986ea5f06d2e45ec865e470ee3d9e90ce955717"><code>6986ea5</code></a> Prepare for 2.13.2.20220328 release</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/74d0f2589863648cc4400819afc43ae62d7fe975"><code>74d0f25</code></a> Merge branch '2.12' into 2.13</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/a3e55198187d190972d336344d8c866c1991cff3"><code>a3e5519</code></a> [maven-release-plugin] prepare for next development iteration</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/60d4eb2652bfbca32cc1712db3113b70140e8f02"><code>60d4eb2</code></a> [maven-release-plugin] prepare release jackson-bom-2.12.6.20220326</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/56ce293c601207b5691436f18bd6ecc45f64f6b5"><code>56ce293</code></a> Prepare for 2.12.6.20220326 release</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/3d90426654a7c815726578d49305f7edc38a1ce6"><code>3d90426</code></a> Back to 2.13.3-SNAPSHOT</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/4cac0e40087e01d9b4e3bf685a449ee5700527bc"><code>4cac0e4</code></a> [maven-release-plugin] prepare for next development iteration</li>
<li><a href="https://github.com/FasterXML/jackson-bom/commit/a2709d0f66b04838bb3643ac9e6f7b4717649e93"><code>a2709d0</code></a> ...</li>
<li>See full diff in <a href="https://github.com/FasterXML/jackson-bom/compare/jackson-bom-2.13.2.20220324...jackson-bom-2.13.2.20220328">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=com.fasterxml.jackson:jackson-bom&package-manager=maven&previous-version=2.13.2.20220324&new-version=2.13.2.20220328)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["rhuss"]} -->
knative-prow[bot]: Hi @dependabot[bot]. Thanks for your PR.

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

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

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

<details>

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


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 1b34b0989..b0c9828ac 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -22,7 +22,6 @@ aliases:
   - julz
   - psschwei
   control-protocol-approvers:
-  - devguyio
   - eric-sap
   - lionelvillard
   - matzew
@@ -65,7 +64,6 @@ aliases:
   eventing-kafka-approvers:
   - aliok
   - davyodom
-  - devguyio
   - eric-sap
   - lberk
   - matzew
@@ -80,11 +78,9 @@ aliases:
   eventing-kafka-mtsource-approvers:
   - steven0711dong
   eventing-kafka-wg-leads:
-  - devguyio
   - lionelvillard
   - travis-minke-sap
   eventing-kafka-writers:
-  - devguyio
   - lionelvillard
   - travis-minke-sap
   eventing-kogito-approvers:
@@ -92,13 +88,13 @@ aliases:
   - spolti
   - vaibhavjainwiz
   eventing-natss-approvers:
-  - devguyio
   - zhaojizhuang
   eventing-prometheus-approvers:
   - lberk
   - matzew
   eventing-rabbitmq-approvers:
   - chunyilyu
+  - gab-satchi
   - gabo1208
   - gvmw
   - ikvmw
@@ -109,11 +105,9 @@ aliases:
   - lionelvillard
   - matzew
   eventing-wg-leads:
-  - devguyio
   - lionelvillard
   eventing-writers:
   - aliok
-  - devguyio
   - lionelvillard
   - matzew
   - odacremolbap
@@ -182,27 +176,25 @@ aliases:
   - nicolaferraro
   - rhuss
   knative-admin:
-  - carlisia
   - csantanapr
   - dprotaso
   - evankanderson
+  - ikvmw
   - itsmurugappan
   - julz
   - knative-prow-releaser-robot
   - knative-prow-robot
   - knative-prow-updater-robot
   - knative-test-reporter-robot
-  - kvmware
   - lance
+  - nader-ziada
   - pmorie
   - rhuss
   - thisisnotapril
   - vaikas
-  - xtreme-sameer-vohra
   knative-release-leads:
-  - carlisia
-  - kvmware
-  - xtreme-sameer-vohra
+  - ikvmw
+  - nader-ziada
   knative-robots:
   - knative-prow-releaser-robot
   - knative-prow-robot
@@ -228,11 +220,13 @@ aliases:
   - JRBANCEL
   - ZhiminXiang
   - nak3
+  - skonto
   - tcnghia
   - vagababov
   net-kourier-approvers:
   - davidor
   - jmprusi
+  - skonto
   - tcnghia
   networking-wg-leads:
   - ZhiminXiang
diff --git a/control-plane/cmd/post-install/kafka_channel_migrator.go b/control-plane/cmd/post-install/kafka_channel_migrator.go
new file mode 100644
index 000000000..e24630e4e
--- /dev/null
+++ b/control-plane/cmd/post-install/kafka_channel_migrator.go
@@ -0,0 +1,270 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"go.uber.org/zap"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	kcs "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	"knative.dev/eventing-kafka/pkg/common/config"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/network"
+	"knative.dev/pkg/system"
+	"sigs.k8s.io/yaml"
+)
+
+type kafkaChannelMigrator struct {
+	kcs kcs.Interface
+	k8s kubernetes.Interface
+}
+
+const (
+	DataPlaneReadinessCheckInterval = 10 * time.Second
+	DataPlaneReadinessCheckTimeout  = 10 * time.Minute
+
+	NewChannelDispatcherDeploymentName = "kafka-channel-dispatcher"
+	NewChannelReceiverDeploymentName   = "kafka-channel-receiver"
+
+	NewChannelDispatcherServiceName = "kafka-channel-ingress"
+
+	OldConfigmapName = "config-kafka"
+
+	NewConfigmapNameKey     = "CHANNEL_GENERAL_CONFIG_MAP_NAME"
+	NewConfigmapNameDefault = "kafka-channel-config"
+)
+
+func (m *kafkaChannelMigrator) Migrate(ctx context.Context) error {
+	logger := logging.FromContext(ctx)
+
+	logger.Infof("Waiting %s for the new data plane to become ready before the migration.", DataPlaneReadinessCheckTimeout)
+
+	// wait until the new data plane is ready
+	err := m.waitForNewDataPlaneReady(ctx)
+	if err != nil {
+		return fmt.Errorf("error while waiting the new data plane to become ready %w", err)
+	}
+
+	logger.Infof("New data plane is ready, progressing with the migration")
+
+	err = m.migrateChannelServices(ctx, logger)
+	if err != nil {
+		return err
+	}
+
+	err = m.migrateConfigmap(ctx, logger)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (m *kafkaChannelMigrator) migrateChannelServices(ctx context.Context, logger *zap.SugaredLogger) error {
+	newDispatcherExternalName := network.GetServiceHostname(NewChannelDispatcherServiceName, system.Namespace())
+
+	logger.Infof("Starting migration of channel services to new dispatcher service: %s.", newDispatcherExternalName)
+
+	// Sample YAML for a consolidated KafkaChannel service:
+	// apiVersion: v1
+	//kind: Service
+	//metadata:
+	//  labels:
+	//    messaging.knative.dev/role: kafka-channel
+	// ...
+	//  name: kafka-channel-kn-channel
+	//  namespace: default
+	//  ownerReferences:
+	//  - apiVersion: messaging.knative.dev/v1beta1
+	//    ...
+	//spec:
+	//  externalName: kafka-ch-dispatcher.knative-eventing.svc.cluster.local
+	//  sessionAffinity: None
+	//  type: ExternalName
+	//status:
+	//  loadBalancer: {}
+
+	// KafkaChannel service (the one that is created per channel instance) in consolidated channel
+	// only has the label "messaging.knative.dev/role: kafka-channel".
+	// Get those services and redirect them to new dispatcher.
+	kafkaChannelServiceLabels := map[string]string{
+		// See https://github.com/knative-sandbox/eventing-kafka/blob/23602e84d707b8c106c93ed9b12737263488ad18/pkg/channel/consolidated/reconciler/controller/resources/service.go#L70
+		"messaging.knative.dev/role": "kafka-channel",
+	}
+
+	// cont takes care of results pagination.
+	// there might be more resources than a single rest call can return, so we need to think about the pagination.
+	cont := ""
+	first := true
+	lsOpts := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(kafkaChannelServiceLabels).String(), Continue: cont}
+	for first || cont != "" {
+		kafkaChannelServiceList, err := m.k8s.CoreV1().
+			Services(corev1.NamespaceAll).
+			List(ctx, lsOpts)
+		if err != nil {
+			return fmt.Errorf("failed to list KafkaChannel services: %w", err)
+		}
+		for _, svc := range kafkaChannelServiceList.Items {
+
+			patch := []byte(fmt.Sprintf(`[{"op":"replace", "path": "/spec/externalName", "value": "%s"}]`, newDispatcherExternalName))
+
+			logger.Infof("Patching service %s/%s with the patch: %s.", svc.Namespace, svc.Name, patch)
+
+			_, err := m.k8s.CoreV1().
+				Services(svc.Namespace).
+				Patch(ctx, svc.Name, types.JSONPatchType, patch, metav1.PatchOptions{})
+			if err != nil {
+				// return error to crash so that the job can retry
+				return fmt.Errorf("error while patching KafkaChannel service %s/%s: %w", svc.Namespace, svc.Name, err)
+			}
+		}
+		cont = kafkaChannelServiceList.Continue
+		first = false
+	}
+	return nil
+}
+
+func (m *kafkaChannelMigrator) migrateConfigmap(ctx context.Context, logger *zap.SugaredLogger) error {
+	// consolidated configmap looks like this:
+	//
+	// apiVersion: v1
+	// data:
+	//   eventing-kafka: |
+	//     kafka:
+	//       brokers: my-cluster-kafka-bootstrap.kafka:9092
+	//       authSecretNamespace: my-namespace
+	//       authSecretName: my-secret
+	//   version: 1.0.0
+	// kind: ConfigMap
+	// metadata:
+	//   name: config-kafka
+	//   namespace: knative-eventing
+
+	// the new configmap should look like this:
+	//
+	// apiVersion: v1
+	// kind: ConfigMap
+	// metadata:
+	//   name: kafka-channel-config
+	//   namespace: knative-eventing
+	//  data:
+	//    bootstrap.servers: "my-cluster-kafka-bootstrap.kafka:9092"
+	//    auth.secret.ref.namespace: my-ns
+	//    auth.secret.ref.name: my-secret
+	// These 2 things don't have anything in old configmap. Defaults of them in the old channel are hardcoded.
+	//    default.topic.partitions: "10"
+	//    default.topic.replication.factor: "3"
+
+	// get the old configmap, extract these and set it in the new configmap:
+	// OLD CM PATH                                   NEW CM PATH
+	// ----------------------------                  ------------------------
+	// eventing-kafka/kafka/brokers              --> bootstrap.servers
+	// eventing-kafka/kafka/authSecretNamespace  --> auth.secret.ref.namespace
+	// eventing-kafka/kafka/authSecretName       --> auth.secret.ref.name
+	//
+
+	logger.Infof("Migrating configmap.")
+
+	oldcm, err := m.k8s.CoreV1().
+		ConfigMaps(system.Namespace()).
+		Get(ctx, OldConfigmapName, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) || len(oldcm.Data) == 0 {
+		logger.Infof("Old configmap %s is either missing or empty. Skipping the configmap migration", OldConfigmapName)
+		return nil
+	}
+
+	if err != nil {
+		// there's some other problem
+		return fmt.Errorf("failed to get consolidated channel configmap for migration %s: %w", OldConfigmapName, err)
+	}
+
+	oldconfig, err := getEventingKafkaConfig(oldcm.Data)
+	if err != nil && !apierrors.IsNotFound(err) {
+		// configmap will be missing if we did the migration already
+		return fmt.Errorf("failed to build config from consolidated channel configmap for migration %s: %w", OldConfigmapName, err)
+	}
+
+	newConfigmapName := os.Getenv(NewConfigmapNameKey)
+	if newConfigmapName == "" {
+		newConfigmapName = NewConfigmapNameDefault
+	}
+
+	patches := []string{
+		fmt.Sprintf(`{"op":"replace", "path": "/data/bootstrap.servers", "value": "%s"}`, oldconfig.Kafka.Brokers),
+		fmt.Sprintf(`{"op":"replace", "path": "/data/auth.secret.ref.namespace", "value": "%s"}`, oldconfig.Kafka.AuthSecretNamespace),
+		fmt.Sprintf(`{"op":"replace", "path": "/data/auth.secret.ref.name", "value": "%s"}`, oldconfig.Kafka.AuthSecretName),
+	}
+
+	logger.Infof("Patching configmap %s with patch %s", newConfigmapName, patches)
+
+	patch := []byte(fmt.Sprintf("[%s]", strings.Join(patches, ",")))
+
+	_, err = m.k8s.CoreV1().
+		ConfigMaps(system.Namespace()).
+		Patch(ctx, newConfigmapName, types.JSONPatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to patch new channel configmap for migration %s: %w", newConfigmapName, err)
+	}
+	return nil
+}
+
+func getEventingKafkaConfig(configMap map[string]string) (*config.EventingKafkaConfig, error) {
+	// Unmarshal The Eventing-Kafka ConfigMap YAML Into A EventingKafkaSettings Struct
+	eventingKafkaConfig := &config.EventingKafkaConfig{}
+	err := yaml.Unmarshal([]byte(configMap[constants.EventingKafkaSettingsConfigKey]), &eventingKafkaConfig)
+	if err != nil {
+		return nil, fmt.Errorf("ConfigMap's eventing-kafka value could not be converted to an EventingKafkaConfig struct: %s : %v", err, configMap[constants.EventingKafkaSettingsConfigKey])
+	}
+
+	return eventingKafkaConfig, nil
+}
+
+func (m *kafkaChannelMigrator) waitForNewDataPlaneReady(ctx context.Context) error {
+	return wait.PollImmediate(DataPlaneReadinessCheckInterval, DataPlaneReadinessCheckTimeout, func() (bool, error) {
+		ready, err := isDeploymentReady(ctx, m.k8s, system.Namespace(), NewChannelDispatcherDeploymentName)
+		if err != nil {
+			return false, err
+		}
+		if !ready {
+			return false, nil
+		}
+
+		return isDeploymentReady(ctx, m.k8s, system.Namespace(), NewChannelReceiverDeploymentName)
+	})
+}
+
+func GetDeploymentCondition(status appsv1.DeploymentStatus, t appsv1.DeploymentConditionType) *appsv1.DeploymentCondition {
+	for _, cond := range status.Conditions {
+		if cond.Type == t {
+			return &cond
+		}
+	}
+	return nil
+}
diff --git a/control-plane/cmd/post-install/kafka_channel_post_migration_deleter.go b/control-plane/cmd/post-install/kafka_channel_post_migration_deleter.go
new file mode 100644
index 000000000..70c4c3a9e
--- /dev/null
+++ b/control-plane/cmd/post-install/kafka_channel_post_migration_deleter.go
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	"knative.dev/pkg/logging"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/pkg/system"
+)
+
+const (
+	ControlPlaneReadinessCheckInterval = 10 * time.Second
+	ControlPlaneReadinessCheckTimeout  = 10 * time.Minute
+
+	NewControllerDeploymentName = "kafka-controller"
+)
+
+type kafkaChannelPostMigrationDeleter struct {
+	k8s kubernetes.Interface
+}
+
+func (d *kafkaChannelPostMigrationDeleter) Delete(ctx context.Context) error {
+	// at this stage,
+	// - we deleted the old controler
+	// - we made the existing channel services to point to new data plane pods
+	// - we can now delete the old dispatcher resources and all the other leftovers
+	// HOWEVER, we need to check if the new controller is up and running.
+
+	logger := logging.FromContext(ctx)
+	logger.Infof("Waiting %s for the new control plane to become ready before the migration.", ControlPlaneReadinessCheckTimeout)
+
+	// wait until the new control plane is ready
+	err := d.waitForNewControlPlaneReady(ctx)
+	if err != nil {
+		return fmt.Errorf("error while waiting the new control plane to become ready %w", err)
+	}
+
+	logger.Infof("New control plane is ready, progressing with the migration")
+
+	///////////////////////////////////////////////////////////////////////////////
+	/////////// START DELETING DISPATCHER RESOURCES
+	///////////////////////////////////////////////////////////////////////////////
+
+	// Delete service knative-eventing/kafka-ch-dispatcher
+	const dispatcherServiceName = "kafka-ch-dispatcher"
+	err = d.k8s.
+		CoreV1().
+		Services(system.Namespace()).
+		Delete(ctx, dispatcherServiceName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service %s/%s: %w", system.Namespace(), dispatcherServiceName, err)
+	}
+
+	// Delete deployment.apps/kafka-ch-dispatcher
+	const dispatcherDeploymentName = "kafka-ch-dispatcher"
+	err = d.k8s.
+		AppsV1().
+		Deployments(system.Namespace()).
+		Delete(ctx, dispatcherDeploymentName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete deployment %s/%s: %w", system.Namespace(), dispatcherDeploymentName, err)
+	}
+
+	// Delete clusterrolebinding/kafka-ch-dispatcher
+	const dispatcherClusterRoleBinding = "kafka-ch-dispatcher"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoleBindings().
+		Delete(ctx, dispatcherClusterRoleBinding, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRoleBinding %s: %w", dispatcherClusterRoleBinding, err)
+	}
+
+	// Delete clusterrole/kafka-ch-dispatcher
+	const dispatcherClusterRole = "kafka-ch-dispatcher"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoles().
+		Delete(ctx, dispatcherClusterRole, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRole %s: %w", dispatcherClusterRole, err)
+	}
+
+	// Delete serviceaccount/kafka-ch-dispatcher
+	const dispatcherServiceAccount = "kafka-ch-dispatcher"
+	err = d.k8s.
+		CoreV1().
+		ServiceAccounts(system.Namespace()).
+		Delete(ctx, dispatcherServiceAccount, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service account %s/%s: %w", system.Namespace(), dispatcherServiceAccount, err)
+	}
+
+	// Delete configmap/config-leader-election-kafkachannel
+	const kafkaChannelLeaderElectionConfigmap = "config-leader-election-kafkachannel"
+	err = d.k8s.
+		CoreV1().
+		ConfigMaps(system.Namespace()).
+		Delete(ctx, kafkaChannelLeaderElectionConfigmap, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete Configmap %s: %w", kafkaChannelLeaderElectionConfigmap, err)
+	}
+
+	// Delete lease/kafkachannel-dispatcher.knative.dev.eventing-kafka.pkg.channel.consolidated.reconciler.dispatcher.reconciler.00-of-01
+	const kafkaChannelDispatcherLease = "kafkachannel-dispatcher.knative.dev.eventing-kafka.pkg.channel.consolidated.reconciler.dispatcher.reconciler.00-of-01"
+	err = d.k8s.
+		CoordinationV1().
+		Leases(system.Namespace()).
+		Delete(ctx, kafkaChannelDispatcherLease, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete Lease %s: %w", kafkaChannelDispatcherLease, err)
+	}
+
+	return nil
+}
+
+func (d *kafkaChannelPostMigrationDeleter) waitForNewControlPlaneReady(ctx context.Context) error {
+	return wait.PollImmediate(ControlPlaneReadinessCheckInterval, ControlPlaneReadinessCheckTimeout, func() (bool, error) {
+		return isDeploymentReady(ctx, d.k8s, system.Namespace(), NewControllerDeploymentName)
+	})
+}
diff --git a/control-plane/cmd/post-install/kafka_channel_pre_migration_deleter.go b/control-plane/cmd/post-install/kafka_channel_pre_migration_deleter.go
new file mode 100644
index 000000000..acf08b290
--- /dev/null
+++ b/control-plane/cmd/post-install/kafka_channel_pre_migration_deleter.go
@@ -0,0 +1,203 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"knative.dev/pkg/system"
+)
+
+type kafkaChannelPreMigrationDeleter struct {
+	k8s kubernetes.Interface
+}
+
+func (d *kafkaChannelPreMigrationDeleter) Delete(ctx context.Context) error {
+	/////////////////////////////////////////////////////////////////////////////
+	///////// START DELETING WEBHOOK RESOURCES
+	/////////////////////////////////////////////////////////////////////////////
+
+	// Delete mutatingwebhookconfiguration/defaulting.webhook.kafka.messaging.knative.dev
+	const channelMutatingWebhookConfigurationName = "defaulting.webhook.kafka.messaging.knative.dev"
+	err := d.k8s.
+		AdmissionregistrationV1().
+		MutatingWebhookConfigurations().
+		Delete(ctx, channelMutatingWebhookConfigurationName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete mutating webhook configuration %s: %w", channelMutatingWebhookConfigurationName, err)
+	}
+
+	// Delete validatingwebhookconfiguration/validation.webhook.kafka.messaging.knative.dev
+	const channelValidatingWebhookConfigurationName = "validation.webhook.kafka.messaging.knative.dev"
+	err = d.k8s.
+		AdmissionregistrationV1().
+		ValidatingWebhookConfigurations().
+		Delete(ctx, channelValidatingWebhookConfigurationName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete validating webhook configuration %s: %w", channelMutatingWebhookConfigurationName, err)
+	}
+
+	// Delete service knative-eventing/kafka-webhook
+	const channelWebhookServiceName = "kafka-webhook"
+	err = d.k8s.
+		CoreV1().
+		Services(system.Namespace()).
+		Delete(ctx, channelWebhookServiceName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service %s/%s: %w", system.Namespace(), channelWebhookServiceName, err)
+	}
+
+	// Delete deployment.apps/kafka-webhook
+	const channelWebhookDeploymentName = "kafka-webhook"
+	err = d.k8s.
+		AppsV1().
+		Deployments(system.Namespace()).
+		Delete(ctx, channelWebhookDeploymentName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete deployment %s/%s: %w", system.Namespace(), channelWebhookDeploymentName, err)
+	}
+
+	// Delete horizontalpodautoscaler/kafka-webhook
+	const channelWebhookHpaName = "kafka-webhook"
+	err = d.k8s.
+		AutoscalingV1().
+		HorizontalPodAutoscalers(system.Namespace()).
+		Delete(ctx, channelWebhookHpaName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete HPA %s/%s: %w", system.Namespace(), channelWebhookHpaName, err)
+	}
+
+	// Delete poddisruptionbudget/kafka-webhook
+	const channelWebhookPdbName = "kafka-webhook"
+	err = d.k8s.
+		PolicyV1().
+		PodDisruptionBudgets(system.Namespace()).
+		Delete(ctx, channelWebhookPdbName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete poddisruptionbudget %s/%s: %w", system.Namespace(), channelWebhookPdbName, err)
+	}
+
+	// Delete serviceaccount/kafka-webhook
+	const channelWebhookServiceAccount = "kafka-webhook"
+	err = d.k8s.
+		CoreV1().
+		ServiceAccounts(system.Namespace()).
+		Delete(ctx, channelWebhookServiceAccount, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service account %s/%s: %w", system.Namespace(), channelWebhookServiceAccount, err)
+	}
+
+	// Delete clusterrolebinding/kafka-webhook
+	const channelWebhookClusterRoleBinding = "kafka-webhook"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoleBindings().
+		Delete(ctx, channelWebhookClusterRoleBinding, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRoleBinding %s: %w", channelWebhookClusterRoleBinding, err)
+	}
+
+	// Delete clusterrole/kafka-webhook
+	const channelWebhookClusterRole = "kafka-webhook"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoles().
+		Delete(ctx, channelWebhookClusterRole, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRole %s: %w", channelWebhookClusterRole, err)
+	}
+
+	// Delete secret/messaging-webhook-certs
+	const channelWebhookSecret = "messaging-webhook-certs" /* #nosec G101 */ /* Potential hardcoded credentials (gosec) */
+	err = d.k8s.
+		CoreV1().
+		Secrets(system.Namespace()).
+		Delete(ctx, channelWebhookSecret, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete secret %s/%s: %w", system.Namespace(), channelWebhookSecret, err)
+	}
+
+	/////////////////////////////////////////////////////////////////////////////
+	///////// START DELETING CONTROLLER RESOURCES
+	/////////////////////////////////////////////////////////////////////////////
+
+	// Delete deployment/kafka-ch-controller
+	const channelControllerDeploymentName = "kafka-ch-controller"
+	err = d.k8s.
+		AppsV1().
+		Deployments(system.Namespace()).
+		Delete(ctx, channelControllerDeploymentName, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete deployment %s/%s: %w", system.Namespace(), channelControllerDeploymentName, err)
+	}
+
+	// Delete clusterrolebinding/kafka-ch-controller
+	const channelControllerClusterRoleBinding = "kafka-ch-controller"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoleBindings().
+		Delete(ctx, channelControllerClusterRoleBinding, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRoleBinding %s: %w", channelControllerClusterRoleBinding, err)
+	}
+
+	// Delete clusterrole/kafka-ch-controller
+	const channelControllerClusterRole = "kafka-ch-controller"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoles().
+		Delete(ctx, channelControllerClusterRole, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRole %s: %w", channelControllerClusterRole, err)
+	}
+
+	// Delete serviceaccount/kafka-ch-controller
+	const channelControllerServiceAccount = "kafka-ch-controller"
+	err = d.k8s.
+		CoreV1().
+		ServiceAccounts(system.Namespace()).
+		Delete(ctx, channelControllerServiceAccount, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete service account %s/%s: %w", system.Namespace(), channelControllerServiceAccount, err)
+	}
+
+	// Delete clusterrole/kafka-addressable-resolver
+	const channelAddressableResolverClusterRole = "kafka-addressable-resolver"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoles().
+		Delete(ctx, channelAddressableResolverClusterRole, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRole %s: %w", channelAddressableResolverClusterRole, err)
+	}
+
+	// Delete clusterrole/kafka-channelable-manipulator
+	const channelChannelableManipulatorClusterRole = "kafka-channelable-manipulator"
+	err = d.k8s.
+		RbacV1().
+		ClusterRoles().
+		Delete(ctx, channelChannelableManipulatorClusterRole, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to delete ClusterRole %s: %w", channelChannelableManipulatorClusterRole, err)
+	}
+
+	return nil
+}
diff --git a/control-plane/cmd/post-install/kafka_source_migrator.go b/control-plane/cmd/post-install/kafka_source_migrator.go
index 85e5af09c..ed79868eb 100644
--- a/control-plane/cmd/post-install/kafka_source_migrator.go
+++ b/control-plane/cmd/post-install/kafka_source_migrator.go
@@ -36,6 +36,7 @@ type KafkaSourceMigrator struct {
 func (m *KafkaSourceMigrator) Migrate(ctx context.Context) error {
 
 	// cont takes care of results pagination.
+	// there might be more resources than a single rest call can return, so we need to think about the pagination.
 	cont := ""
 	first := true
 	for first || cont != "" {
diff --git a/control-plane/cmd/post-install/main.go b/control-plane/cmd/post-install/main.go
index 90e3cdd36..fd00a5a4b 100644
--- a/control-plane/cmd/post-install/main.go
+++ b/control-plane/cmd/post-install/main.go
@@ -70,7 +70,29 @@ func run(ctx context.Context) error {
 		k8s: kubernetes.NewForConfigOrDie(config),
 	}
 	if err := sourceDeleter.Delete(ctx); err != nil {
-		return fmt.Errorf("source migration failed: %w", err)
+		return fmt.Errorf("source deletion failed: %w", err)
+	}
+
+	channelPreMigrationDeleter := &kafkaChannelPreMigrationDeleter{
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+	if err := channelPreMigrationDeleter.Delete(ctx); err != nil {
+		return fmt.Errorf("channel pre-deletion failed: %w", err)
+	}
+
+	channelMigrator := &kafkaChannelMigrator{
+		kcs: kcs.NewForConfigOrDie(config),
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+	if err := channelMigrator.Migrate(ctx); err != nil {
+		return fmt.Errorf("channel migration failed: %w", err)
+	}
+
+	channelPostMigrationDeleter := &kafkaChannelPostMigrationDeleter{
+		k8s: kubernetes.NewForConfigOrDie(config),
+	}
+	if err := channelPostMigrationDeleter.Delete(ctx); err != nil {
+		return fmt.Errorf("channel post-deletion failed: %w", err)
 	}
 
 	return nil
diff --git a/control-plane/cmd/post-install/util.go b/control-plane/cmd/post-install/util.go
new file mode 100644
index 000000000..7ff0116c0
--- /dev/null
+++ b/control-plane/cmd/post-install/util.go
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+func isDeploymentReady(ctx context.Context, k8s kubernetes.Interface, namespace, name string) (bool, error) {
+	deployment, err := k8s.
+		AppsV1().
+		Deployments(namespace).
+		Get(ctx, name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		// Return false as we are not done yet.
+		// We swallow the error to keep on polling.
+		return false, nil
+	} else if err != nil {
+		// Return error to stop the polling.
+		return false, err
+	}
+
+	ready := GetDeploymentCondition(deployment.Status, appsv1.DeploymentAvailable)
+	return ready != nil && ready.Status == corev1.ConditionTrue, nil
+}
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
index cfc3310f1..594152846 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
@@ -24,9 +24,7 @@ data:
                   [
                     {"Name": "PodFitsResources"},
                     {"Name": "NoMaxResourceCount",
-                    "Args": "{\"NumPartitions\": 100}"},
-                    {"Name": "EvenPodSpread",
-                    "Args": "{\"MaxSkew\": 2}"}
+                    "Args": "{\"NumPartitions\": 100}"}
                   ]
     priorities: |+
                   [
@@ -34,5 +32,8 @@ data:
                     "Weight": 10,
                     "Args":  "{\"MaxSkew\": 2}"},
                     {"Name": "LowestOrdinalPriority",
-                    "Weight": 2}
+                    "Weight": 2},
+                    {"Name": "EvenPodSpread",
+                    "Weight": 2,
+                    "Args": "{\"MaxSkew\": 2}"}
                   ]
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index ee9d3c2ea..c34adcc4b 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -25,6 +25,7 @@ rules:
       - "*"
     resources:
       - configmaps
+      - services
     verbs:
       - get
       - list
diff --git a/control-plane/config/post-install/200-controller-cluster-role.yaml b/control-plane/config/post-install/200-controller-cluster-role.yaml
index 9b1d9ff2b..1beefab9e 100644
--- a/control-plane/config/post-install/200-controller-cluster-role.yaml
+++ b/control-plane/config/post-install/200-controller-cluster-role.yaml
@@ -45,6 +45,7 @@ rules:
       - mutatingwebhookconfigurations
     resourceNames:
       - defaulting.webhook.kafka.sources.knative.dev
+      - defaulting.webhook.kafka.messaging.knative.dev
     verbs:
       - delete
 
@@ -55,6 +56,7 @@ rules:
     resourceNames:
       - validation.webhook.kafka.sources.knative.dev
       - config.webhook.kafka.sources.knative.dev
+      - validation.webhook.kafka.messaging.knative.dev
     verbs:
       - delete
 
@@ -65,18 +67,44 @@ rules:
     resourceNames:
       - kafka-source-webhook
       - kafka-controller
+      - kafka-webhook
+      - kafka-ch-dispatcher
     verbs:
       - delete
 
+  # to be able to list channel services and patch them
+  - apiGroups:
+      - "*"
+    resources:
+      - services
+    verbs:
+      - list
+      - patch
+
   - apiGroups:
       - apps
     resources:
       - deployments
     resourceNames:
       - kafka-controller-manager
+      - kafka-webhook
+      - kafka-ch-controller
+      - kafka-ch-dispatcher
     verbs:
       - delete
 
+  # to be able to check readiness of the new dataplane
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    resourceNames:
+      - kafka-channel-dispatcher
+      - kafka-channel-receiver
+      - kafka-controller
+    verbs:
+      - get
+
   - apiGroups:
       - rbac.authorization.k8s.io
     resources:
@@ -84,6 +112,9 @@ rules:
     resourceNames:
       - eventing-sources-kafka-controller
       - eventing-sources-kafka-controller-addressable-resolver
+      - kafka-webhook
+      - kafka-ch-controller
+      - kafka-ch-dispatcher
     verbs:
       - delete
   - apiGroups:
@@ -92,6 +123,11 @@ rules:
       - clusterroles
     resourceNames:
       - eventing-sources-kafka-controller
+      - kafka-webhook
+      - kafka-ch-controller
+      - kafka-addressable-resolver
+      - kafka-channelable-manipulator
+      - kafka-ch-dispatcher
     verbs:
       - delete
 
@@ -101,6 +137,9 @@ rules:
       - serviceaccounts
     resourceNames:
       - kafka-controller-manager
+      - kafka-webhook
+      - kafka-ch-controller
+      - kafka-ch-dispatcher
     verbs:
       - delete
 
@@ -110,5 +149,62 @@ rules:
       - secrets
     resourceNames:
       - kafka-source-webhook-certs
+      - messaging-webhook-certs
+    verbs:
+      - delete
+
+  - apiGroups:
+      - "*"
+    resources:
+      - horizontalpodautoscalers
+    resourceNames:
+      - kafka-webhook
+    verbs:
+      - delete
+
+  - apiGroups:
+      - "*"
+    resources:
+      - poddisruptionbudgets
+    resourceNames:
+      - kafka-webhook
+    verbs:
+      - delete
+
+  - apiGroups:
+    - "*"
+    resources:
+      - leases
+    resourceNames:
+      - kafkachannel-dispatcher.knative.dev.eventing-kafka.pkg.channel.consolidated.reconciler.dispatcher.reconciler.00-of-01
+    verbs:
+      - delete
+
+  - apiGroups:
+      - "*"
+    resources:
+      - configmaps
+    resourceNames:
+      - config-leader-election-kafkachannel
     verbs:
       - delete
+
+  # to be able to read the old configmap and migrate it
+  - apiGroups:
+      - "*"
+    resources:
+      - configmaps
+    resourceNames:
+      - config-kafka
+    verbs:
+      - get
+
+  # to be able update config in the new configmap
+  - apiGroups:
+      - "*"
+    resources:
+      - configmaps
+    resourceNames:
+      - kafka-channel-config
+    verbs:
+      - patch
diff --git a/control-plane/config/post-install/500-post-install-job.yaml b/control-plane/config/post-install/500-post-install-job.yaml
index 53f970550..c5be8e2cb 100644
--- a/control-plane/config/post-install/500-post-install-job.yaml
+++ b/control-plane/config/post-install/500-post-install-job.yaml
@@ -41,3 +41,5 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: CHANNEL_GENERAL_CONFIG_MAP_NAME
+              value: kafka-channel-config
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index b22878227..06cd7c68c 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -7,10 +7,11 @@
 package contract
 
 import (
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	reflect "reflect"
 	sync "sync"
+
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 )
 
 const (
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 77f7fd11d..0f5959212 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -22,6 +22,10 @@ import (
 	"strconv"
 	"time"
 
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/Shopify/sarama"
 
 	"go.uber.org/multierr"
@@ -40,6 +44,7 @@ import (
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
 	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
@@ -55,8 +60,9 @@ import (
 
 const (
 	// TopicPrefix is the Kafka Channel topic prefix - (topic name: knative-messaging-kafka.<channel-namespace>.<channel-name>).
-	TopicPrefix          = "knative-messaging-kafka"
-	DefaultDeliveryOrder = contract.DeliveryOrder_ORDERED
+	TopicPrefix                  = "knative-messaging-kafka"
+	DefaultDeliveryOrder         = contract.DeliveryOrder_ORDERED
+	NewChannelIngressServiceName = "kafka-channel-dispatcher"
 )
 
 type Reconciler struct {
@@ -79,6 +85,7 @@ type Reconciler struct {
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
 	ConfigMapLister corelisters.ConfigMapLister
+	ServiceLister   corelisters.ServiceLister
 
 	Prober prober.Prober
 
@@ -283,6 +290,12 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	}
 	statusConditionManager.Addressable(address)
 
+	err = r.reconcileChannelService(ctx, channel)
+	if err != nil {
+		logger.Error("Error reconciling the backwards compatibility channel service.", zap.Error(err))
+		return err
+	}
+
 	return nil
 }
 
@@ -516,8 +529,6 @@ func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messaging
 }
 
 func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
-	// TODO: do we want to support namespaced channels? they're not supported at the moment.
-
 	namespace := system.Namespace()
 	cm, err := r.ConfigMapLister.ConfigMaps(namespace).Get(r.Env.GeneralConfigMapName)
 	if err != nil {
@@ -589,6 +600,38 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
+func (r *Reconciler) reconcileChannelService(ctx context.Context, channel *messagingv1beta1.KafkaChannel) error {
+	expected, err := resources.MakeK8sService(channel, resources.ExternalService(system.Namespace(), NewChannelIngressServiceName))
+	if err != nil {
+		return fmt.Errorf("failed to create the channel service object: %w", err)
+	}
+
+	svc, err := r.ServiceLister.Services(channel.Namespace).Get(resources.MakeChannelServiceName(channel.Name))
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Create(ctx, expected, metav1.CreateOptions{})
+			if err != nil {
+				return fmt.Errorf("failed to create the channel service object: %w", err)
+			}
+			return nil
+		}
+		return err
+	} else if !equality.Semantic.DeepEqual(svc.Spec, expected.Spec) {
+		svc = svc.DeepCopy()
+		svc.Spec = expected.Spec
+
+		_, err = r.KubeClient.CoreV1().Services(channel.Namespace).Update(ctx, svc, metav1.UpdateOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to update the channel service: %w", err)
+		}
+	}
+	// Check to make sure that the KafkaChannel owns this service and if not, complain.
+	if !metav1.IsControlledBy(svc, channel) {
+		return fmt.Errorf("kafkachannel: %s/%s does not own Service: %q", channel.Namespace, channel.Name, svc.Name)
+	}
+	return nil
+}
+
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 8ccfa800b..95b327420 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -34,6 +34,7 @@ import (
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
 
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -52,6 +53,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
 	configmapInformer := configmapinformer.Get(ctx)
+	serviceInformer := serviceinformer.Get(ctx)
 
 	reconciler := &Reconciler{
 		Reconciler: &base.Reconciler{
@@ -70,6 +72,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		InitOffsetsFunc:            offset.InitOffsets,
 		Env:                        configs,
 		ConfigMapLister:            configmapInformer.Lister(),
+		ServiceLister:              serviceInformer.Lister(),
 	}
 
 	logger := logging.FromContext(ctx)
@@ -96,6 +99,11 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		impl.GlobalResync(channelInformer.Informer())
 	}
 
+	serviceInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterController(&messagingv1beta.KafkaChannel{}),
+		Handler:    controller.HandleAll(globalResync),
+	})
+
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.DataPlaneConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 20bd88104..672326a08 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -45,7 +45,7 @@
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.4</micrometer.version>
     <opentelemetry.version>1.12.0</opentelemetry.version>
-    <jackson.version>2.13.2.20220324</jackson.version>
+    <jackson.version>2.13.2.20220328</jackson.version>
     <protobuf.version>3.19.4</protobuf.version>
     <slf4j.version>1.7.36</slf4j.version>
     <ch.qos.logback.version>1.2.11</ch.qos.logback.version>
diff --git a/go.mod b/go.mod
index f42c4d6d2..2068175b6 100644
--- a/go.mod
+++ b/go.mod
@@ -28,4 +28,5 @@ require (
 	knative.dev/hack v0.0.0-20220328133751-f06773764ce3
 	knative.dev/pkg v0.0.0-20220325200448-1f7514acd0c2
 	knative.dev/reconciler-test v0.0.0-20220328072550-7d32310c9b3a
+	sigs.k8s.io/yaml v1.3.0
 )
diff --git a/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
new file mode 100644
index 000000000..d817d109b
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
@@ -0,0 +1,200 @@
+/*
+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 (
+	v1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/system"
+
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
+)
+
+const (
+	DispatcherContainerName = "dispatcher"
+)
+
+var (
+	dispatcherName   = "kafka-ch-dispatcher"
+	dispatcherLabels = map[string]string{
+		"messaging.knative.dev/channel": "kafka-channel",
+		"messaging.knative.dev/role":    "dispatcher",
+	}
+)
+
+type DispatcherDeploymentBuilder struct {
+	deployment *v1.Deployment
+	args       *DispatcherDeploymentArgs
+}
+
+type DispatcherDeploymentArgs struct {
+	DispatcherScope       string
+	DispatcherNamespace   string
+	Image                 string
+	Replicas              int32
+	ServiceAccount        string
+	ConfigMapHash         string
+	OwnerRef              metav1.OwnerReference
+	DeploymentAnnotations map[string]string
+	DeploymentLabels      map[string]string
+	PodAnnotations        map[string]string
+	PodLabels             map[string]string
+}
+
+// NewDispatcherDeploymentBuilder returns a builder which builds from scratch a dispatcher deployment.
+// Intended to be used when creating the dispatcher deployment for the first time.
+func NewDispatcherDeploymentBuilder() *DispatcherDeploymentBuilder {
+	b := &DispatcherDeploymentBuilder{}
+	b.deployment = dispatcherTemplate()
+	return b
+}
+
+// NewDispatcherDeploymentBuilderFromDeployment returns a builder which builds a dispatcher deployment from the given deployment.
+// Intended to be used when updating an existing dispatcher deployment.
+func NewDispatcherDeploymentBuilderFromDeployment(d *v1.Deployment) *DispatcherDeploymentBuilder {
+	b := &DispatcherDeploymentBuilder{}
+	b.deployment = d
+	return b
+}
+
+func (b *DispatcherDeploymentBuilder) WithArgs(args *DispatcherDeploymentArgs) *DispatcherDeploymentBuilder {
+	b.args = args
+	return b
+}
+
+func (b *DispatcherDeploymentBuilder) 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.ObjectMeta.Annotations = commonconfig.JoinStringMaps(b.deployment.ObjectMeta.Annotations, b.args.DeploymentAnnotations)
+	b.deployment.ObjectMeta.Labels = commonconfig.JoinStringMaps(b.deployment.ObjectMeta.Labels, b.args.DeploymentLabels)
+
+	b.deployment.Spec.Replicas = &replicas
+	defaultAnnotations := map[string]string{commonconstants.ConfigMapHashAnnotationKey: b.args.ConfigMapHash}
+	b.deployment.Spec.Template.ObjectMeta.Annotations = commonconfig.JoinStringMaps(defaultAnnotations, b.args.PodAnnotations)
+	b.deployment.Spec.Template.ObjectMeta.Labels = commonconfig.JoinStringMaps(b.deployment.Spec.Template.ObjectMeta.Labels, b.args.PodLabels)
+	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{
+			APIVersion: "apps/v1",
+			Kind:       "Deployments",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: dispatcherName,
+		},
+		Spec: v1.DeploymentSpec{
+			Selector: &metav1.LabelSelector{
+				MatchLabels: dispatcherLabels,
+			},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: dispatcherLabels,
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: DispatcherContainerName,
+							Ports: []corev1.ContainerPort{{
+								Name:          "metrics",
+								ContainerPort: 9090,
+							}},
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      constants.SettingsConfigMapName,
+									MountPath: constants.SettingsConfigMapMountPath,
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: constants.SettingsConfigMapName,
+							VolumeSource: corev1.VolumeSource{
+								ConfigMap: &corev1.ConfigMapVolumeSource{
+									LocalObjectReference: corev1.LocalObjectReference{
+										Name: constants.SettingsConfigMapName,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func makeEnv(args *DispatcherDeploymentArgs) []corev1.EnvVar {
+	vars := []corev1.EnvVar{{
+		Name:  system.NamespaceEnvKey,
+		Value: system.Namespace(),
+	}, {
+		Name:  "METRICS_DOMAIN",
+		Value: "knative.dev/eventing",
+	}, {
+		Name:  "CONFIG_LOGGING_NAME",
+		Value: "config-logging",
+	}, {
+		Name:  "CONFIG_LEADERELECTION_NAME",
+		Value: "config-leader-election",
+	}}
+
+	if args.DispatcherScope == "namespace" {
+		vars = append(vars, corev1.EnvVar{
+			Name: "NAMESPACE",
+			ValueFrom: &corev1.EnvVarSource{
+				FieldRef: &corev1.ObjectFieldSelector{
+					FieldPath: "metadata.namespace",
+				},
+			},
+		})
+	}
+
+	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/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go
new file mode 100644
index 000000000..e29c25c53
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/dispatcher_service.go
@@ -0,0 +1,88 @@
+/*
+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 (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+)
+
+type DispatcherServiceBuilder struct {
+	service *corev1.Service
+	args    *DispatcherServiceArgs
+}
+
+type DispatcherServiceArgs struct {
+	DispatcherNamespace string
+	ServiceAnnotations  map[string]string
+	ServiceLabels       map[string]string
+}
+
+// NewDispatcherServiceBuilder returns a builder which builds from scratch a dispatcher service.
+// Intended to be used when creating the dispatcher service for the first time.
+func NewDispatcherServiceBuilder() *DispatcherServiceBuilder {
+	b := &DispatcherServiceBuilder{}
+	b.service = dispatcherServiceTemplate()
+	return b
+}
+
+// NewDispatcherServiceBuilderFromService returns a builder which builds a dispatcher service from the given service.
+// Intended to be used when updating an existing dispatcher service.
+func NewDispatcherServiceBuilderFromService(s *corev1.Service) *DispatcherServiceBuilder {
+	b := &DispatcherServiceBuilder{}
+	b.service = s
+	return b
+}
+
+func (b *DispatcherServiceBuilder) WithArgs(args *DispatcherServiceArgs) *DispatcherServiceBuilder {
+	b.args = args
+	return b
+}
+
+func (b *DispatcherServiceBuilder) Build() *corev1.Service {
+	b.service.ObjectMeta.Namespace = b.args.DispatcherNamespace
+	b.service.ObjectMeta.Annotations = commonconfig.JoinStringMaps(b.service.Annotations, b.args.ServiceAnnotations)
+	b.service.ObjectMeta.Labels = commonconfig.JoinStringMaps(b.service.Labels, b.args.ServiceLabels)
+	return b.service
+}
+
+func dispatcherServiceTemplate() *corev1.Service {
+	return &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Service",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   dispatcherName,
+			Labels: dispatcherLabels,
+		},
+		Spec: corev1.ServiceSpec{
+			Selector: dispatcherLabels,
+			Ports: []corev1.ServicePort{
+				{
+					Name:       "http-dispatcher",
+					Protocol:   corev1.ProtocolTCP,
+					Port:       80,
+					TargetPort: intstr.IntOrString{IntVal: 8080},
+				},
+			},
+		},
+	}
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/role_binding.go b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/role_binding.go
new file mode 100644
index 000000000..bdd6fe227
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/role_binding.go
@@ -0,0 +1,46 @@
+/*
+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 (
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// MakeRoleBinding creates a RoleBinding object for the Kafka dispatcher
+// service account 'sa' in the Namespace 'ns'.
+func MakeRoleBinding(ns, name string, sa *corev1.ServiceAccount, roleName string) *rbacv1.RoleBinding {
+	return &rbacv1.RoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+		},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: "rbac.authorization.k8s.io",
+			Kind:     "ClusterRole",
+			Name:     roleName,
+		},
+		Subjects: []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: sa.Namespace,
+				Name:      sa.Name,
+			},
+		},
+	}
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go
new file mode 100644
index 000000000..6dc5f148b
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service.go
@@ -0,0 +1,92 @@
+/*
+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 (
+	"fmt"
+
+	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"
+	"knative.dev/pkg/network"
+)
+
+const (
+	portName           = "http"
+	portNumber         = 80
+	MessagingRoleLabel = "messaging.knative.dev/role"
+	MessagingRole      = "kafka-channel"
+)
+
+// ServiceOption can be used to optionally modify the K8s service in MakeK8sService.
+type ServiceOption func(*corev1.Service) error
+
+func MakeChannelServiceName(name string) string {
+	return fmt.Sprintf("%s-kn-channel", name)
+}
+
+// ExternalService is a functional option for MakeK8sService to create a K8s service of type ExternalName
+// pointing to the specified service in a namespace.
+func ExternalService(namespace, service string) ServiceOption {
+	return func(svc *corev1.Service) error {
+		// TODO this overrides the current serviceSpec. Is this correct?
+		svc.Spec = corev1.ServiceSpec{
+			Type:         corev1.ServiceTypeExternalName,
+			ExternalName: network.GetServiceHostname(service, namespace),
+		}
+		return nil
+	}
+}
+
+// MakeK8sService creates a new K8s Service for a Channel resource. It also sets the appropriate
+// OwnerReferences on the resource so handleObject can discover the Channel resource that 'owns' it.
+// As well as being garbage collected when the Channel is deleted.
+func MakeK8sService(kc *v1beta1.KafkaChannel, opts ...ServiceOption) (*corev1.Service, error) {
+	// Add annotations
+	svc := &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Service",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      MakeChannelServiceName(kc.ObjectMeta.Name),
+			Namespace: kc.Namespace,
+			Labels: map[string]string{
+				MessagingRoleLabel: MessagingRole,
+			},
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(kc),
+			},
+		},
+		Spec: corev1.ServiceSpec{
+			Ports: []corev1.ServicePort{
+				{
+					Name:     portName,
+					Protocol: corev1.ProtocolTCP,
+					Port:     portNumber,
+				},
+			},
+		},
+	}
+	for _, opt := range opts {
+		if err := opt(svc); err != nil {
+			return nil, err
+		}
+	}
+	return svc, nil
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service_account.go b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service_account.go
new file mode 100644
index 000000000..d42d045d2
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources/service_account.go
@@ -0,0 +1,32 @@
+/*
+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 (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// MakeServiceAccount creates a ServiceAccount object for the Namespace 'ns'.
+func MakeServiceAccount(namespace, name string) *corev1.ServiceAccount {
+	return &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+	}
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go
new file mode 100644
index 000000000..27974c092
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	service "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
+	fake "knative.dev/pkg/client/injection/kube/informers/factory/fake"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = service.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Core().V1().Services()
+	return context.WithValue(ctx, service.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go
new file mode 100644
index 000000000..2ae3fab57
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/service/service.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package service
+
+import (
+	context "context"
+
+	apicorev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/core/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/listers/core/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Core().V1().Services()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.ServiceInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/core/v1.ServiceInformer from context.")
+	}
+	return untyped.(v1.ServiceInformer)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1.ServiceInformer = (*wrapper)(nil)
+var _ corev1.ServiceLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apicorev1.Service{}, 0, nil)
+}
+
+func (w *wrapper) Lister() corev1.ServiceLister {
+	return w
+}
+
+func (w *wrapper) Services(namespace string) corev1.ServiceNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apicorev1.Service, err error) {
+	lo, err := w.client.CoreV1().Services(w.namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apicorev1.Service, error) {
+	return w.client.CoreV1().Services(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 28259433f..5caf2dce6 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1310,6 +1310,7 @@ knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1
 knative.dev/eventing-kafka/pkg/apis/sources/config
 knative.dev/eventing-kafka/pkg/apis/sources/v1beta1
 knative.dev/eventing-kafka/pkg/channel/consolidated/apis/messaging
+knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources
 knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/testing
 knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants
 knative.dev/eventing-kafka/pkg/client/clientset/versioned
@@ -1394,6 +1395,8 @@ knative.dev/pkg/client/injection/kube/informers/core/v1/pod
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake
+knative.dev/pkg/client/injection/kube/informers/core/v1/service
+knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake
 knative.dev/pkg/client/injection/kube/informers/factory
 knative.dev/pkg/client/injection/kube/informers/factory/fake
 knative.dev/pkg/codegen/cmd/injection-gen
@@ -1496,4 +1499,5 @@ sigs.k8s.io/structured-merge-diff/v4/schema
 sigs.k8s.io/structured-merge-diff/v4/typed
 sigs.k8s.io/structured-merge-diff/v4/value
 # sigs.k8s.io/yaml v1.3.0
+## explicit
 sigs.k8s.io/yaml


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index e60521a06..e9d7045bf 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -74,7 +74,7 @@ var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
 	GeneralConfigMapName:        "kafka-channel-config",
-	IngressName:                 "kafka-channel-ingress",
+	IngressName:                 "kafka-channel-dispatcher",
 	SystemNamespace:             "knative-eventing",
 	DataPlaneConfigFormat:       base.Json,
 }
@@ -85,12 +85,13 @@ var DefaultEnv = &config.Env{
 // TODO: test if things from channel spec is used properly?
 // TODO: test if things from subscription spec is used properly?
 // TODO: rename test cases consistently
-// TODO: test finalize?
 // TODO: fix the order of status updates
 // TODO: add tests where there are no CM updates
 
 func TestReconcileKind(t *testing.T) {
 
+	t.Setenv("SYSTEM_NAMESPACE", "knative-eventing")
+
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
 	env := *DefaultEnv
@@ -189,6 +190,7 @@ func TestReconcileKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
+				NewPerChannelService(DefaultEnv),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -221,6 +223,7 @@ func TestReconcileKind(t *testing.T) {
 					}),
 				),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -505,6 +508,7 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -578,6 +582,7 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithUnreadySubscriber))),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -651,6 +656,7 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1())),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -727,6 +733,7 @@ func TestReconcileKind(t *testing.T) {
 					WithSubscribers(Subscriber2(WithFreshSubscriber)),
 				),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -1002,6 +1009,7 @@ func TestReconcileKind(t *testing.T) {
 					WithRetentionDuration("1000"),
 				),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -1076,6 +1084,7 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -1160,6 +1169,7 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewChannel(),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -1227,6 +1237,7 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewChannel(),
 				NewService(),
+				NewPerChannelService(DefaultEnv),
 				ChannelReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -1339,6 +1350,71 @@ func TestReconcileKind(t *testing.T) {
 	useTable(t, table, env)
 }
 
+func TestFinalizeKind(t *testing.T) {
+
+	t.Setenv("SYSTEM_NAMESPACE", "knative-eventing")
+
+	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
+
+	env := *DefaultEnv
+	testKey := fmt.Sprintf("%s/%s", ChannelNamespace, ChannelName)
+
+	table := TableTest{
+		{
+			Name: "Finalize normal - no auth",
+			Objects: []runtime.Object{
+				NewDeletedChannel(),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
+							},
+						},
+					},
+				}, &env),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 2,
+					Resources:  []*contract.Resource{},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+	}
+
+	useTable(t, table, env)
+}
+
 func useTable(t *testing.T, table TableTest, env config.Env) {
 	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
@@ -1371,6 +1447,7 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 			},
 			Env:             env,
 			ConfigMapLister: listers.GetConfigMapLister(),
+			ServiceLister:   listers.GetServiceLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index cc3ca16ce..1717bcf8e 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -29,6 +29,7 @@ import (
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake"
 	"knative.dev/pkg/configmap"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index a56639ac8..20b1b4ce9 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -17,7 +17,6 @@
 package consumergroup
 
 import (
-	"os"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -87,10 +86,10 @@ func TestNewController(t *testing.T) {
 		},
 	)
 
-	os.Setenv("AUTOSCALER_REFRESH_PERIOD", RefreshPeriod)
-	os.Setenv("POD_CAPACITY", PodCapacity)
-	os.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
-	os.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
+	t.Setenv("AUTOSCALER_REFRESH_PERIOD", RefreshPeriod)
+	t.Setenv("POD_CAPACITY", PodCapacity)
+	t.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
+	t.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
 	controller := NewController(ctx)
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 3fd1175ef..31a261c78 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -130,6 +130,10 @@ func (l *Listers) GetConfigMapLister() corelisters.ConfigMapLister {
 	return corelisters.NewConfigMapLister(l.indexerFor(&corev1.ConfigMap{}))
 }
 
+func (l *Listers) GetServiceLister() corelisters.ServiceLister {
+	return corelisters.NewServiceLister(l.indexerFor(&corev1.Service{}))
+}
+
 func (l *Listers) GetKafkaSinkLister() eventingkafkabrokerlisters.KafkaSinkLister {
 	return eventingkafkabrokerlisters.NewKafkaSinkLister(l.indexerFor(&eventingkafkabroker.KafkaSink{}))
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index 32deead76..f52359054 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -25,7 +25,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/utils/pointer"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/reconciler/controller/resources"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -43,6 +45,7 @@ const (
 	ChannelNamespace        = "test-nc"
 	ChannelUUID             = "c1234567-8901-2345-6789-123456789101"
 	ChannelBootstrapServers = "kafka-1:9092,kafka-2:9093"
+	ChannelServiceName      = "kc-kn-channel"
 
 	Subscription1Name     = "sub-1"
 	Subscription2Name     = "sub-2"
@@ -73,6 +76,15 @@ func NewChannel(options ...KRShapedOption) *messagingv1beta1.KafkaChannel {
 	return c
 }
 
+func NewDeletedChannel(options ...KRShapedOption) runtime.Object {
+	return NewChannel(
+		append(
+			options,
+			WithDeletedTimeStamp,
+		)...,
+	)
+}
+
 func WithNumPartitions(np int32) KRShapedOption {
 	return func(obj duckv1.KRShaped) {
 		ch := obj.(*messagingv1beta1.KafkaChannel)
@@ -302,3 +314,39 @@ func WithUnknownSubscriber(sub *SubscriberInfo) {
 	sub.status.Ready = "Unknown"
 	sub.status.Message = fmt.Sprintf("Subscriber %v not ready: %v", sub.spec.UID, "consumer group status unknown")
 }
+
+func NewPerChannelService(env *config.Env) *corev1.Service {
+	s := &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Service",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ChannelServiceName,
+			Namespace: ChannelNamespace,
+			Labels: map[string]string{
+				resources.MessagingRoleLabel: resources.MessagingRole,
+			},
+			OwnerReferences: []metav1.OwnerReference{
+				ChannelAsOwnerReference(),
+			},
+		},
+		Spec: corev1.ServiceSpec{
+			Type:         corev1.ServiceTypeExternalName,
+			ExternalName: network.GetServiceHostname(env.IngressName, env.SystemNamespace),
+		},
+	}
+
+	return s
+}
+
+func ChannelAsOwnerReference() metav1.OwnerReference {
+	return metav1.OwnerReference{
+		APIVersion:         messagingv1beta1.SchemeGroupVersion.String(),
+		Kind:               "KafkaChannel",
+		Name:               ChannelName,
+		UID:                ChannelUUID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(true),
+	}
+}