Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Stateful set implementation of KafkaSource not working
**Describe the bug**
Upgrading eventing kafka components (controller, broker, source) from v1.5.8 to v1.6.9 doesn't seem to be working. The older deployment of KafkaSource is still running while the statefull set remains at 0/0 pods. When the stateful sets pod tried starting initially it warned `'MountVolume.Setup failed for volume "contract-resources" : configmap "kafka-source-dispatcher-0" not found`. 


After running the post-install job the logs showed 
`{"severity":"EMERGENCY","timestamp":"2022-10-25T19:33:19.923070308Z","logger":"kafka-broker-post-install","caller":"post-install/main.go:46","message":"source dispatcher deletion failed: failed while waiting for statefulset to come up: failed to get statefulset knative-eventing/kafka-source-dispatcher: statefulsets.apps \"kafka-source-dispatcher\" is forbidden: User \"system:serviceaccount:knative-eventing:knative-kafka-controller-post-install\" cannot get resource \"statefulsets\" in API group \"apps\" in the namespace \"knative-eventing\"","stacktrace":"main.main\n\tknative.dev/eventing-kafka-broker/control-plane/cmd/post-install/main.go:46\nruntime.main\n\truntime/proc.go:250"}`

 **Expected behavior**
I wasn't able to find much about this new implementation to know what should have happened, but I expected the KafkaSource stateful set to 'take over' the older KafkaSource deployment. I expected the post-install job to to remove the old kafkaSource implementation.

 **To Reproduce**
starting with v1.5.8 eventing kafka broker components (controller, broker, source), upgrade to 1.6.9 with the following commands:

`kubectl apply -f https://github.com/knative-sandbox/eventing-kafka-broker/releases/download/knative-v1.6.9/eventing-kafka-controller.yaml`

`kubectl apply -f https://github.com/knative-sandbox/eventing-kafka-broker/releases/download/knative-v1.6.9/eventing-kafka-source.yaml
`

`kubectl apply --filename https://github.com/knative-sandbox/eventing-kafka-broker/releases/download/knative-v1.6.9/eventing-kafka-broker.yaml
`

`https://github.com/knative-sandbox/eventing-kafka-broker/releases/download/knative-v1.6.9/eventing-kafka-post-install.yaml`

 **Knative release version**
1.6.9
 **Additional context**
Add any other context about the problem here such as proposed priority


Pull request description:
Create contract ConfigMaps in ConsumerGroup reconciler
Since we create the contract ConfigMaps in the Consumer reconciler
but the scheduler is called in the ConsumerGroup reconciler and it
fails with "not enough pod replicas" when the data plane pods are
in pending state.

This is not evident in our CI but it might be a signal of an edge
case caused by a certain sequence of events.

In this PR, I'm creating ConfigMaps before calling the scheduler
in the ConsumerGroup reconciler so that the pods can become
ready before the scheduler verifies which pods are available
to schedule vpods.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

Fixes #2750

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Create contract ConfigMaps in ConsumerGroup reconciler

<!--
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
Fix Stateful set implementation of KafkaSource not working, pods remain in pending phase
```


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/autoscaler/keda/keda.go b/control-plane/pkg/autoscaler/keda/keda.go
index 6760fde59..379d79b3d 100644
--- a/control-plane/pkg/autoscaler/keda/keda.go
+++ b/control-plane/pkg/autoscaler/keda/keda.go
@@ -22,6 +22,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	"knative.dev/pkg/kmeta"
 
@@ -206,21 +207,19 @@ func GenerateTriggerAuthentication(cg *kafkainternals.ConsumerGroup, secretData
 }
 
 func retrieveSaslTypeIfPresent(cg *kafkainternals.ConsumerGroup, secret corev1.Secret) *string {
-	var saslType string
-
 	if cg.Spec.Template.Spec.Auth.NetSpec != nil && cg.Spec.Template.Spec.Auth.NetSpec.SASL.Enable && cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef != nil {
 		secretKeyRefKey := cg.Spec.Template.Spec.Auth.NetSpec.SASL.Type.SecretKeyRef.Key
 		if saslTypeValue, ok := secret.Data[secretKeyRefKey]; ok {
-			saslType = string(saslTypeValue)
+			return pointer.String(string(saslTypeValue))
 		}
 	}
 
 	if cg.Spec.Template.Spec.Auth.SecretSpec != nil && cg.Spec.Template.Spec.Auth.SecretSpec.Ref != nil {
 		if saslTypeValue, ok := secret.Data[security.SaslType]; ok {
-			saslType = string(saslTypeValue)
+			return pointer.String(string(saslTypeValue))
 		}
 	}
-	return &saslType
+	return nil
 }
 
 func addAuthSecretTargetRef(parameter string, secretKeyRef v1beta1.SecretValueFromSource, secretTargetRefs []kedav1alpha1.AuthSecretTargetRef) []kedav1alpha1.AuthSecretTargetRef {
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 5c88214b8..596862cf7 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 82a43af0d..54f83c8d9 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -43,13 +43,14 @@ 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"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler/keda"
 	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"
 	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-
+	"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"
 )
@@ -59,13 +60,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
@@ -93,7 +101,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,7 +146,7 @@ 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.Int32(0)
-	err := r.schedule(cg) //de-schedule placements
+	err := r.schedule(ctx, cg) //de-schedule placements
 
 	if err != nil {
 		cg.Status.Placements = nil
@@ -295,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)
 	}
@@ -596,6 +613,54 @@ func (r *Reconciler) isKEDAEnabled(ctx context.Context, cg *kafkainternals.Consu
 	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{}
 	_ consumergroup.Finalizer = &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 69b538908..d7f5c3901 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -52,8 +52,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)
@@ -160,6 +161,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{
@@ -1555,12 +1643,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{},
@@ -1688,12 +1784,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{},
@@ -1921,12 +2022,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) {