Repository: knative-sandbox/eventing-kafka-broker
Pull request #2821 (2022-12-02 11:16:06)

[release-1.8] Handle ConsumerGroup finalization for deleted topics
This is an automated cherry-pick of #2816

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2821#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign odacremolbap for approval by writing `/assign @odacremolbap` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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/release-1.8/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":["odacremolbap"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2821](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (09b0ace) into [release-1.8](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/cd3751eb04a3e60a9c1f49f4e7dc693407c9bbf2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (cd3751e) will **decrease** coverage by `0.05%`.
> The diff coverage is `56.52%`.

```diff
@@               Coverage Diff               @@
##           release-1.8    #2821      +/-   ##
===============================================
- Coverage        55.19%   55.13%   -0.06%     
===============================================
  Files               75       75              
  Lines             8035     8080      +45     
===============================================
+ Hits              4435     4455      +20     
- Misses            3276     3297      +21     
- Partials           324      328       +4     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | `61.77% <0.00%> (-4.42%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `54.46% <62.50%> (+1.56%)` | :arrow_up: |
| [...l-plane/pkg/reconciler/consumergroup/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnRyb2xsZXIuZ28=) | `61.49% <90.00%> (+1.23%)` | :arrow_up: |
| [control-plane/pkg/apis/core/pod\_defaulting.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb3JlL3BvZF9kZWZhdWx0aW5nLmdv) | `87.75% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2821/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | `47.98% <100.00%> (-2.19%)` | :arrow_down: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)


Patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index a65de74d0..eb9602b5a 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -186,12 +186,11 @@ func (r Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.Consume
 	defer kafkaClusterAdminClient.Close()
 
 	groupId := cg.Spec.Template.Spec.Configs.Configs["group.id"]
-	if err := kafkaClusterAdminClient.DeleteConsumerGroup(groupId); err != nil && !errors.Is(sarama.ErrGroupIDNotFound, err) {
-		logging.FromContext(ctx).Errorw("unable to delete the consumer group", zap.String("id", groupId), zap.Error(err))
-		return err
+	if err = kafkaClusterAdminClient.DeleteConsumerGroup(groupId); err != nil && !errorIsOneOf(err, sarama.ErrUnknownTopicOrPartition, sarama.ErrGroupIDNotFound) {
+		return fmt.Errorf("unable to delete the consumer group %s: %w", groupId, err)
 	}
 
-	logging.FromContext(ctx).Infow("consumer group deleted", zap.String("id", groupId))
+	logging.FromContext(ctx).Debug("consumer group deleted", zap.String("id", groupId))
 	return nil
 }
 
@@ -652,6 +651,16 @@ func (r Reconciler) ensureContractConfigMapExists(ctx context.Context, p *corev1
 	return nil
 }
 
+func errorIsOneOf(err error, errs ...error) bool {
+	for _, e := range errs {
+		if errors.Is(err, e) {
+			return true
+		}
+	}
+
+	return false
+}
+
 var (
 	_ consumergroup.Interface = &Reconciler{}
 )


Test patch:
diff --git a/control-plane/pkg/kafka/testing/admin_mock.go b/control-plane/pkg/kafka/testing/admin_mock.go
index edb857529..5b7b2c096 100644
--- a/control-plane/pkg/kafka/testing/admin_mock.go
+++ b/control-plane/pkg/kafka/testing/admin_mock.go
@@ -26,6 +26,10 @@ import (
 
 var _ sarama.ClusterAdmin = &MockKafkaClusterAdmin{}
 
+const (
+	ErrorOnDeleteConsumerGroupTestKey = "error-on-delete-consumer-group"
+)
+
 type MockKafkaClusterAdmin struct {
 	// (Create|Delete)Topic
 	ExpectedTopicName string
@@ -45,6 +49,8 @@ type MockKafkaClusterAdmin struct {
 	ExpectedErrorOnDescribeTopics          error
 	ExpectedTopicsMetadataOnDescribeTopics []*sarama.TopicMetadata
 
+	ErrorOnDeleteConsumerGroup error
+
 	T *testing.T
 }
 
@@ -166,7 +172,7 @@ func (m *MockKafkaClusterAdmin) ListConsumerGroupOffsets(group string, topicPart
 }
 
 func (m *MockKafkaClusterAdmin) DeleteConsumerGroup(group string) error {
-	return nil
+	return m.ErrorOnDeleteConsumerGroup
 }
 
 func (m *MockKafkaClusterAdmin) DescribeCluster() (brokers []*sarama.Broker, controllerID int32, err error) {
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 19a161ecb..525c6a40f 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -90,13 +90,14 @@ var (
 )
 
 var DefaultEnv = &config.Env{
-	DataPlaneConfigMapNamespace: "knative-eventing",
-	ContractConfigMapName:       "kafka-broker-brokers-triggers",
-	GeneralConfigMapName:        "kafka-broker-config",
-	IngressName:                 "kafka-broker-ingress",
-	SystemNamespace:             "knative-eventing",
-	ContractConfigMapFormat:     base.Json,
-	DefaultBackoffDelayMs:       1000,
+	DataPlaneConfigMapNamespace:  "knative-eventing",
+	DataPlaneConfigConfigMapName: "config-kafka-broker-data-plane",
+	ContractConfigMapName:        "kafka-broker-brokers-triggers",
+	GeneralConfigMapName:         "kafka-broker-config",
+	IngressName:                  "kafka-broker-ingress",
+	SystemNamespace:              "knative-eventing",
+	ContractConfigMapFormat:      base.Json,
+	DefaultBackoffDelayMs:        1000,
 }
 
 func TestBrokerReconciler(t *testing.T) {
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
new file mode 100644
index 000000000..1c344963d
--- /dev/null
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -0,0 +1,425 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package broker_test // different package name due to import cycles. (broker -> testing -> broker)
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/utils/pointer"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	"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/kafka"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
+
+	"github.com/Shopify/sarama"
+	"github.com/manifestival/client-go-client"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/controller"
+	dynamicclientfake "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
+	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+)
+
+func TestNamespacedBrokerReconciler(t *testing.T) {
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
+
+	t.Parallel()
+
+	for _, f := range Formats {
+		namespacedBrokerReconciliation(t, f, *DefaultEnv)
+	}
+}
+
+func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env) {
+
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
+
+	env.ContractConfigMapFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewNamespacedBroker(
+					WithBrokerConfig(
+						KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
+					),
+				),
+				BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace)),
+				DataPlaneConfigMap(SystemNamespace, env.DataPlaneConfigConfigMapName, ConsumerConfigKey,
+					DataPlaneConfigInitialOffset(ConsumerConfigKey, sources.OffsetLatest),
+				),
+				reconcilertesting.NewConfigMap("config-tracing", SystemNamespace),
+				reconcilertesting.NewConfigMap("kafka-config-logging", SystemNamespace),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				NewService(),
+				BrokerReceiverPod(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPod(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				reconcilertesting.NewDeployment("kafka-broker-receiver", SystemNamespace),
+				reconcilertesting.NewDeployment("kafka-broker-dispatcher", SystemNamespace),
+				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
+				NewClusterRoleBinding("knative-kafka-broker-data-plane",
+					WithClusterRoleBindingSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+					WithClusterRoleBindingRoleRef("knative-kafka-broker-data-plane"),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantCreates: []runtime.Object{
+				ToManifestivalResource(t,
+					DataPlaneConfigMap(BrokerNamespace, env.DataPlaneConfigConfigMapName, ConsumerConfigKey,
+						DataPlaneConfigInitialOffset(ConsumerConfigKey, sources.OffsetLatest),
+					),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewConfigMap(
+						"config-tracing",
+						BrokerNamespace,
+					),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewConfigMap(
+						"kafka-config-logging",
+						BrokerNamespace,
+					),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewDeployment("kafka-broker-receiver", BrokerNamespace),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewDeployment("kafka-broker-dispatcher", BrokerNamespace),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					NewServiceAccount(BrokerNamespace, "knative-kafka-broker-data-plane"),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewService("kafka-broker-ingress", BrokerNamespace),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t, NewRoleBinding(BrokerNamespace, "knative-kafka-broker-data-plane",
+					WithRoleBindingSubjectServiceAccount(BrokerNamespace, "knative-kafka-broker-data-plane"),
+					WithRoleBindingClusterRoleRef("knative-kafka-broker-data-plane"),
+				),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(BrokerNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat,
+					&contract.Contract{
+						Resources: []*contract.Resource{
+							{
+								Uid:              BrokerUUID,
+								Topics:           []string{BrokerTopic()},
+								Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+								BootstrapServers: bootstrapServers,
+								Reference:        BrokerReference(),
+							},
+						},
+						Generation: 1,
+					},
+					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
+					WithConfigmapOwnerRef(&metav1.OwnerReference{
+						APIVersion:         eventing.SchemeGroupVersion.String(),
+						Kind:               "Broker",
+						Name:               BrokerName,
+						UID:                BrokerUUID,
+						Controller:         pointer.Bool(false),
+						BlockOwnerDeletion: pointer.Bool(true),
+					}),
+				),
+				BrokerReceiverPodUpdate(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewNamespacedBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						WithBrokerConfig(
+							KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
+						),
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusBrokerTopicReady,
+						NamespacedBrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerConfigMapAnnotations(),
+						WithTopicStatusAnnotation(BrokerTopic()),
+					),
+				},
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTableNamespaced(t, table, &env)
+}
+
+func TestNamespacedBrokerFinalizer(t *testing.T) {
+	t.Parallel()
+
+	for _, f := range Formats {
+		namespacedBrokerFinalization(t, f, *DefaultEnv)
+	}
+}
+
+func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
+
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
+
+	env.ContractConfigMapFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewDeletedBroker(reconcilertesting.WithBrokerClass(kafka.NamespacedBrokerClass)),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+						},
+					},
+					Generation: 1,
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTableNamespaced(t, table, &env)
+}
+
+func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
+
+	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		defaultTopicDetail := sarama.TopicDetail{
+			NumPartitions:     DefaultNumPartitions,
+			ReplicationFactor: DefaultReplicationFactor,
+		}
+
+		var onCreateTopicError error
+		if want, ok := row.OtherTestData[wantErrorOnCreateTopic]; ok {
+			onCreateTopicError = want.(error)
+		}
+
+		var onDeleteTopicError error
+		if want, ok := row.OtherTestData[wantErrorOnDeleteTopic]; ok {
+			onDeleteTopicError = want.(error)
+		}
+
+		expectedTopicDetail := defaultTopicDetail
+		if td, ok := row.OtherTestData[ExpectedTopicDetail]; ok {
+			expectedTopicDetail = td.(sarama.TopicDetail)
+		}
+
+		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+		if t, ok := row.OtherTestData[externalTopic]; ok {
+			expectedTopicName = t.(string)
+		}
+
+		var metadata []*sarama.TopicMetadata
+		metadata = append(metadata, &sarama.TopicMetadata{
+			Name:       ExternalTopicName,
+			IsInternal: false,
+			Partitions: []*sarama.PartitionMetadata{{}},
+		})
+
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
+		mfcMockClient, _ := client.NewUnsafeDynamicClient(dynamicclientfake.Get(ctx))
+
+		reconciler := &NamespacedReconciler{
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				SecretLister:                listers.GetSecretLister(),
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				ContractConfigMapName:       env.ContractConfigMapName,
+				ContractConfigMapFormat:     env.ContractConfigMapFormat,
+				DataPlaneNamespace:          env.SystemNamespace,
+				DispatcherLabel:             base.BrokerDispatcherLabel,
+				ReceiverLabel:               base.BrokerReceiverLabel,
+			},
+			ConfigMapLister:          listers.GetConfigMapLister(),
+			DeploymentLister:         listers.GetDeploymentLister(),
+			ServiceAccountLister:     listers.GetServiceAccountLister(),
+			ServiceLister:            listers.GetServiceLister(),
+			ClusterRoleBindingLister: listers.GetClusterRoleBindingLister(),
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopicName:                      expectedTopicName,
+					ExpectedTopicDetail:                    expectedTopicDetail,
+					ErrorOnCreateTopic:                     onCreateTopicError,
+					ErrorOnDeleteTopic:                     onDeleteTopicError,
+					ExpectedTopics:                         []string{expectedTopicName},
+					ExpectedTopicsMetadataOnDescribeTopics: metadata,
+					T:                                      t,
+				}, nil
+			},
+			Env:                env,
+			Prober:             proberMock,
+			ManifestivalClient: mfcMockClient,
+		}
+
+		reconciler.ConfigMapTracker = &FakeTracker{}
+		reconciler.SecretTracker = &FakeTracker{}
+
+		r := brokerreconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingclient.Get(ctx),
+			listers.GetBrokerLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+			kafka.NamespacedBrokerClass,
+		)
+
+		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
+		reconciler.IPsLister = prober.NewIPListerWithMapping()
+
+		return r
+	}))
+}
+
+func WithNamespacedBrokerOwnerRef(u *unstructured.Unstructured) {
+	refs := u.GetOwnerReferences()
+	if refs == nil {
+		refs = []metav1.OwnerReference{}
+	}
+	refs = append(refs, metav1.OwnerReference{
+		APIVersion:         eventing.SchemeGroupVersion.String(),
+		Kind:               "Broker",
+		Name:               BrokerName,
+		UID:                BrokerUUID,
+		Controller:         pointer.Bool(false),
+		BlockOwnerDeletion: pointer.Bool(true),
+	})
+	u.SetOwnerReferences(refs)
+}
+
+func WithNamespacedLabel(u *unstructured.Unstructured) {
+	labels := u.GetLabels()
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	labels[kafka.NamespacedBrokerDataplaneLabelKey] = kafka.NamespacedBrokerDataplaneLabelValue
+	u.SetLabels(labels)
+}
+
+func ToManifestivalResource(t *testing.T, obj runtime.Object, mutators ...UnstructuredMutator) runtime.Object {
+	m := func(u *unstructured.Unstructured) {
+		annotations := u.GetAnnotations()
+		if annotations == nil {
+			annotations = make(map[string]string)
+		}
+		annotations["manifestival"] = "new"
+		u.SetAnnotations(annotations)
+
+		annotations[corev1.LastAppliedConfigAnnotation] = lastApplied(u)
+		u.SetAnnotations(annotations)
+	}
+	mutators = append(mutators, m)
+	return ToUnstructured(t, obj, mutators...)
+}
+
+// lastApplied returns a JSON string denoting the resource's state
+func lastApplied(obj *unstructured.Unstructured) string {
+	ann := obj.GetAnnotations()
+	if len(ann) > 0 {
+		delete(ann, corev1.LastAppliedConfigAnnotation)
+		obj.SetAnnotations(ann)
+	}
+	bytes, _ := obj.MarshalJSON()
+	return string(bytes)
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index 287ec71e7..1df01666d 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -433,7 +433,7 @@ func TestReconcileKind(t *testing.T) {
 							{PodName: "p1", VReplicas: 1},
 							{PodName: "p2", VReplicas: 1},
 						}
-						cg.MarkReconcileConsumersFailedCondition(&apis.Condition{
+						_ = cg.MarkReconcileConsumersFailedCondition(&apis.Condition{
 							Type:    kafkainternals.ConsumerConditionBind,
 							Status:  corev1.ConditionFalse,
 							Reason:  "ConsumerBinding",
@@ -1995,11 +1995,161 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
 		},
-		//need to add support for DeleteConsumerGroup in admin_mock.go before adding more testcases
+		{
+			Name: "Finalize normal - failed consumer group deletion with " + sarama.ErrUnknownTopicOrPartition.Error(),
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrUnknownTopicOrPartition,
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
+		{
+			Name: "Finalize normal - failed consumer group deletion with " + sarama.ErrGroupIDNotFound.Error(),
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrGroupIDNotFound,
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
+		{
+			Name: "Finalize error - failed consumer group deletion with " + sarama.ErrClusterAuthorizationFailed.Error(),
+			Objects: []runtime.Object{
+				NewConsumer(1,
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics("t1", "t2"),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: systemNamespace}),
+						ConsumerSubscriber(NewSourceSinkReference()),
+					)),
+				),
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIdConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			WantErr: true,
+			Key:     testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrClusterAuthorizationFailed,
+			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace: ConsumerGroupNamespace,
+						Resource: schema.GroupVersionResource{
+							Group:    kafkainternals.SchemeGroupVersion.Group,
+							Version:  kafkainternals.SchemeGroupVersion.Version,
+							Resource: "consumers",
+						},
+					},
+					Name: fmt.Sprintf("%s-%d", ConsumerNamePrefix, 1),
+				},
+			},
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"unable to delete the consumer group my.group.id: "+sarama.ErrClusterAuthorizationFailed.Error(),
+				),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
 	}
 
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
+		errorOnDeleteKafkaCG := row.OtherTestData[kafkatesting.ErrorOnDeleteConsumerGroupTestKey]
+
 		r := &Reconciler{
 			SchedulerFunc: func(s string) Scheduler {
 				ss := row.OtherTestData[testSchedulerKey].(scheduler.Scheduler)
@@ -2017,7 +2167,8 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
-					T: t,
+					T:                          t,
+					ErrorOnDeleteConsumerGroup: ErrorAssertOrNil(errorOnDeleteKafkaCG),
 				}, nil
 			},
 			KafkaFeatureFlags: configapis.DefaultFeaturesConfig(),
diff --git a/control-plane/pkg/reconciler/testing/errors.go b/control-plane/pkg/reconciler/testing/errors.go
new file mode 100644
index 000000000..2ed85491b
--- /dev/null
+++ b/control-plane/pkg/reconciler/testing/errors.go
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2022 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package testing
+
+// ErrorAssertOrNil returns an error given an interface{} variable
+func ErrorAssertOrNil(e interface{}) error {
+	if e == nil {
+		return nil
+	}
+	return e.(error)
+}
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index 4e7061638..02e52d45c 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -66,7 +66,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 
 		ctx, dynamicClient := fakedynamicclient.With(ctx,
 			newScheme(),
-			ToUnstructured(t, row.Objects)...,
+			ToUnstructuredList(t, row.Objects)...,
 		)
 
 		dynamicScheme := runtime.NewScheme()
@@ -118,36 +118,48 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 	}
 }
 
-// ToUnstructured takes a list of k8s resources and converts them to
+// ToUnstructuredList takes a list of k8s resources and converts them to
 // Unstructured objects.
 // We must pass objects as Unstructured to the dynamic client fake, or it
 // won't handle them properly.
-func ToUnstructured(t *testing.T, objs []runtime.Object) (us []runtime.Object) {
-	sch := newScheme()
+func ToUnstructuredList(t *testing.T, objs []runtime.Object) (us []runtime.Object) {
 	for _, obj := range objs {
-		obj = obj.DeepCopyObject() // Don't mess with the primary copy
-		// Determine and set the TypeMeta for this object based on our test scheme.
-		gvks, _, err := sch.ObjectKinds(obj)
-		if err != nil {
-			t.Fatalf("Unable to determine kind for type: %v", err)
-		}
-		apiv, k := gvks[0].ToAPIVersionAndKind()
-		ta, err := meta.TypeAccessor(obj)
-		if err != nil {
-			t.Fatalf("Unable to create type accessor: %v", err)
-		}
-		ta.SetAPIVersion(apiv)
-		ta.SetKind(k)
-
-		b, err := json.Marshal(obj)
-		if err != nil {
-			t.Fatalf("Unable to marshal: %v", err)
-		}
-		u := &unstructured.Unstructured{}
-		if err := json.Unmarshal(b, u); err != nil {
-			t.Fatalf("Unable to unmarshal: %v", err)
-		}
+		u := ToUnstructured(t, obj)
 		us = append(us, u)
 	}
 	return
 }
+
+type UnstructuredMutator func(u *unstructured.Unstructured)
+
+// ToUnstructured takes a single k8s resource and converts it to
+// Unstructured object.
+func ToUnstructured(t *testing.T, obj runtime.Object, mutators ...UnstructuredMutator) runtime.Object {
+	sch := newScheme()
+	obj = obj.DeepCopyObject() // Don't mess with the primary copy
+	// Determine and set the TypeMeta for this object based on our test scheme.
+	gvks, _, err := sch.ObjectKinds(obj)
+	if err != nil {
+		t.Fatalf("Unable to determine kind for type: %v", err)
+	}
+	apiv, k := gvks[0].ToAPIVersionAndKind()
+	ta, err := meta.TypeAccessor(obj)
+	if err != nil {
+		t.Fatalf("Unable to create type accessor: %v", err)
+	}
+	ta.SetAPIVersion(apiv)
+	ta.SetKind(k)
+
+	b, err := json.Marshal(obj)
+	if err != nil {
+		t.Fatalf("Unable to marshal: %v", err)
+	}
+	u := &unstructured.Unstructured{}
+	if err := json.Unmarshal(b, u); err != nil {
+		t.Fatalf("Unable to unmarshal: %v", err)
+	}
+	for _, m := range mutators {
+		m(u)
+	}
+	return u
+}
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 31a261c78..7169afed7 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -17,11 +17,15 @@
 package testing
 
 import (
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	fakeapiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	appslisters "k8s.io/client-go/listers/apps/v1"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	rbaclisters "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
@@ -114,6 +118,10 @@ func (l *Listers) GetPodLister() corelisters.PodLister {
 	return corelisters.NewPodLister(l.indexerFor(&corev1.Pod{}))
 }
 
+func (l *Listers) GetDeploymentLister() appslisters.DeploymentLister {
+	return appslisters.NewDeploymentLister(l.indexerFor(&appsv1.Deployment{}))
+}
+
 func (l *Listers) GetSecretLister() corelisters.SecretLister {
 	return corelisters.NewSecretLister(l.indexerFor(&corev1.Secret{}))
 }
@@ -134,6 +142,10 @@ func (l *Listers) GetServiceLister() corelisters.ServiceLister {
 	return corelisters.NewServiceLister(l.indexerFor(&corev1.Service{}))
 }
 
+func (l *Listers) GetServiceAccountLister() corelisters.ServiceAccountLister {
+	return corelisters.NewServiceAccountLister(l.indexerFor(&corev1.ServiceAccount{}))
+}
+
 func (l *Listers) GetKafkaSinkLister() eventingkafkabrokerlisters.KafkaSinkLister {
 	return eventingkafkabrokerlisters.NewKafkaSinkLister(l.indexerFor(&eventingkafkabroker.KafkaSink{}))
 }
@@ -154,6 +166,10 @@ func (l *Listers) GetConsumerLister() consumerlisters.ConsumerLister {
 	return consumerlisters.NewConsumerLister(l.indexerFor(&eventingkafkabrokerconsumer.Consumer{}))
 }
 
+func (l *Listers) GetClusterRoleBindingLister() rbaclisters.ClusterRoleBindingLister {
+	return rbaclisters.NewClusterRoleBindingLister(l.indexerFor(&rbacv1.ClusterRoleBinding{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 51c821832..ad618f94a 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -192,6 +192,12 @@ func WithBootstrapServerStatusAnnotation(servers string) reconcilertesting.Broke
 
 type CMOption func(cm *corev1.ConfigMap)
 
+func WithConfigMapNamespace(namespace string) CMOption {
+	return func(cm *corev1.ConfigMap) {
+		cm.ObjectMeta.Namespace = namespace
+	}
+}
+
 func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int, options ...CMOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -272,17 +278,25 @@ func StatusBrokerConfigNotParsed(reason string) func(broker *eventing.Broker) {
 }
 
 func BrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
-
 	return func(broker *eventing.Broker) {
+		brokerAddressable(broker, env.IngressName, env.SystemNamespace)
+	}
+}
 
-		broker.Status.Address.URL = &apis.URL{
-			Scheme: "http",
-			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
-		}
+func NamespacedBrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		brokerAddressable(broker, env.IngressName, broker.Namespace)
+	}
+}
 
-		broker.GetConditionSet().Manage(&broker.Status).MarkTrue(base.ConditionAddressable)
+func brokerAddressable(broker *eventing.Broker, serviceName, serviceNamespace string) {
+	broker.Status.Address.URL = &apis.URL{
+		Scheme: "http",
+		Host:   network.GetServiceHostname(serviceName, serviceNamespace),
+		Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
 	}
+
+	broker.GetConditionSet().Manage(&broker.Status).MarkTrue(base.ConditionAddressable)
 }
 
 func BrokerReference() *contract.Reference {
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index b279d4f23..29d87e3b3 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -158,6 +158,66 @@ func NewRoleBinding(namespace string, name string, mutations ...func(account *rb
 	return s
 }
 
+func WithRoleBindingSubjectServiceAccount(saNamespace, saName string) func(account *rbacv1.RoleBinding) {
+	return func(cb *rbacv1.RoleBinding) {
+		cb.Subjects = []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: saNamespace,
+				Name:      saName,
+			},
+		}
+	}
+}
+
+func WithRoleBindingClusterRoleRef(roleName string) func(account *rbacv1.RoleBinding) {
+	return func(cb *rbacv1.RoleBinding) {
+		cb.RoleRef = rbacv1.RoleRef{
+			Kind:     "ClusterRole",
+			Name:     roleName,
+			APIGroup: "rbac.authorization.k8s.io",
+		}
+	}
+}
+
+func WithClusterRoleBindingSubjectServiceAccount(saNamespace, saName string) func(account *rbacv1.ClusterRoleBinding) {
+	return func(cb *rbacv1.ClusterRoleBinding) {
+		cb.Subjects = []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: saNamespace,
+				Name:      saName,
+			},
+		}
+	}
+}
+
+func WithClusterRoleBindingRoleRef(roleName string) func(account *rbacv1.ClusterRoleBinding) {
+	return func(cb *rbacv1.ClusterRoleBinding) {
+		cb.RoleRef = rbacv1.RoleRef{
+			Kind:     "ClusterRole",
+			Name:     roleName,
+			APIGroup: "rbac.authorization.k8s.io",
+		}
+	}
+}
+
+func NewClusterRoleBinding(name string, mutations ...func(account *rbacv1.ClusterRoleBinding)) *rbacv1.ClusterRoleBinding {
+	s := &rbacv1.ClusterRoleBinding{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ClusterRoleBinding",
+			APIVersion: rbacv1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+	for _, mut := range mutations {
+		mut(s)
+	}
+	return s
+}
+
 func NewConfigMapWithBinaryData(namespace string, name string, data []byte, options ...reconcilertesting.ConfigMapOption) runtime.Object {
 	return reconcilertesting.NewConfigMap(
 		name,
@@ -601,7 +661,7 @@ func PodAnnotations(annotations map[string]string) PodOption {
 	}
 }
 
-func DataPlaneConfigMap(namespace, name, key string, options ...base.ConfigMapOption) *corev1.ConfigMap {
+func DataPlaneConfigMap(namespace, name, key string, options ...reconcilertesting.ConfigMapOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: namespace,
@@ -619,7 +679,7 @@ func DataPlaneConfigMap(namespace, name, key string, options ...base.ConfigMapOp
 	return cm
 }
 
-func DataPlaneConfigInitialOffset(key string, offset sources.Offset) base.ConfigMapOption {
+func DataPlaneConfigInitialOffset(key string, offset sources.Offset) reconcilertesting.ConfigMapOption {
 	return func(cm *corev1.ConfigMap) {
 		props := properties.MustLoadString(cm.Data[key])
 		_, _, _ = props.Set("auto.offset.reset", string(offset))
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 4f776743a..da91ead8c 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -19,6 +19,8 @@ package testing
 import (
 	"time"
 
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -227,3 +229,9 @@ func NewDeletedConsumeGroup(opts ...ConsumerGroupOption) runtime.Object {
 func WithDeletedTimeStampConsumeGroup(cg *kafkainternals.ConsumerGroup) {
 	cg.GetObjectMeta().SetDeletionTimestamp(&metav1.Time{Time: time.Now()})
 }
+
+func WithConfigmapOwnerRef(ownerref *metav1.OwnerReference) reconcilertesting.ConfigMapOption {
+	return func(cg *corev1.ConfigMap) {
+		cg.ObjectMeta.OwnerReferences = []metav1.OwnerReference{*ownerref}
+	}
+}