Repository: knative-sandbox/eventing-kafka-broker
Pull request #1162 (2021-08-23 13:37:32)

Channel lifecycle
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Implement lifecycle for KafkaChannel
- Mostly taken from distributed ch impl
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1162#" title="Author self-approved">aliok</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **julz** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @julz` 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":["julz"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1162/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429799987436326912/artifacts/line-cov.html#file0) | 84.6% | 87.5% | 2.9

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1162/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1429802051205861376/artifacts/line-cov.html#file0) | 84.6% | 86.5% | 1.9

aliok: cc @devguyio @matzew 
Now you can comment about the lifecycle related stuff !
matzew: The discussion around doing this on the lifecycle:

```go
// 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()
}

```

I'd like to also separate in a new PR (upstream _and_ here). I will send one upstream first
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1162](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ff3873a) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/94fa04e9ba0d6480042d8c7842054d1f61161c59?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (94fa04e) will **increase** coverage by `4.07%`.
> The diff coverage is `n/a`.

> :exclamation: Current head ff3873a differs from pull request most recent head 1515eb8. Consider uploading reports for the commit 1515eb8 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1162      +/-   ##
============================================
+ Coverage     75.97%   80.04%   +4.07%     
  Complexity      460      460              
============================================
  Files            93       58      -35     
  Lines          3088     1654    -1434     
  Branches        136      136              
============================================
- Hits           2346     1324    -1022     
+ Misses          577      260     -317     
+ Partials        165       70      -95     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `80.04% <ø> (ø)` | |

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

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3ZhbGlkYXRpb24uZ28=) | | |
| [control-plane/pkg/config/env\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29uZmlnL2Vudl9jb25maWcuZ28=) | | |
| [.../pkg/apis/messaging/v1beta1/kafka\_channel\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9tZXNzYWdpbmcvdjFiZXRhMS9rYWZrYV9jaGFubmVsX3R5cGVzLmdv) | | |
| [control-plane/pkg/receiver/path.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvcGF0aC5nbw==) | | |
| [control-plane/pkg/kafka/consumer\_group\_lag.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvY29uc3VtZXJfZ3JvdXBfbGFnLmdv) | | |
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | | |
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| [control-plane/pkg/security/scram.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2NyYW0uZ28=) | | |
| ... and [24 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1162/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

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

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

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

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

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

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1162/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1430058078190964736/artifacts/line-cov.html#file0) | 84.6% | 78.0% | -6.6

matzew: I added them upstream too: https://github.com/knative-sandbox/eventing-kafka/pull/825  with tests and stuff

Since here we just get started, I think we can move forward and fix any other issues there (since eventually we reused the code directly from there, instead of copy-paste here)

/lgtm
/approve
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1162#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1162#issuecomment-904486033" title="Approved">matzew</a>*

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

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

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

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

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

Patch:
diff --git a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go
index 2eb14dde1..047c13079 100644
--- a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle.go
@@ -17,30 +17,88 @@
 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"
 )
 
-// ConditionAddressable is a reference to the Addressable condition in the base package.
-// KafkaChannel uses the common Condition set defined in the base package.
+var kc = apis.NewLivingConditionSet(
+	KafkaChannelConditionTopicReady,
+	KafkaChannelConditionDispatcherReady,
+	KafkaChannelConditionServiceReady,
+	KafkaChannelConditionEndpointsReady,
+	KafkaChannelConditionAddressable,
+	KafkaChannelConditionChannelServiceReady,
+	KafkaChannelConditionConfigReady)
+var channelCondSetLock = sync.RWMutex{}
+
 const (
-	ConditionAddressable apis.ConditionType = "Addressable"
-)
+	// KafkaChannelConditionReady has status True when all subconditions below have been set to True.
+	KafkaChannelConditionReady = apis.ConditionReady
 
-var conditionSet apis.ConditionSet
+	// 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"
 
-func RegisterConditionSet(cs apis.ConditionSet) {
-	conditionSet = cs
-}
+	// 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 {
-	return 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 {
-	return conditionSet
+	channelCondSetLock.RLock()
+	defer channelCondSetLock.RUnlock()
+
+	return kc
 }
 
 // GetCondition returns the condition currently associated with the given type, or nil.
@@ -65,9 +123,76 @@ func (cs *KafkaChannelStatus) SetAddress(url *apis.URL) {
 	}
 	if url != nil {
 		cs.Address.URL = url
-		cs.GetConditionSet().Manage(cs).MarkTrue(ConditionAddressable)
+		cs.GetConditionSet().Manage(cs).MarkTrue(KafkaChannelConditionAddressable)
 	} else {
 		cs.Address.URL = nil
-		cs.GetConditionSet().Manage(cs).MarkFalse(ConditionAddressable, "nil URL", "URL is 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...)
+}


Test patch:
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
index 57b1dca22..588c0aa99 100644
--- a/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go
+++ b/control-plane/pkg/apis/messaging/v1beta1/kafka_channel_lifecycle_test.go
@@ -19,47 +19,418 @@ package v1beta1
 import (
 	"testing"
 
-	"github.com/stretchr/testify/assert"
 	"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"
 )
 
-func TestKafkaChannelStatusSetAddress(t *testing.T) {
-	status := KafkaChannelStatus{}
-	status.SetAddress(apis.HTTP("localhost"))
+var condReady = apis.Condition{
+	Type:   KafkaChannelConditionReady,
+	Status: corev1.ConditionTrue,
+}
 
-	addressable := status.GetCondition(ConditionAddressable).IsTrue()
+var condDispatcherNotReady = apis.Condition{
+	Type:   KafkaChannelConditionDispatcherReady,
+	Status: corev1.ConditionFalse,
+}
 
-	assert.True(t, addressable)
+var deploymentConditionReady = appsv1.DeploymentCondition{
+	Type:   appsv1.DeploymentAvailable,
+	Status: corev1.ConditionTrue,
+}
 
-	status.SetAddress(nil)
+var deploymentConditionNotReady = appsv1.DeploymentCondition{
+	Type:   appsv1.DeploymentAvailable,
+	Status: corev1.ConditionFalse,
+}
 
-	notAddressable := status.GetCondition(ConditionAddressable).IsFalse()
+var deploymentStatusReady = &appsv1.DeploymentStatus{Conditions: []appsv1.DeploymentCondition{deploymentConditionReady}}
+var deploymentStatusNotReady = &appsv1.DeploymentStatus{Conditions: []appsv1.DeploymentCondition{deploymentConditionNotReady}}
 
-	assert.True(t, notAddressable)
-}
+var ignoreAllButTypeAndStatus = cmpopts.IgnoreFields(
+	apis.Condition{},
+	"LastTransitionTime", "Message", "Reason", "Severity")
 
 func TestKafkaChannelGetConditionSet(t *testing.T) {
+	r := &KafkaChannel{}
 
-	cs := (&KafkaChannel{}).GetConditionSet()
-
-	assert.Equal(t, conditionSet, cs)
+	if got, want := r.GetConditionSet().GetTopLevelConditionType(), apis.ConditionReady; got != want {
+		t.Errorf("GetTopLevelCondition=%v, want=%v", got, want)
+	}
 }
 
-func TestKafkaChannelStatusGetConditionSet(t *testing.T) {
-
-	cs := (&KafkaChannelStatus{}).GetConditionSet()
-
-	assert.Equal(t, conditionSet, cs)
+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 TestRegisterAlternateKafkaChannelConditionSet(t *testing.T) {
+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,
+					}},
+				},
+			},
+		},
+	}}
 
-	cs := apis.NewLivingConditionSet(apis.ConditionReady, "hello")
-
-	RegisterConditionSet(cs)
+	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)
+			}
+		})
+	}
+}
 
-	kc := KafkaChannel{}
+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)
+			}
+		})
+	}
+}
 
-	assert.Equal(t, cs, kc.GetConditionSet())
-	assert.Equal(t, cs, kc.Status.GetConditionSet())
+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)
+			}
+		})
+	}
 }