Repository: knative-sandbox/eventing-kafka-broker
Pull request #2370 (2022-07-01 17:14:25)

[release-1.5] Adding finalizer for auth secret
This is an automated cherry-pick of #2346

```release-note

```

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/2370#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.5/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":["pierdipi"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2370?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2370](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2370?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (93abe3d) into [release-1.5](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/1d3b22a716fb9622a510e88189a5e86510ae89a2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1d3b22a) will **increase** coverage by `0.03%`.
> The diff coverage is `56.00%`.

```diff
@@               Coverage Diff               @@
##           release-1.5    #2370      +/-   ##
===============================================
+ Coverage        60.03%   60.06%   +0.03%     
===============================================
  Files               71       71              
  Lines             7096     7165      +69     
===============================================
+ Hits              4260     4304      +44     
- Misses            2527     2544      +17     
- Partials           309      317       +8     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2370?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/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2370/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | `71.73% <56.00%> (-1.57%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2370/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `70.91% <0.00%> (ø)` | |

------

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

knative-prow[bot]: @knative-prow-robot: 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.5 | 93abe3d9734c6da6376deb722ff094e990e00a62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2370/unit-tests_eventing-kafka-broker_release-1.5/1542919582140862464) | true | `/test unit-tests_eventing-kafka-broker_release-1.5`

[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 -->
knative-prow[bot]: @knative-prow-robot: 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.5 | 93abe3d9734c6da6376deb722ff094e990e00a62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2370/unit-tests_eventing-kafka-broker_release-1.5/1542925108035719168) | true | `/test unit-tests_eventing-kafka-broker_release-1.5`

[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 -->
knative-prow[bot]: @knative-prow-robot: The following tests **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.5 | 93abe3d9734c6da6376deb722ff094e990e00a62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2370/unit-tests_eventing-kafka-broker_release-1.5/1542925108035719168) | true | `/test unit-tests_eventing-kafka-broker_release-1.5`
integration-tests_eventing-kafka-broker_release-1.5 | 93abe3d9734c6da6376deb722ff094e990e00a62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2370/integration-tests_eventing-kafka-broker_release-1.5/1542919582199582720) | true | `/test integration-tests_eventing-kafka-broker_release-1.5`

[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 -->
matzew: /retest 
matzew: /lgtm
/approve
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/2370#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2370#issuecomment-1172685578" title="Approved">matzew</a>*

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

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

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

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

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

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index c34adcc4b..5f588a4ea 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -78,6 +78,7 @@ rules:
     verbs:
       - list
       - get
+      - update
       - watch
 
   # Scheduler permissions
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index dfeb9cbf9..02da4dda3 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -116,6 +116,10 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 			zap.String("namespace", secret.Namespace),
 			zap.String("kind", secret.Kind),
 		)
+
+		if err := r.addFinalizerSecret(ctx, finalizerSecret(broker), secret); err != nil {
+			return err
+		}
 	}
 
 	// get security option for Sarama with secret info in it
@@ -332,19 +336,28 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return controller.NewRequeueAfter(5 * time.Second)
 	}
 
+	_, brokerConfig, err := r.brokerConfigMap(logger, broker)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+
+	secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: brokerConfig, UseNamespaceInConfigmap: false}, r.SecretProviderFunc())
+	if err != nil {
+		return fmt.Errorf("failed to get secret: %w", err)
+	}
+
 	// External topics are not managed by the broker,
 	// therefore we do not delete them
 	_, externalTopic := isExternalTopic(broker)
 	if !externalTopic {
-		topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
+
+		// I do not like the use of `_`
+		// TODO: refactor
+		topicConfig, _, err := r.topicConfig(logger, broker)
 		if err != nil {
 			return fmt.Errorf("failed to resolve broker config: %w", err)
 		}
 
-		secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: brokerConfig, UseNamespaceInConfigmap: false}, r.SecretProviderFunc())
-		if err != nil {
-			return fmt.Errorf("failed to get secret: %w", err)
-		}
 		if secret != nil {
 			logger.Debug("Secret reference",
 				zap.String("apiVersion", secret.APIVersion),
@@ -359,7 +372,13 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		if err := r.finalizeNonExternalBrokerTopic(broker, securityOption, topicConfig, logger); err != nil {
 			return err
 		}
+
 	}
+
+	if err := r.removeFinalizerSecret(ctx, finalizerSecret(broker), secret); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -388,19 +407,23 @@ func (r *Reconciler) finalizeNonExternalBrokerTopic(broker *eventing.Broker, sec
 	return nil
 }
 
-func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*kafka.TopicConfig, *corev1.ConfigMap, error) {
+func (r *Reconciler) brokerNamespace(broker *eventing.Broker) string {
+	namespace := broker.Spec.Config.Namespace
+	if namespace == "" {
+		// Namespace not specified, use broker namespace.
+		namespace = broker.Namespace
+	}
+	return namespace
+}
 
+func (r *Reconciler) brokerConfigMap(logger *zap.Logger, broker *eventing.Broker) (bool, *corev1.ConfigMap, error) {
 	logger.Debug("broker config", zap.Any("broker.spec.config", broker.Spec.Config))
 
 	if strings.ToLower(broker.Spec.Config.Kind) != "configmap" { // TODO: is there any constant?
-		return nil, nil, fmt.Errorf("supported config Kind: ConfigMap - got %s", broker.Spec.Config.Kind)
+		return false, nil, fmt.Errorf("supported config Kind: ConfigMap - got %s", broker.Spec.Config.Kind)
 	}
 
-	namespace := broker.Spec.Config.Namespace
-	if namespace == "" {
-		// Namespace not specified, use broker namespace.
-		namespace = broker.Namespace
-	}
+	namespace := r.brokerNamespace(broker)
 
 	// There might be cases where the ConfigMap is deleted before the Broker.
 	// In these cases, we rebuild the ConfigMap from broker status annotations.
@@ -414,17 +437,30 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*
 	isRebuilt := false
 	cm, getCmError := r.ConfigMapLister.ConfigMaps(namespace).Get(broker.Spec.Config.Name)
 	if getCmError != nil && !apierrors.IsNotFound(getCmError) {
-		return nil, nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, broker.Spec.Config.Name, getCmError)
+		return isRebuilt, cm, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, broker.Spec.Config.Name, getCmError)
 	}
 	if apierrors.IsNotFound(getCmError) {
 		cm = rebuildCMFromAnnotations(broker)
 		isRebuilt = true
 	}
 
+	return isRebuilt, cm, getCmError
+}
+
+// TODO: pass the configmap to this function
+func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*kafka.TopicConfig, *corev1.ConfigMap, error) {
+
+	isRebuilt, cm, getCmError := r.brokerConfigMap(logger, broker)
+	// For generic errors, we return the error, when the ConfigMap wasn't found we try to get topic information from
+	// the rebuilt ConfigMap, if it's still not possible we return the `getCmError` down below.
+	if getCmError != nil && !apierrors.IsNotFound(getCmError) {
+		return nil, nil, getCmError
+	}
+
 	topicConfig, err := kafka.TopicConfigFromConfigMap(logger, cm)
 	if err != nil {
 		if isRebuilt {
-			return nil, cm, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, broker.Spec.Config.Name, getCmError)
+			return nil, cm, fmt.Errorf("failed to get configmap %s/%s: %w", r.brokerNamespace(broker), broker.Spec.Config.Name, getCmError)
 		}
 		return nil, cm, fmt.Errorf("unable to build topic config from configmap: %w - ConfigMap data: %v", err, cm.Data)
 	}
@@ -496,3 +532,51 @@ func isExternalTopic(broker *eventing.Broker) (string, bool) {
 	topicAnnotationValue, ok := broker.Annotations[ExternalTopicAnnotation]
 	return topicAnnotationValue, ok
 }
+
+func (r *Reconciler) addFinalizerSecret(ctx context.Context, finalizer string, secret *corev1.Secret) error {
+	if !containsFinalizerSecret(secret, finalizer) {
+		secret := secret.DeepCopy() // Do not modify informer copy.
+		secret.Finalizers = append(secret.Finalizers, finalizer)
+		_, err := r.KubeClient.CoreV1().Secrets(secret.GetNamespace()).Update(ctx, secret, metav1.UpdateOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return fmt.Errorf("failed to add finalizer to Secret %s/%s: %w", secret.GetNamespace(), secret.GetName(), err)
+		}
+	}
+	return nil
+}
+
+func (r *Reconciler) removeFinalizerSecret(ctx context.Context, finalizer string, secret *corev1.Secret) error {
+	if secret != nil {
+		newFinalizers := make([]string, 0, len(secret.Finalizers))
+		for _, f := range secret.Finalizers {
+			if f != finalizer {
+				newFinalizers = append(newFinalizers, f)
+			}
+		}
+		if len(newFinalizers) != len(secret.Finalizers) {
+			secret := secret.DeepCopy() // Do not modify informer copy.
+			secret.Finalizers = newFinalizers
+			_, err := r.KubeClient.CoreV1().Secrets(secret.GetNamespace()).Update(ctx, secret, metav1.UpdateOptions{})
+			if err != nil && !apierrors.IsNotFound(err) {
+				return fmt.Errorf("failed to remove finalizer %s from Secret %s/%s: %w", finalizer, secret.GetNamespace(), secret.GetName(), err)
+			}
+		}
+	}
+	return nil
+}
+
+func containsFinalizerSecret(secret *corev1.Secret, finalizer string) bool {
+	if secret == nil {
+		return false
+	}
+	for _, f := range secret.Finalizers {
+		if f == finalizer {
+			return true
+		}
+	}
+	return false
+}
+
+func finalizerSecret(object metav1.Object) string {
+	return fmt.Sprintf("%s/%s", "kafka.eventing", object.GetUID())
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 4b53865b4..2863757cd 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -22,6 +22,8 @@ import (
 	"net/url"
 	"testing"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/network"
 
@@ -1219,6 +1221,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				SecretFinalizerUpdate("secret-1", SecretFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1878,6 +1881,30 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 	useTable(t, table, &env)
 }
 
+func SecretFinalizerUpdate(secretName, finalizerName string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Secret",
+		},
+		ConfigMapNamespace,
+		BrokerSecretWithFinalizer(ConfigMapNamespace, secretName, finalizerName),
+	)
+}
+
+func SecretFinalizerUpdateRemove(secretName string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Secret",
+		},
+		ConfigMapNamespace,
+		NewSSLSecret(ConfigMapNamespace, secretName),
+	)
+}
+
 func TestBrokerFinalizer(t *testing.T) {
 	t.Parallel()
 
@@ -2005,6 +2032,57 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
+		{
+			Name: "Reconciled normal - with auth config",
+			Objects: []runtime.Object{
+				NewDeletedBroker(
+					WithExternalTopic(ExternalTopicName),
+					WithBrokerConfig(KReference(BrokerConfig(bootstrapServers, 20, 5,
+						BrokerAuthConfig("secret-1"),
+					))),
+					BrokerConfigMapSecretAnnotation("secret-1"),
+				),
+				BrokerSecretWithFinalizer(ConfigMapNamespace, "secret-1", SecretFinalizerName),
+				BrokerConfig(bootstrapServers, 20, 5, BrokerAuthConfig("secret-1")),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+						},
+					},
+					Generation: 1,
+				}, &env),
+				NewService(),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve": "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Resources:  []*contract.Resource{},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				SecretFinalizerUpdateRemove("secret-1"),
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+
 		{
 			Name: "Failed to delete topic",
 			Objects: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 408bf34c4..4827f1ccc 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -48,6 +48,8 @@ const (
 	BrokerName        = "test-broker"
 	ExternalTopicName = "test-topic"
 
+	SecretFinalizerName = "kafka.eventing/" + BrokerUUID
+
 	TriggerName      = "test-trigger"
 	TriggerNamespace = "test-namespace"
 )
@@ -77,6 +79,12 @@ func NewBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
 	)
 }
 
+func BrokerSecretWithFinalizer(ns, name, finalizerName string) *corev1.Secret {
+	secret := NewSSLSecret(ns, name)
+	secret.Finalizers = append(secret.Finalizers, finalizerName)
+	return secret
+}
+
 func NewDeletedBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
 	return NewBroker(
 		append(
@@ -145,6 +153,7 @@ func WithExternalTopic(topic string) func(*eventing.Broker) {
 }
 
 type CMOption func(cm *corev1.ConfigMap)
+type SecretOption func(secret *corev1.Secret)
 
 func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int, options ...CMOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
@@ -164,6 +173,12 @@ func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int,
 	return cm
 }
 
+func BrokerSecretFinalizer(finalizerName string) SecretOption {
+	return func(secret *corev1.Secret) {
+		secret.Finalizers = append(secret.Finalizers, finalizerName)
+	}
+}
+
 func BrokerAuthConfig(name string) CMOption {
 	return func(cm *corev1.ConfigMap) {
 		if cm.Data == nil {