Repository: knative-sandbox/eventing-kafka-broker
Pull request #2806 (2022-11-24 14:12:53)

[release-1.8] Create contract ConfigMaps in ConsumerGroup reconciler (#2773)
Cherry-pick of https://github.com/knative-sandbox/eventing-kafka-broker/pull/2773

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2806#" title="Author self-approved">pierDipi</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/release-1.8/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.8 | a52897b6a4d8fb46e5095903f60cf19a776ec75d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2806/unit-tests_eventing-kafka-broker_release-1.8/1595782518617935872) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.8 | a52897b6a4d8fb46e5095903f60cf19a776ec75d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2806/reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.8/1595782519070920704) | true | `/test reconciler-tests-namespaced-broker`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2806#pullrequestreview-1193935340" title="LGTM">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2806#" title="Author self-approved">pierDipi</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/release-1.8/OWNERS)~~ [pierDipi]

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/core/pod_defaulting.go b/control-plane/pkg/apis/core/pod_defaulting.go
index 2ffaa586b..399d36359 100644
--- a/control-plane/pkg/apis/core/pod_defaulting.go
+++ b/control-plane/pkg/apis/core/pod_defaulting.go
@@ -28,7 +28,7 @@ import (
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
 
-	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
 // DispatcherPodsDefaulting returns the defaulting Callback for dispatcher pods
@@ -57,7 +57,7 @@ func podDefaultingCallback(generator generatorFunc) defaulting.CallbackFunc {
 		}
 
 		volume := &corev1.Volume{
-			Name: kafkainternals.DispatcherVolumeName,
+			Name: eventing.DispatcherVolumeName,
 			VolumeSource: corev1.VolumeSource{
 				ConfigMap: &corev1.ConfigMapVolumeSource{
 					LocalObjectReference: corev1.LocalObjectReference{Name: cmName},
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
index bbc185a6e..a2e822ceb 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
@@ -16,7 +16,29 @@
 
 package eventing
 
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
 const (
 	// ConfigMapVolumeName is the volume name of the data plane ConfigMap
 	ConfigMapVolumeName = "kafka-resources"
+
+	DispatcherVolumeName = "contract-resources"
 )
+
+func ConfigMapNameFromPod(p *corev1.Pod) (string, error) {
+	var vDp *corev1.Volume
+	for i, v := range p.Spec.Volumes {
+		if v.Name == DispatcherVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
+			vDp = &p.Spec.Volumes[i]
+			break
+		}
+	}
+	if vDp == nil {
+		return "", fmt.Errorf("failed to get data plane volume %s in pod %s/%s", ConfigMapVolumeName, p.GetNamespace(), p.GetName())
+	}
+	return vDp.ConfigMap.Name, nil
+}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index e447f8c55..68e3386a7 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -25,12 +25,14 @@ import (
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/utils/pointer"
 	"knative.dev/pkg/tracker"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -155,6 +157,25 @@ func (r *Reconciler) createDataPlaneConfigMap(ctx context.Context) (*corev1.Conf
 	return r.KubeClient.CoreV1().ConfigMaps(r.DataPlaneConfigMapNamespace).Create(ctx, cm, metav1.CreateOptions{})
 }
 
+func PodOwnerReference(p *corev1.Pod) ConfigMapOption {
+	expected := metav1.OwnerReference{
+		APIVersion:         corev1.SchemeGroupVersion.String(),
+		Kind:               "Pod",
+		Name:               p.Name,
+		UID:                p.UID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(true),
+	}
+	return func(cm *corev1.ConfigMap) {
+		for _, or := range cm.OwnerReferences {
+			if equality.Semantic.DeepDerivative(expected, or) {
+				return
+			}
+		}
+		cm.OwnerReferences = append(cm.OwnerReferences, expected)
+	}
+}
+
 // GetDataPlaneConfigMapData extracts contract from the given config map.
 func (r *Reconciler) GetDataPlaneConfigMapData(logger *zap.Logger, dataPlaneConfigMap *corev1.ConfigMap) (*contract.Contract, error) {
 	return GetDataPlaneConfigMapData(logger, dataPlaneConfigMap, r.ContractConfigMapFormat)
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index f85d5b56c..c103cd4c1 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -22,12 +22,9 @@ import (
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/utils/pointer"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
@@ -332,7 +329,7 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 	}
 
 	// Get contract associated with the pod.
-	cmName, err := cmNameFromPod(p, c)
+	cmName, err := eventing.ConfigMapNameFromPod(p)
 	if err != nil {
 		return false, err
 	}
@@ -383,41 +380,7 @@ func (r Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconcil
 		DataPlaneNamespace:            p.GetNamespace(),
 		DispatcherLabel:               "",
 		ReceiverLabel:                 "",
-		DataPlaneConfigMapTransformer: podOwnerReference(p),
-	}
-}
-
-func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
-	var vDp *corev1.Volume
-	for i, v := range p.Spec.Volumes {
-		if v.Name == kafkainternals.DispatcherVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
-			vDp = &p.Spec.Volumes[i]
-			break
-		}
-	}
-	if vDp == nil {
-		return "", fmt.Errorf("failed to get data plane volume %s in pod %s/%s", eventing.ConfigMapVolumeName, c.Spec.PodBind.PodNamespace, c.Spec.PodBind.PodName)
-	}
-	return vDp.ConfigMap.Name, nil
-}
-
-func podOwnerReference(p *corev1.Pod) base.ConfigMapOption {
-
-	expected := metav1.OwnerReference{
-		APIVersion:         corev1.SchemeGroupVersion.String(),
-		Kind:               "Pod",
-		Name:               p.Name,
-		UID:                p.UID,
-		Controller:         pointer.Bool(true),
-		BlockOwnerDeletion: pointer.Bool(true),
-	}
-	return func(cm *corev1.ConfigMap) {
-		for _, or := range cm.OwnerReferences {
-			if equality.Semantic.DeepDerivative(expected, or) {
-				return
-			}
-		}
-		cm.OwnerReferences = append(cm.OwnerReferences, expected)
+		DataPlaneConfigMapTransformer: base.PodOwnerReference(p),
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 58d1da7d6..a65de74d0 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -43,6 +43,7 @@ import (
 	"knative.dev/eventing/pkg/scheduler"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	internalv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned/typed/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumergroup"
@@ -50,7 +51,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
 
-	kedafunc "knative.dev/eventing-kafka-broker/control-plane/pkg/keda"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	kedav1alpha1 "knative.dev/eventing-kafka-broker/third_party/pkg/apis/keda/v1alpha1"
 	kedaclientset "knative.dev/eventing-kafka-broker/third_party/pkg/client/clientset/versioned"
 )
@@ -60,13 +61,20 @@ var (
 	ErrNoDeadLetterSinkURI = errors.New("no dead letter sink URI resolved")
 )
 
-type schedulerFunc func(s string) scheduler.Scheduler
+type Scheduler struct {
+	scheduler.Scheduler
+	SchedulerConfig
+}
+
+type schedulerFunc func(s string) Scheduler
 
 type Reconciler struct {
 	SchedulerFunc   schedulerFunc
 	ConsumerLister  kafkainternalslisters.ConsumerLister
 	InternalsClient internalv1alpha1.InternalV1alpha1Interface
 	SecretLister    corelisters.SecretLister
+	ConfigMapLister corelisters.ConfigMapLister
+	PodLister       corelisters.PodLister
 	KubeClient      kubernetes.Interface
 
 	NameGenerator names.NameGenerator
@@ -94,7 +102,7 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 		return cg.MarkInitializeOffsetFailed("InitializeOffset", err)
 	}
 
-	if err := r.schedule(cg); err != nil {
+	if err := r.schedule(ctx, cg); err != nil {
 		return err
 	}
 	cg.MarkScheduleSucceeded()
@@ -138,9 +146,8 @@ func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consum
 
 func (r Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
 
-	cg.Spec.Replicas = pointer.Int32Ptr(0)
-	err := r.schedule(cg) //de-schedule placements
-
+	cg.Spec.Replicas = pointer.Int32(0)
+	err := r.schedule(ctx, cg) //de-schedule placements
 	if err != nil {
 		cg.Status.Placements = nil
 
@@ -296,8 +303,17 @@ func (r Reconciler) finalizeConsumer(ctx context.Context, consumer *kafkainterna
 	return nil
 }
 
-func (r Reconciler) schedule(cg *kafkainternals.ConsumerGroup) error {
-	placements, err := r.SchedulerFunc(cg.GetUserFacingResourceRef().Kind).Schedule(cg)
+func (r Reconciler) schedule(ctx context.Context, cg *kafkainternals.ConsumerGroup) error {
+	statefulSetScheduler := r.SchedulerFunc(cg.GetUserFacingResourceRef().Kind)
+
+	// Ensure Contract configmaps are created before scheduling to avoid having pending pods due to missing
+	// volumes.
+	// See https://github.com/knative-sandbox/eventing-kafka-broker/issues/2750#issuecomment-1304244017
+	if err := r.ensureContractConfigmapsExist(ctx, statefulSetScheduler); err != nil {
+		return cg.MarkScheduleConsumerFailed("Schedule", err)
+	}
+
+	placements, err := statefulSetScheduler.Schedule(cg)
 	if err != nil {
 		return cg.MarkScheduleConsumerFailed("Schedule", err)
 	}
@@ -455,18 +471,18 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 			return err
 		}
 
-		triggerAuthentication, secret, err = kedafunc.GenerateTriggerAuthentication(cg, saslType)
+		triggerAuthentication, secret, err = keda.GenerateTriggerAuthentication(cg, saslType)
 		if err != nil {
 			return err
 		}
 	}
 
-	triggers, err := kedafunc.GenerateScaleTriggers(cg, triggerAuthentication)
+	triggers, err := keda.GenerateScaleTriggers(cg, triggerAuthentication)
 	if err != nil {
 		return err
 	}
 
-	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), kedafunc.GenerateScaleTarget(cg), triggers)
+	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), keda.GenerateScaleTarget(cg), triggers)
 	if err != nil {
 		return err
 	}
@@ -588,6 +604,54 @@ func (r *Reconciler) isKEDAEnabled(ctx context.Context, namespace string) bool {
 	return false
 }
 
+func (r Reconciler) ensureContractConfigmapsExist(ctx context.Context, scheduler Scheduler) error {
+	selector := labels.SelectorFromSet(map[string]string{"app": scheduler.StatefulSetName})
+	pods, err := r.PodLister.
+		Pods(r.SystemNamespace).
+		List(selector)
+	if err != nil {
+		return fmt.Errorf("failed to list statefulset pods with selector %v: %w", selector.String(), err)
+	}
+
+	for _, p := range pods {
+		cmName, err := eventing.ConfigMapNameFromPod(p)
+		if err != nil {
+			return err
+		}
+		if err := r.ensureContractConfigMapExists(ctx, p, cmName); err != nil {
+			return fmt.Errorf("failed to get ConfigMap %s/%s: %w", r.SystemNamespace, cmName, err)
+		}
+	}
+
+	return nil
+}
+
+func (r Reconciler) ensureContractConfigMapExists(ctx context.Context, p *corev1.Pod, name string) error {
+	// Check if ConfigMap exists in lister cache
+	_, err := r.ConfigMapLister.ConfigMaps(r.SystemNamespace).Get(name)
+	// ConfigMap already exists, return
+	if err == nil {
+		return nil
+	}
+
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to get ConfigMap %s/%s: %w", r.SystemNamespace, name, err)
+	}
+
+	b := base.Reconciler{
+		KubeClient:                    r.KubeClient,
+		DataPlaneConfigMapNamespace:   r.SystemNamespace,
+		ContractConfigMapName:         name,
+		DataPlaneNamespace:            r.SystemNamespace,
+		DataPlaneConfigMapTransformer: base.PodOwnerReference(p),
+	}
+
+	if _, err := b.GetOrCreateDataPlaneConfigMap(ctx); err != nil && apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("failed to create ConfigMap %s/%s: %w", r.SystemNamespace, name, err)
+	}
+	return nil
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 )
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index 938b206d8..1c5ddd364 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -37,7 +37,9 @@ import (
 	"knative.dev/eventing/pkg/scheduler"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
+	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	nodeinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/node"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -98,17 +100,19 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		DeSchedulerPolicy: schedulerPolicyFromConfigMapOrFail(ctx, env.DeSchedulerPolicyConfigMap),
 	}
 
-	schedulers := map[string]scheduler.Scheduler{
+	schedulers := map[string]Scheduler{
 		KafkaSourceScheduler: createKafkaScheduler(ctx, c, kafkainternals.SourceStatefulSetName),
 		//KafkaTriggerScheduler: createKafkaScheduler(ctx, c, kafkainternals.BrokerStatefulSetName), //To be added with trigger/v2 reconciler version only
 		//KafkaChannelScheduler: createKafkaScheduler(ctx, c, kafkainternals.ChannelStatefulSetName), //To be added with channel/v2 reconciler version only
 	}
 
 	r := &Reconciler{
-		SchedulerFunc:              func(s string) scheduler.Scheduler { return schedulers[strings.ToLower(s)] },
+		SchedulerFunc:              func(s string) Scheduler { return schedulers[strings.ToLower(s)] },
 		ConsumerLister:             consumer.Get(ctx).Lister(),
 		InternalsClient:            internalsclient.Get(ctx).InternalV1alpha1(),
 		SecretLister:               secretinformer.Get(ctx).Lister(),
+		ConfigMapLister:            configmapinformer.Get(ctx).Lister(),
+		PodLister:                  podinformer.Get(ctx).Lister(),
 		KubeClient:                 kubeclient.Get(ctx),
 		NameGenerator:              names.SimpleNameGenerator,
 		NewKafkaClient:             sarama.NewClient,
@@ -167,7 +171,7 @@ func enqueueConsumerGroupFromConsumer(enqueue func(name types.NamespacedName)) f
 	}
 }
 
-func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string) scheduler.Scheduler {
+func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string) Scheduler {
 	lister := consumergroup.Get(ctx).Lister()
 	return createStatefulSetScheduler(
 		ctx,
@@ -212,8 +216,8 @@ func getSelectorLabel(ssName string) map[string]string {
 	return selectorLabel
 }
 
-func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister scheduler.VPodLister) scheduler.Scheduler {
-	return statefulsetscheduler.NewScheduler(
+func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister scheduler.VPodLister) Scheduler {
+	ss := statefulsetscheduler.NewScheduler(
 		ctx,
 		system.Namespace(),
 		c.StatefulSetName,
@@ -226,6 +230,11 @@ func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister s
 		c.SchedulerPolicy,
 		c.DeSchedulerPolicy,
 	)
+
+	return Scheduler{
+		Scheduler:       ss,
+		SchedulerConfig: c,
+	}
 }
 
 // schedulerPolicyFromConfigMapOrFail reads predicates and priorities data from configMap


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index d1bbb2bb8..287ec71e7 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -53,8 +53,9 @@ import (
 	configapis "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 
-	kedaclient "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client/fake"
 	cm "knative.dev/pkg/configmap/testing"
+
+	kedaclient "knative.dev/eventing-kafka-broker/third_party/pkg/client/injection/client/fake"
 )
 
 type SchedulerFunc func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error)
@@ -161,6 +162,93 @@ func TestReconcileKind(t *testing.T) {
 				finalizerUpdatedEvent,
 			},
 		},
+		{
+			Name: "Consumers in multiple pods, with pods pending and unknown phase",
+			Objects: []runtime.Object{
+				NewService(),
+				NewDispatcherPod("p1", PodLabel(kafkainternals.SourceStatefulSetName), PodPending()),
+				NewDispatcherPod("p2", PodLabel(kafkainternals.SourceStatefulSetName)),
+				NewConsumerGroup(
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+					ConsumerGroupReplicas(2),
+					ConsumerForTrigger(),
+				),
+			},
+			Key: ConsumerGroupTestKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return []eventingduckv1alpha1.Placement{
+						{PodName: "p1", VReplicas: 1},
+						{PodName: "p2", VReplicas: 1},
+					}, nil
+				}),
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(systemNamespace, "p1", nil),
+				NewConfigMapWithBinaryData(systemNamespace, "p2", nil),
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+					)),
+				),
+				NewConsumer(2,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p2", PodNamespace: systemNamespace}),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: func() runtime.Object {
+						cg := NewConsumerGroup(
+							ConsumerGroupConsumerSpec(NewConsumerSpec(
+								ConsumerTopics("t1", "t2"),
+								ConsumerConfigs(
+									ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+									ConsumerGroupIdConfig("my.group.id"),
+								),
+							)),
+							ConsumerGroupReplicas(2),
+							ConsumerGroupStatusReplicas(0),
+							ConsumerForTrigger(),
+						)
+						cg.Status.Placements = []eventingduckv1alpha1.Placement{
+							{PodName: "p1", VReplicas: 1},
+							{PodName: "p2", VReplicas: 1},
+						}
+						_ = cg.MarkReconcileConsumersFailed("PropagateSubscriberURI", ErrNoSubscriberURI)
+						cg.MarkScheduleSucceeded()
+						cg.MarkAutoscalerDisabled() // KEDA not installed
+						return cg
+					}(),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			SkipNamespaceValidation: true,
+		},
 		{
 			Name: "Consumers in multiple pods, with auth spec, one exists - secret not found",
 			Objects: []runtime.Object{
@@ -1534,12 +1622,20 @@ func TestReconcileKind(t *testing.T) {
 		store.OnConfigChanged(exampleConfig)
 
 		r := Reconciler{
-			SchedulerFunc: func(s string) scheduler.Scheduler {
-				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			SchedulerFunc: func(s string) Scheduler {
+				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+				return Scheduler{
+					Scheduler: ss,
+					SchedulerConfig: SchedulerConfig{
+						StatefulSetName: kafkainternals.SourceStatefulSetName,
+					},
+				}
 			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
+			PodLister:       listers.GetPodLister(),
 			KubeClient:      kubeclient.Get(ctx),
 			KedaClient:      kedaclient.Get(ctx),
 			NameGenerator:   &CounterGenerator{},
@@ -1667,12 +1763,17 @@ func TestReconcileKindNoAutoscaler(t *testing.T) {
 		ctx, _ = kedaclient.With(ctx)
 
 		r := Reconciler{
-			SchedulerFunc: func(s string) scheduler.Scheduler {
-				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			SchedulerFunc: func(s string) Scheduler {
+				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+				return Scheduler{
+					Scheduler: ss,
+				}
 			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
+			PodLister:       listers.GetPodLister(),
 			KubeClient:      kubeclient.Get(ctx),
 			KedaClient:      kedaclient.Get(ctx),
 			NameGenerator:   &CounterGenerator{},
@@ -1900,12 +2001,17 @@ func TestFinalizeKind(t *testing.T) {
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		r := &Reconciler{
-			SchedulerFunc: func(s string) scheduler.Scheduler {
-				return row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+			SchedulerFunc: func(s string) Scheduler {
+				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
+				return Scheduler{
+					Scheduler: ss,
+				}
 			},
 			ConsumerLister:  listers.GetConsumerLister(),
 			InternalsClient: fakekafkainternalsclient.Get(ctx).InternalV1alpha1(),
 			SecretLister:    listers.GetSecretLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
+			PodLister:       listers.GetPodLister(),
 			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
 				return &kafkatesting.MockKafkaClient{}, nil
 			},
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index 80a1148de..dc148abaf 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -26,6 +26,7 @@ import (
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index abcb3f8a2..b279d4f23 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -40,7 +40,7 @@ import (
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
-	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	kafkaeventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
@@ -497,7 +497,7 @@ func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
 		},
 		Spec: corev1.PodSpec{
 			Volumes: []corev1.Volume{{
-				Name: kafkainternals.DispatcherVolumeName,
+				Name: kafkaeventing.DispatcherVolumeName,
 				VolumeSource: corev1.VolumeSource{
 					ConfigMap: &corev1.ConfigMapVolumeSource{
 						LocalObjectReference: corev1.LocalObjectReference{
@@ -516,6 +516,21 @@ func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
 	return p
 }
 
+func PodLabel(value string) PodOption {
+	return func(pod *corev1.Pod) {
+		if pod.Labels == nil {
+			pod.Labels = make(map[string]string, 2)
+		}
+		pod.Labels["app"] = value
+	}
+}
+
+func PodPending() PodOption {
+	return func(pod *corev1.Pod) {
+		pod.Status.Phase = corev1.PodPending
+	}
+}
+
 func DispatcherPodAsOwnerReference(name string) reconcilertesting.ConfigMapOption {
 	d := NewDispatcherPod(name)
 	return func(configMap *corev1.ConfigMap) {