Repository: knative-sandbox/eventing-kafka-broker
Pull request #163 (2020-09-10 15:49:48)

Check if data plane pods are running
Part of https://github.com/knative-sandbox/eventing-kafka-broker/issues/160

Since we're splitting the installation of the control plane and the
data plane to allow installing the Broker or the KafkaSink, we need
to make sure that the respective data plane is running, so check
is at least one pod of the data plane is running and reflect that
in the status.
The check is performed using a lister so the information might be
stale, but this allows avoid performing an API call against the
API server.

## Proposed Changes

- Check if data plane pods are running

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/163#" title="Author self-approved">pierDipi</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **grantr**
You can assign the PR to them by writing `/assign @grantr` 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/master/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":["grantr"]} -->
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/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/163/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1304085020821950467/artifacts/line-cov.html#file0) | 89.8% | 90.1% | 0.2
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/163/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1304085020821950467/artifacts/line-cov.html#file1) | 76.3% | 78.0% | 1.7

slinkydeveloper: /lgtm
/approve

Patch:
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index f548a5414..5dfe4c353 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -32,19 +32,26 @@ import (
 )
 
 const (
-	ConditionAddressable      apis.ConditionType = "Addressable"
-	ConditionTopicReady       apis.ConditionType = "TopicReady"
-	ConditionConfigMapUpdated apis.ConditionType = "ConfigMapUpdated"
-	ConditionConfigParsed     apis.ConditionType = "ConfigParsed"
+	ConditionAddressable        apis.ConditionType = "Addressable"
+	ConditionDataPlaneAvailable apis.ConditionType = "DataPlaneAvailable"
+	ConditionTopicReady         apis.ConditionType = "TopicReady"
+	ConditionConfigMapUpdated   apis.ConditionType = "ConfigMapUpdated"
+	ConditionConfigParsed       apis.ConditionType = "ConfigParsed"
 )
 
 var ConditionSet = apis.NewLivingConditionSet(
 	ConditionAddressable,
+	ConditionDataPlaneAvailable,
 	ConditionTopicReady,
 	ConditionConfigMapUpdated,
 	ConditionConfigParsed,
 )
 
+const (
+	ReasonDataPlaneNotAvailable  = "Data plane not available"
+	MessageDataPlaneNotAvailable = "Did you install the data plane for this component?"
+)
+
 type Object interface {
 	duckv1.KRShaped
 	runtime.Object
@@ -60,6 +67,21 @@ type StatusConditionManager struct {
 	Recorder record.EventRecorder
 }
 
+func (manager *StatusConditionManager) DataPlaneAvailable() {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(ConditionDataPlaneAvailable)
+}
+
+func (manager *StatusConditionManager) DataPlaneNotAvailable() reconciler.Event {
+
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
+		ConditionDataPlaneAvailable,
+		ReasonDataPlaneNotAvailable,
+		MessageDataPlaneNotAvailable,
+	)
+
+	return fmt.Errorf("%s: %s", ReasonDataPlaneNotAvailable, MessageDataPlaneNotAvailable)
+}
+
 func (manager *StatusConditionManager) FailedToGetConfigMap(err error) reconciler.Event {
 
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index cc3de06b4..1ba5ccd4d 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -69,6 +69,25 @@ type Reconciler struct {
 	ReceiverLabel   string
 }
 
+func (r *Reconciler) IsReceiverRunning() bool {
+	pods, err := r.PodLister.List(r.receiverSelector())
+	return err == nil && len(pods) > 0 && isAtLeastOneRunning(pods)
+}
+
+func (r *Reconciler) IsDispatcherRunning() bool {
+	pods, err := r.PodLister.List(r.dispatcherSelector())
+	return err == nil && len(pods) > 0 && isAtLeastOneRunning(pods)
+}
+
+func isAtLeastOneRunning(pods []*corev1.Pod) bool {
+	for _, pod := range pods {
+		if pod.Status.Phase == corev1.PodRunning {
+			return true
+		}
+	}
+	return false
+}
+
 func (r *Reconciler) GetOrCreateDataPlaneConfigMap() (*corev1.ConfigMap, error) {
 
 	cm, err := r.KubeClient.CoreV1().
@@ -172,7 +191,7 @@ func (r *Reconciler) UpdateDispatcherPodsAnnotation(logger *zap.Logger, volumeGe
 
 	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
 
-		labelSelector := labels.SelectorFromSet(map[string]string{"app": r.DispatcherLabel})
+		labelSelector := r.dispatcherSelector()
 		pods, errors := r.PodLister.Pods(r.SystemNamespace).List(labelSelector)
 		if errors != nil {
 			return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
@@ -186,7 +205,7 @@ func (r *Reconciler) UpdateReceiverPodsAnnotation(logger *zap.Logger, volumeGene
 
 	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
 
-		labelSelector := labels.SelectorFromSet(map[string]string{"app": r.ReceiverLabel})
+		labelSelector := r.receiverSelector()
 		pods, errors := r.PodLister.Pods(r.SystemNamespace).List(labelSelector)
 		if errors != nil {
 			return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
@@ -226,3 +245,11 @@ func (r *Reconciler) updatePodsAnnotation(logger *zap.Logger, component string,
 	}
 	return errors
 }
+
+func (r *Reconciler) receiverSelector() labels.Selector {
+	return labels.SelectorFromSet(map[string]string{"app": r.ReceiverLabel})
+}
+
+func (r *Reconciler) dispatcherSelector() labels.Selector {
+	return labels.SelectorFromSet(map[string]string{"app": r.DispatcherLabel})
+}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index d0ab19d2f..63ab25554 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -88,6 +88,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
+	if !r.IsReceiverRunning() || !r.IsDispatcherRunning() {
+		return statusConditionManager.DataPlaneNotAvailable()
+	}
+	statusConditionManager.DataPlaneAvailable()
+
 	topicConfig, err := r.topicConfig(logger, broker)
 	if err != nil {
 		return statusConditionManager.FailedToResolveConfig(err)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 843e84581..4f0ed15a5 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -65,6 +65,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
+	if !r.IsReceiverRunning() {
+		return statusConditionManager.DataPlaneNotAvailable()
+	}
+	statusConditionManager.DataPlaneAvailable()
+
 	if ks.Spec.NumPartitions != nil && ks.Spec.ReplicationFactor != nil {
 		topicConfig := &kafka.TopicConfig{
 			TopicDetail: sarama.TopicDetail{


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 4ef8b5f68..ae4b21a8d 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -142,6 +142,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerTopicReady,
 						BrokerAddressable(&configs),
@@ -199,6 +200,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						WithDelivery(),
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
 						BrokerAddressable(&configs),
@@ -213,6 +215,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 			Name: "Failed to create topic",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerReceiverPod(configs.SystemNamespace, nil),
+				BrokerDispatcherPod(configs.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -232,6 +236,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerFailedToCreateTopic,
 					),
@@ -294,8 +299,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						WithDelivery(),
-						BrokerConfigParsed,
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigParsed,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerTopicReady,
 						BrokerAddressable(&configs),
@@ -348,6 +354,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerConfigParsed,
 						BrokerTopicReady,
@@ -428,6 +435,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerConfigParsed,
 						BrokerTopicReady,
@@ -524,6 +532,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							}
 						},
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerConfigParsed,
 						BrokerTopicReady,
@@ -607,6 +616,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							broker.Spec.Delivery = &eventingduck.DeliverySpec{}
 						},
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerConfigParsed,
 						BrokerTopicReady,
@@ -683,6 +693,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerConfigParsed,
 						BrokerTopicReady,
@@ -752,6 +763,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							}
 						},
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerTopicReady,
 					),
@@ -769,6 +781,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Brokers:          []*coreconfig.Broker{},
 					VolumeGeneration: 1,
 				}, &configs),
+				BrokerReceiverPod(configs.SystemNamespace, nil),
+				BrokerDispatcherPod(configs.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -787,6 +801,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigNotParsed("no bootstrap.servers provided"),
 					),
 				},
@@ -849,6 +864,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						),
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerTopicReady,
 						BrokerAddressable(&configs),
@@ -871,6 +887,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						KReference(BrokerConfig(bootstrapServers, 20, 5)),
 					),
 				),
+				BrokerReceiverPod(configs.SystemNamespace, nil),
+				BrokerDispatcherPod(configs.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -892,6 +910,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							KReference(BrokerConfig(bootstrapServers, 20, 5)),
 						),
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigNotParsed(fmt.Sprintf(`failed to get configmap %s/%s: configmap %q not found`, ConfigMapNamespace, ConfigMapName, ConfigMapName)),
 					),
 				},
@@ -921,6 +940,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						Name:      BrokerName,
 					},
 				},
+				BrokerReceiverPod(configs.SystemNamespace, nil),
+				BrokerDispatcherPod(configs.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -945,6 +966,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							APIVersion: "v1",
 						}),
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigNotParsed(`supported config Kind: ConfigMap - got Pod`),
 					),
 				},
@@ -1049,6 +1071,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneAvailable,
 						BrokerConfigMapUpdatedReady(&configs),
 						BrokerConfigParsed,
 						BrokerTopicReady,
@@ -1060,6 +1083,33 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				BootstrapServersConfigMapKey: bootstrapServers,
 			},
 		},
+		{
+			Name: "no data plane pods running",
+			Objects: []runtime.Object{
+				NewBroker(),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf("%s: %s", base.ReasonDataPlaneNotAvailable, base.MessageDataPlaneNotAvailable),
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerDataPlaneNotAvailable,
+					),
+				},
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 854cf24a9..c53016052 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -125,6 +125,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				{
 					Object: NewSink(
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
@@ -181,6 +182,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 							sink.Spec.BootstrapServers = []string{"kafka-broker:10000"}
 						},
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers([]string{"kafka-broker:10000"}),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReadyWithName("my-topic-1"),
@@ -198,6 +200,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				NewSink(
 					BootstrapServers(bootstrapServersArr),
 				),
+				SinkReceiverPod(configs.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -217,6 +220,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				{
 					Object: NewSink(
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
 						SinkFailedToCreateTopic,
 					),
@@ -273,6 +277,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				{
 					Object: NewSink(
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
@@ -318,6 +323,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				{
 					Object: NewSink(
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
@@ -386,6 +392,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				{
 					Object: NewSink(
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
@@ -452,6 +459,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				{
 					Object: NewSink(
 						InitSinkConditions,
+						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
@@ -460,6 +468,33 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				},
 			},
 		},
+		{
+			Name: "no data plane pods running",
+			Objects: []runtime.Object{
+				NewSink(),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf("%s: %s", base.ReasonDataPlaneNotAvailable, base.MessageDataPlaneNotAvailable),
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						InitSinkConditions,
+						SinkDataPlaneNotAvailable,
+					),
+				},
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 5e8fb8db1..f921518ff 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -147,6 +147,18 @@ func BrokerTopicReady(broker *eventing.Broker) {
 	)
 }
 
+func BrokerDataPlaneAvailable(broker *eventing.Broker) {
+	broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrue(base.ConditionDataPlaneAvailable)
+}
+
+func BrokerDataPlaneNotAvailable(broker *eventing.Broker) {
+	broker.GetConditionSet().Manage(broker.GetStatus()).MarkFalse(
+		base.ConditionDataPlaneAvailable,
+		base.ReasonDataPlaneNotAvailable,
+		base.MessageDataPlaneNotAvailable,
+	)
+}
+
 func BrokerConfigParsed(broker *eventing.Broker) {
 	broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrue(base.ConditionConfigParsed)
 }
@@ -212,6 +224,9 @@ func BrokerDispatcherPod(namespace string, annotations map[string]string) runtim
 				"app": base.BrokerDispatcherLabel,
 			},
 		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
 	}
 }
 
@@ -229,6 +244,9 @@ func BrokerReceiverPod(namespace string, annotations map[string]string) runtime.
 				"app": base.BrokerReceiverLabel,
 			},
 		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/testing/objects_sink.go b/control-plane/pkg/reconciler/testing/objects_sink.go
index f9eb2b3d3..137d42a9a 100644
--- a/control-plane/pkg/reconciler/testing/objects_sink.go
+++ b/control-plane/pkg/reconciler/testing/objects_sink.go
@@ -117,6 +117,18 @@ func SinkTopicReady(sink *eventing.KafkaSink) {
 	SinkTopicReadyWithName(SinkTopic())(sink)
 }
 
+func SinkDataPlaneAvailable(sink *eventing.KafkaSink) {
+	sink.GetConditionSet().Manage(sink.GetStatus()).MarkTrue(base.ConditionDataPlaneAvailable)
+}
+
+func SinkDataPlaneNotAvailable(sink *eventing.KafkaSink) {
+	sink.GetConditionSet().Manage(sink.GetStatus()).MarkFalse(
+		base.ConditionDataPlaneAvailable,
+		base.ReasonDataPlaneNotAvailable,
+		base.MessageDataPlaneNotAvailable,
+	)
+}
+
 func SinkAddressable(configs *config.Env) func(sink *eventing.KafkaSink) {
 
 	return func(sink *eventing.KafkaSink) {
@@ -172,6 +184,9 @@ func SinkReceiverPod(namespace string, annotations map[string]string) runtime.Ob
 				"app": base.SinkReceiverLabel,
 			},
 		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
 	}
 }