Repository: knative-sandbox/eventing-kafka-broker
Pull request #1158 (2021-08-23 11:05:08)

[Automated] Update eventing-kafka-broker nightly
/assign @knative/delivery-wg-leads
/cc @knative/delivery-wg-leads

Produced via:
```shell
for x in eventing-crds.yaml eventing-core.yaml; do
  curl https://storage.googleapis.com/knative-nightly/eventing/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-latest/$x
done
```

Details:
```

```

knative-prow-robot: @knative-automation: GitHub didn't allow me to request PR reviews from the following users: knative/delivery-wg-leads.

Note that only [knative-sandbox members](https://github.com/orgs/knative-sandbox/people) and repo collaborators can review this PR, and authors cannot review their own PRs.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1158):

>/assign @knative/delivery-wg-leads
>/cc @knative/delivery-wg-leads
>
>Produced via:
>```shell
>for x in eventing-crds.yaml eventing-core.yaml; do
>  curl https://storage.googleapis.com/knative-nightly/eventing/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-latest/$x
>done
>```
>
>Details:
>```
>
>```


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1158#" title="Author self-approved">knative-automation</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"]} -->

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


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
index 4cdd7f5a3..2a331a678 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package v1alpha1
 
 import (
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
index 377b8177b..22784c098 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package v1alpha1
 
 import (
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion_test.go
new file mode 100644
index 000000000..6fd4b8fda
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_conversion_test.go
@@ -0,0 +1,34 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"testing"
+)
+
+func TestKafkaChannelConversionBadType(t *testing.T) {
+	good, bad := &KafkaChannel{}, &KafkaChannel{}
+
+	if err := good.ConvertTo(context.Background(), bad); err == nil {
+		t.Errorf("ConvertTo() = %#v, wanted error", bad)
+	}
+
+	if err := good.ConvertFrom(context.Background(), bad); err == nil {
+		t.Errorf("ConvertFrom() = %#v, wanted error", good)
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
new file mode 100644
index 000000000..05c872b1b
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	testNumPartitions     = 10
+	testReplicationFactor = 5
+)
+
+func TestKafkaChannelDefaults(t *testing.T) {
+	testCases := map[string]struct {
+		initial  KafkaChannel
+		expected KafkaChannel
+	}{
+		"nil spec": {
+			initial: KafkaChannel{},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     DefaultNumPartitions,
+					ReplicationFactor: DefaultReplicationFactor,
+				},
+			},
+		},
+		"numPartitions not set": {
+			initial: KafkaChannel{
+				Spec: KafkaChannelSpec{
+					ReplicationFactor: testReplicationFactor,
+				},
+			},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     DefaultNumPartitions,
+					ReplicationFactor: testReplicationFactor,
+				},
+			},
+		},
+		"replicationFactor not set": {
+			initial: KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions: testNumPartitions,
+				},
+			},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     testNumPartitions,
+					ReplicationFactor: DefaultReplicationFactor,
+				},
+			},
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			tc.initial.SetDefaults(context.TODO())
+			if diff := cmp.Diff(tc.expected, tc.initial); diff != "" {
+				t.Fatalf("Unexpected defaults (-want, +got): %s", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go
new file mode 100644
index 000000000..588c0aa99
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go
@@ -0,0 +1,436 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"testing"
+
+	"knative.dev/pkg/apis"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var condReady = apis.Condition{
+	Type:   KafkaChannelConditionReady,
+	Status: corev1.ConditionTrue,
+}
+
+var condDispatcherNotReady = apis.Condition{
+	Type:   KafkaChannelConditionDispatcherReady,
+	Status: corev1.ConditionFalse,
+}
+
+var deploymentConditionReady = appsv1.DeploymentCondition{
+	Type:   appsv1.DeploymentAvailable,
+	Status: corev1.ConditionTrue,
+}
+
+var deploymentConditionNotReady = appsv1.DeploymentCondition{
+	Type:   appsv1.DeploymentAvailable,
+	Status: corev1.ConditionFalse,
+}
+
+var deploymentStatusReady = &appsv1.DeploymentStatus{Conditions: []appsv1.DeploymentCondition{deploymentConditionReady}}
+var deploymentStatusNotReady = &appsv1.DeploymentStatus{Conditions: []appsv1.DeploymentCondition{deploymentConditionNotReady}}
+
+var ignoreAllButTypeAndStatus = cmpopts.IgnoreFields(
+	apis.Condition{},
+	"LastTransitionTime", "Message", "Reason", "Severity")
+
+func TestKafkaChannelGetConditionSet(t *testing.T) {
+	r := &KafkaChannel{}
+
+	if got, want := r.GetConditionSet().GetTopLevelConditionType(), apis.ConditionReady; got != want {
+		t.Errorf("GetTopLevelCondition=%v, want=%v", got, want)
+	}
+}
+
+func TestChannelGetCondition(t *testing.T) {
+	tests := []struct {
+		name      string
+		cs        *KafkaChannelStatus
+		condQuery apis.ConditionType
+		want      *apis.Condition
+	}{{
+		name: "single condition",
+		cs: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						condReady,
+					},
+				},
+			},
+		},
+		condQuery: apis.ConditionReady,
+		want:      &condReady,
+	}, {
+		name: "unknown condition",
+		cs: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						condReady,
+						condDispatcherNotReady,
+					},
+				},
+			},
+		},
+		condQuery: apis.ConditionType("foo"),
+		want:      nil,
+	}}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := test.cs.GetCondition(test.condQuery)
+			if diff := cmp.Diff(test.want, got); diff != "" {
+				t.Errorf("unexpected condition (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestChannelInitializeConditions(t *testing.T) {
+	tests := []struct {
+		name string
+		cs   *KafkaChannelStatus
+		want *KafkaChannelStatus
+	}{{
+		name: "empty",
+		cs:   &KafkaChannelStatus{},
+		want: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{{
+						Type:   KafkaChannelConditionAddressable,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionChannelServiceReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionConfigReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionDispatcherReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionEndpointsReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionServiceReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionTopicReady,
+						Status: corev1.ConditionUnknown,
+					}},
+				},
+			},
+		},
+	}, {
+		name: "one false",
+		cs: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{{
+						Type:   KafkaChannelConditionDispatcherReady,
+						Status: corev1.ConditionFalse,
+					}},
+				},
+			},
+		},
+		want: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{{
+						Type:   KafkaChannelConditionAddressable,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionChannelServiceReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionConfigReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionDispatcherReady,
+						Status: corev1.ConditionFalse,
+					}, {
+						Type:   KafkaChannelConditionEndpointsReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionServiceReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionTopicReady,
+						Status: corev1.ConditionUnknown,
+					}},
+				},
+			},
+		},
+	}, {
+		name: "one true",
+		cs: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{{
+						Type:   KafkaChannelConditionDispatcherReady,
+						Status: corev1.ConditionTrue,
+					}},
+				},
+			},
+		},
+		want: &KafkaChannelStatus{
+			ChannelableStatus: eventingduckv1.ChannelableStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{{
+						Type:   KafkaChannelConditionAddressable,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionChannelServiceReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionConfigReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionDispatcherReady,
+						Status: corev1.ConditionTrue,
+					}, {
+						Type:   KafkaChannelConditionEndpointsReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionServiceReady,
+						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   KafkaChannelConditionTopicReady,
+						Status: corev1.ConditionUnknown,
+					}},
+				},
+			},
+		},
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			test.cs.InitializeConditions()
+			if diff := cmp.Diff(test.want, test.cs, ignoreAllButTypeAndStatus); diff != "" {
+				t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestChannelIsReady(t *testing.T) {
+	tests := []struct {
+		name                    string
+		markServiceReady        bool
+		markChannelServiceReady bool
+		markConfigurationReady  bool
+		setAddress              bool
+		markEndpointsReady      bool
+		markTopicReady          bool
+		wantReady               bool
+		dispatcherStatus        *appsv1.DeploymentStatus
+	}{{
+		name:                    "all happy",
+		markServiceReady:        true,
+		markChannelServiceReady: true,
+		markConfigurationReady:  true,
+		markEndpointsReady:      true,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              true,
+		markTopicReady:          true,
+		wantReady:               true,
+	}, {
+		name:                    "service not ready",
+		markServiceReady:        false,
+		markChannelServiceReady: false,
+		markConfigurationReady:  true,
+		markEndpointsReady:      true,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              true,
+		markTopicReady:          true,
+		wantReady:               false,
+	}, {
+		name:                    "endpoints not ready",
+		markServiceReady:        true,
+		markChannelServiceReady: false,
+		markConfigurationReady:  true,
+		markEndpointsReady:      false,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              true,
+		markTopicReady:          true,
+		wantReady:               false,
+	}, {
+		name:                    "deployment not ready",
+		markServiceReady:        true,
+		markConfigurationReady:  true,
+		markEndpointsReady:      true,
+		markChannelServiceReady: false,
+		dispatcherStatus:        deploymentStatusNotReady,
+		setAddress:              true,
+		markTopicReady:          true,
+		wantReady:               false,
+	}, {
+		name:                    "address not set",
+		markServiceReady:        true,
+		markConfigurationReady:  true,
+		markChannelServiceReady: false,
+		markEndpointsReady:      true,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              false,
+		markTopicReady:          true,
+		wantReady:               false,
+	}, {
+		name:                    "channel service not ready",
+		markServiceReady:        true,
+		markConfigurationReady:  true,
+		markChannelServiceReady: false,
+		markEndpointsReady:      true,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              true,
+		markTopicReady:          true,
+		wantReady:               false,
+	}, {
+		name:                    "topic not ready",
+		markServiceReady:        true,
+		markConfigurationReady:  true,
+		markChannelServiceReady: true,
+		markEndpointsReady:      true,
+		dispatcherStatus:        deploymentStatusReady,
+		setAddress:              true,
+		markTopicReady:          false,
+		wantReady:               false,
+	}}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cs := &KafkaChannelStatus{}
+			cs.InitializeConditions()
+			if test.markServiceReady {
+				cs.MarkServiceTrue()
+			} else {
+				cs.MarkServiceFailed("NotReadyService", "testing")
+			}
+			if test.markChannelServiceReady {
+				cs.MarkChannelServiceTrue()
+			} else {
+				cs.MarkChannelServiceFailed("NotReadyChannelService", "testing")
+			}
+			if test.markConfigurationReady {
+				cs.MarkConfigTrue()
+			} else {
+				cs.MarkConfigFailed("NotReadyConfiguration", "testing")
+			}
+			if test.setAddress {
+				cs.SetAddress(&apis.URL{Scheme: "http", Host: "foo.bar"})
+			}
+			if test.markEndpointsReady {
+				cs.MarkEndpointsTrue()
+			} else {
+				cs.MarkEndpointsFailed("NotReadyEndpoints", "testing")
+			}
+			if test.dispatcherStatus != nil {
+				cs.PropagateDispatcherStatus(test.dispatcherStatus)
+			} else {
+				cs.MarkDispatcherFailed("NotReadyDispatcher", "testing")
+			}
+			if test.markTopicReady {
+				cs.MarkTopicTrue()
+			} else {
+				cs.MarkTopicFailed("NotReadyTopic", "testing")
+			}
+			got := cs.IsReady()
+			if test.wantReady != got {
+				t.Errorf("unexpected readiness: want %v, got %v", test.wantReady, got)
+			}
+		})
+	}
+}
+
+func TestKafkaChannelStatus_SetAddressable(t *testing.T) {
+	testCases := map[string]struct {
+		url  *apis.URL
+		want *KafkaChannelStatus
+	}{
+		"empty string": {
+			want: &KafkaChannelStatus{
+				ChannelableStatus: eventingduckv1.ChannelableStatus{
+					Status: duckv1.Status{
+						Conditions: []apis.Condition{
+							{
+								Type:   KafkaChannelConditionAddressable,
+								Status: corev1.ConditionFalse,
+							},
+							// Note that Ready is here because when the condition is marked False, duck
+							// automatically sets Ready to false.
+							{
+								Type:   KafkaChannelConditionReady,
+								Status: corev1.ConditionFalse,
+							},
+						},
+					},
+					AddressStatus: duckv1.AddressStatus{Address: &duckv1.Addressable{}},
+				},
+			},
+		},
+		"has domain": {
+			url: &apis.URL{Scheme: "http", Host: "test-domain"},
+			want: &KafkaChannelStatus{
+				ChannelableStatus: eventingduckv1.ChannelableStatus{
+					AddressStatus: duckv1.AddressStatus{
+						Address: &duckv1.Addressable{
+							URL: &apis.URL{
+								Scheme: "http",
+								Host:   "test-domain",
+							},
+						},
+					},
+					Status: duckv1.Status{
+						Conditions: []apis.Condition{{
+							Type:   KafkaChannelConditionAddressable,
+							Status: corev1.ConditionTrue,
+						}, {
+							// Ready unknown comes from other dependent conditions via MarkTrue.
+							Type:   KafkaChannelConditionReady,
+							Status: corev1.ConditionUnknown,
+						}},
+					},
+				},
+			},
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			cs := &KafkaChannelStatus{}
+			cs.SetAddress(tc.url)
+			if diff := cmp.Diff(tc.want, cs, ignoreAllButTypeAndStatus); diff != "" {
+				t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
new file mode 100644
index 000000000..934a30c20
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+func TestKafkaChannelGetGroupVersionKind(t *testing.T) {
+	ch := KafkaChannel{}
+	gvk := ch.GetGroupVersionKind()
+
+	if gvk.Kind != "KafkaChannel" {
+		t.Errorf("Should be 'KafkaChannel'.")
+	}
+}
+
+func TestKafkaChannelGetStatus(t *testing.T) {
+	status := &duckv1.Status{}
+	config := KafkaChannel{
+		Status: KafkaChannelStatus{
+			ChannelableStatus: eventingduck.ChannelableStatus{
+				Status: *status,
+			},
+		},
+	}
+
+	if !cmp.Equal(config.GetStatus(), status) {
+		t.Errorf("GetStatus did not retrieve status. Got=%v Want=%v", config.GetStatus(), status)
+	}
+}
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
new file mode 100644
index 000000000..f8adaef02
--- /dev/null
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v1beta1
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/webhook/resourcesemantics"
+
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/apis/eventing"
+	"knative.dev/pkg/apis"
+)
+
+func TestKafkaChannelValidation(t *testing.T) {
+
+	testCases := map[string]struct {
+		cr   resourcesemantics.GenericCRD
+		want *apis.FieldError
+	}{
+		"empty spec": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrInvalidValue(0, "spec.numPartitions")
+				errs = errs.Also(fe)
+				fe = apis.ErrInvalidValue(0, "spec.replicationFactor")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		"negative numPartitions": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     -10,
+					ReplicationFactor: 1,
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue(-10, "spec.numPartitions")
+				return fe
+			}(),
+		},
+		"negative replicationFactor": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: -10,
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue(-10, "spec.replicationFactor")
+				return fe
+			}(),
+		},
+		"valid subscribers array": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					ChannelableSpec: eventingduck.ChannelableSpec{
+						SubscribableSpec: eventingduck.SubscribableSpec{
+							Subscribers: []eventingduck.SubscriberSpec{{
+								SubscriberURI: apis.HTTP("subscriberendpoint"),
+								ReplyURI:      apis.HTTP("resultendpoint"),
+							}},
+						}},
+				},
+			},
+			want: nil,
+		},
+		"empty subscriber at index 1": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					ChannelableSpec: eventingduck.ChannelableSpec{
+						SubscribableSpec: eventingduck.SubscribableSpec{
+							Subscribers: []eventingduck.SubscriberSpec{{
+								SubscriberURI: apis.HTTP("subscriberendpoint"),
+								ReplyURI:      apis.HTTP("replyendpoint"),
+							}, {}},
+						}},
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrMissingField("spec.subscribable.subscriber[1].replyURI", "spec.subscribable.subscriber[1].subscriberURI")
+				fe.Details = "expected at least one of, got none"
+				return fe
+			}(),
+		},
+		"two empty subscribers": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					ChannelableSpec: eventingduck.ChannelableSpec{
+						SubscribableSpec: eventingduck.SubscribableSpec{
+							Subscribers: []eventingduck.SubscriberSpec{{}, {}},
+						},
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrMissingField("spec.subscribable.subscriber[0].replyURI", "spec.subscribable.subscriber[0].subscriberURI")
+				fe.Details = "expected at least one of, got none"
+				errs = errs.Also(fe)
+				fe = apis.ErrMissingField("spec.subscribable.subscriber[1].replyURI", "spec.subscribable.subscriber[1].subscriberURI")
+				fe.Details = "expected at least one of, got none"
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		"invalid scope annotation": {
+			cr: &KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						eventing.ScopeAnnotationKey: "notvalid",
+					},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue("notvalid", "metadata.annotations.[eventing.knative.dev/scope]")
+				fe.Details = "expected either 'cluster' or 'namespace'"
+				return fe
+			}(),
+		},
+	}
+
+	for n, test := range testCases {
+		t.Run(n, func(t *testing.T) {
+			got := test.cr.Validate(context.Background())
+			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+				t.Errorf("%s: validate (-want, +got) = %v", n, diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/receiver/path_test.go b/control-plane/pkg/receiver/path_test.go
index 09f564377..b6df44cf9 100644
--- a/control-plane/pkg/receiver/path_test.go
+++ b/control-plane/pkg/receiver/path_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package receiver
 
 import (
diff --git a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
index 9ee01a2ac..5a3cb6bdc 100644
--- a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
+++ b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package testing
 
 import (
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/reconciler/kafka/topic_test.go
index b2884fa7c..804d12492 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/reconciler/kafka/topic_test.go
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package kafka
 
 import (
diff --git a/test/e2e_new/resources/kafkasink/kafkasink.go b/test/e2e_new/resources/kafkasink/kafkasink.go
index 410ef010c..0fa409894 100644
--- a/test/e2e_new/resources/kafkasink/kafkasink.go
+++ b/test/e2e_new/resources/kafkasink/kafkasink.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package kafkasink
 
diff --git a/test/e2e_new/resources/kafkasink/kafkasink_test.go b/test/e2e_new/resources/kafkasink/kafkasink_test.go
index f1f86e6f1..54427418a 100644
--- a/test/e2e_new/resources/kafkasink/kafkasink_test.go
+++ b/test/e2e_new/resources/kafkasink/kafkasink_test.go
@@ -1,18 +1,18 @@
 /*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 
 package kafkasink_test
 
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 69f3b9a43..d1379ae0f 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,7 +16,7 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 
 ---
 # Copyright 2018 The Knative Authors
@@ -39,14 +39,14 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -61,7 +61,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -76,7 +76,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -91,7 +91,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -106,7 +106,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -137,14 +137,14 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: pingsource-mt-adapter
@@ -175,14 +175,14 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -198,7 +198,7 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -213,7 +213,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -228,7 +228,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -259,7 +259,7 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 data:
   channelTemplateSpec: |
     apiVersion: messaging.knative.dev/v1
@@ -286,7 +286,7 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
   default-br-config: |
@@ -318,7 +318,7 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
   default-ch-config: |
@@ -351,7 +351,7 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
 data:
@@ -396,7 +396,7 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
 data:
@@ -428,7 +428,7 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
   annotations:
     knative.dev/example-checksum: "96896b00"
 data:
@@ -488,7 +488,7 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
 data:
@@ -540,7 +540,7 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -615,7 +615,7 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -676,7 +676,7 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/high-availability: "true"
 spec:
   selector:
@@ -686,7 +686,7 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20210819-4b18ab876"
+        eventing.knative.dev/release: "v20210822-128a666cc"
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
       affinity:
@@ -703,7 +703,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:19f4cf9147d6375e3b4b77aa30b7393d2aab540ca50667024f3a25ef5e4d51d9
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:b3c206aa166f04a2d95277a48bda031031d2f8d52e47a8345b2f4dd6062f71af
           resources:
             requests:
               cpu: 100m
@@ -721,7 +721,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:4ec4fbe7ab4b653dc9a54a6968ded39ba3bada0649bad85780a2b0e78144075d
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:b830315de32a5bdc260597163bf80a56e91ef3b615ae603892a8b695da4a0624
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -765,7 +765,7 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
   replicas: 0
@@ -777,7 +777,7 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20210819-4b18ab876"
+        eventing.knative.dev/release: "v20210822-128a666cc"
     spec:
       affinity:
         podAntiAffinity:
@@ -790,7 +790,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:836d121f16359079a19f4d44d4f8490e1a084be2b66c54106248c955b20ebe92
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:5d90606f646439bbb448208cf0229edf8dfdd2e4d6df20327701e10ea065e4f1
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -848,7 +848,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 spec:
   scaleTargetRef:
     apiVersion: apps/v1
@@ -871,7 +871,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 spec:
   minAvailable: 80%
   selector:
@@ -899,7 +899,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 spec:
   selector:
     matchLabels: &labels
@@ -926,7 +926,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:ade2d1269c29c934e4e7d72a2261ccd1367820d86cea38afda4fcb9a9ed00755
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:7f3a69806483c63d8cf776cc8d8c240c28516c2796ea753f0eeb2f4158f94105
           resources:
             requests:
               # taken from serving.
@@ -990,7 +990,7 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     role: eventing-webhook
   name: eventing-webhook
   namespace: knative-eventing
@@ -1022,7 +1022,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1236,7 +1236,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1398,7 +1398,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -1668,7 +1668,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1820,7 +1820,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -1947,7 +1947,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2200,7 +2200,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -2389,7 +2389,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2708,7 +2708,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -2898,7 +2898,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -3108,7 +3108,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -3249,6 +3249,9 @@ spec:
                       type:
                         description: 'Type of condition.'
                         type: string
+                deadLetterSinkUri:
+                  description: DeadLetterSinkURI is the resolved URI of the dead letter sink for this Trigger.
+                  type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
@@ -3287,7 +3290,7 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3299,7 +3302,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3317,7 +3320,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3338,7 +3341,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3363,7 +3366,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3382,7 +3385,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: messaging-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3403,7 +3406,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3439,7 +3442,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups:
       - ""
@@ -3464,7 +3467,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups:
       - ""
@@ -3480,7 +3483,7 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups:
       - ""
@@ -3512,7 +3515,7 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3524,7 +3527,7 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/channelable: "true"
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3561,7 +3564,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3573,7 +3576,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3585,7 +3588,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3597,7 +3600,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3609,7 +3612,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3622,7 +3625,7 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3634,7 +3637,7 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3660,7 +3663,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups:
       - ""
@@ -3794,7 +3797,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups:
       - ""
@@ -3860,7 +3863,7 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3872,7 +3875,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/podspecable: "true"
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -3918,7 +3921,7 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3930,7 +3933,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     duck.knative.dev/source: "true"
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -3966,7 +3969,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   - apiGroups:
       - ""
@@ -4051,7 +4054,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   # For watching logging configuration and getting certs.
   - apiGroups:
@@ -4149,7 +4152,7 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 rules:
   # For manipulating certs into secrets.
   - apiGroups:
@@ -4184,7 +4187,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4220,7 +4223,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4252,7 +4255,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4285,7 +4288,7 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 # The data is populated at install time.
 
 ---
@@ -4308,7 +4311,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 709ab872e..96c2936a1 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -231,7 +231,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -393,7 +393,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -663,7 +663,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -815,7 +815,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -942,7 +942,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1195,7 +1195,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1384,7 +1384,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1703,7 +1703,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -1893,7 +1893,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -2103,7 +2103,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210819-4b18ab876"
+    eventing.knative.dev/release: "v20210822-128a666cc"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -2244,6 +2244,9 @@ spec:
                       type:
                         description: 'Type of condition.'
                         type: string
+                deadLetterSinkUri:
+                  description: DeadLetterSinkURI is the resolved URI of the dead letter sink for this Trigger.
+                  type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer