Repository: knative-sandbox/eventing-kafka-broker
Pull request #1599 (2021-12-14 11:05:27)

[Automated] Update eventing-kafka-broker-eventing-kafka nightly
/assign @knative/delivery-wg-leads
/cc @knative/delivery-wg-leads

Produced via:
```shell
for x in channel-crds.yaml source-crd.yaml; do
  curl https://storage.googleapis.com/knative-nightly/eventing-kafka/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-kafka-latest/$x
done
```

Details:
```

```

knative-prow-robot: @knative-automation: GitHub didn't allow me to request PR reviews from the following users: knative/delivery-wg-leads.

Note that only [knative-sandbox members](https://github.com/orgs/knative-sandbox/people) and repo collaborators can review this PR, and authors cannot review their own PRs.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1599):

>/assign @knative/delivery-wg-leads
>/cc @knative/delivery-wg-leads
>
>Produced via:
>```shell
>for x in channel-crds.yaml source-crd.yaml; do
>  curl https://storage.googleapis.com/knative-nightly/eventing-kafka/latest/$x > ${GITHUB_WORKSPACE}/./third_party/eventing-kafka-latest/$x
>done
>```
>
>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.
</details>
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1599#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **markusthoemmes** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @markusthoemmes` in a comment when ready.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/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":["markusthoemmes"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1599](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5cb44cd) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/941cc4ad5ee0d58277297b7be90db97123638055?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (941cc4a) will **decrease** coverage by `15.57%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff              @@
##               main    #1599       +/-   ##
=============================================
- Coverage     81.80%   66.22%   -15.58%     
=============================================
  Files            66       52       -14     
  Lines          1929     2419      +490     
  Branches        171        0      -171     
=============================================
+ Hits           1578     1602       +24     
- Misses          261      648      +387     
- Partials         90      169       +79     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...nciler/impl/ResourcesReconcilerMessageHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJNZXNzYWdlSGFuZGxlci5qYXZh) | | |
| [...ka/broker/core/security/KubernetesCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS3ViZXJuZXRlc0NyZWRlbnRpYWxzLmphdmE=) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| [.../dispatcher/impl/RecordDispatcherMutatorChain.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVyTXV0YXRvckNoYWluLmphdmE=) | | |
| [...kafka/broker/dispatcher/impl/filter/AllFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQWxsRmlsdGVyLmphdmE=) | | |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | | |
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| [...roker/dispatcher/impl/filter/AttributesFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvQXR0cmlidXRlc0ZpbHRlci5qYXZh) | | |
| [...tive/eventing/kafka/broker/core/utils/BaseEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQmFzZUVudi5qYXZh) | | |
| [...nting/kafka/broker/core/security/AuthProvider.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQXV0aFByb3ZpZGVyLmphdmE=) | | |
| ... and [108 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599?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/1599?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [941cc4a...5cb44cd](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1599?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).


Patch:
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 84b3458bb..d5596fc8a 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -23,6 +23,8 @@ import (
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
@@ -87,7 +89,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	}
 	statusConditionManager.DataPlaneAvailable()
 
-	topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
+	topicConfig, brokerConfig, err := r.topicConfig(ctx, logger, broker)
 	if err != nil {
 		return statusConditionManager.FailedToResolveConfig(err)
 	}
@@ -288,7 +290,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
 	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
 
-	topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
+	topicConfig, brokerConfig, err := r.topicConfig(ctx, logger, broker)
 	if err != nil {
 		return fmt.Errorf("failed to resolve broker config: %w", err)
 	}
@@ -331,10 +333,14 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 
 	logger.Debug("Topic deleted", zap.String("topic", topic))
 
+	if err := r.removeFinalizerCM(ctx, finalizerCM(broker), brokerConfig); err != nil {
+		return err
+	}
+
 	return nil
 }
 
-func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*kafka.TopicConfig, *corev1.ConfigMap, error) {
+func (r *Reconciler) topicConfig(ctx context.Context, logger *zap.Logger, broker *eventing.Broker) (*kafka.TopicConfig, *corev1.ConfigMap, error) {
 
 	logger.Debug("broker config", zap.Any("broker.spec.config", broker.Spec.Config))
 
@@ -352,6 +358,10 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*
 		return nil, nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, broker.Spec.Config.Name, err)
 	}
 
+	if err := r.addFinalizerCM(ctx, finalizerCM(broker), cm); err != nil {
+		return nil, nil, err
+	}
+
 	topicConfig, err := kafka.TopicConfigFromConfigMap(logger, cm)
 	if err != nil {
 		return nil, cm, fmt.Errorf("unable to build topic config from configmap: %w - ConfigMap data: %v", err, cm.Data)
@@ -391,3 +401,49 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 
 	return resource, nil
 }
+
+func (r *Reconciler) addFinalizerCM(ctx context.Context, finalizer string, cm *corev1.ConfigMap) error {
+	if !containsFinalizerCM(cm, finalizer) {
+		cm := cm.DeepCopy() // Do not modify informer copy.
+		cm.Finalizers = append(cm.Finalizers, finalizer)
+		_, err := r.KubeClient.CoreV1().ConfigMaps(cm.GetNamespace()).Update(ctx, cm, metav1.UpdateOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return fmt.Errorf("failed to add finalizer to ConfigMap %s/%s: %w", cm.GetNamespace(), cm.GetName(), err)
+		}
+	}
+	return nil
+}
+
+func containsFinalizerCM(cm *corev1.ConfigMap, finalizer string) bool {
+	if cm == nil {
+		return false
+	}
+	for _, f := range cm.Finalizers {
+		if f == finalizer {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *Reconciler) removeFinalizerCM(ctx context.Context, finalizer string, cm *corev1.ConfigMap) error {
+	newFinalizers := make([]string, 0, len(cm.Finalizers))
+	for _, f := range cm.Finalizers {
+		if f != finalizer {
+			newFinalizers = append(newFinalizers, f)
+		}
+	}
+	if len(newFinalizers) != len(cm.Finalizers) {
+		cm := cm.DeepCopy() // Do not modify informer copy.
+		cm.Finalizers = newFinalizers
+		_, err := r.KubeClient.CoreV1().ConfigMaps(cm.GetNamespace()).Update(ctx, cm, metav1.UpdateOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return fmt.Errorf("failed to remove finalizer %s to ConfigMap %s/%s: %w", finalizer, cm.GetNamespace(), cm.GetName(), err)
+		}
+	}
+	return nil
+}
+
+func finalizerCM(object metav1.Object) string {
+	return fmt.Sprintf("%s/%s-%s", "kafka.brokers.eventing.knative.dev", object.GetNamespace(), object.GetName())
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index a655eb2eb..7f4f3f2c3 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -22,6 +22,7 @@ import (
 	"net/url"
 	"testing"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/network"
 
@@ -135,6 +136,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -193,6 +195,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -253,6 +256,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -311,6 +315,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -368,6 +373,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -425,6 +431,9 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					BrokerTopic(), createTopicError,
 				),
 			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
@@ -468,6 +477,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -523,6 +533,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -591,6 +602,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -681,6 +693,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -771,6 +784,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -849,6 +863,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -927,6 +942,9 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			Key:     testKey,
 			WantErr: true,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
+			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 				Eventf(
@@ -969,6 +987,11 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			Key:     testKey,
 			WantErr: true,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName, func(cm *corev1.ConfigMap) {
+					cm.Data["bootstrap.servers"] = ""
+				}),
+			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 				Eventf(
@@ -1015,6 +1038,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1085,6 +1109,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName, BrokerAuthConfig("secret-1")),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1279,6 +1304,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1388,6 +1414,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1452,6 +1479,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1516,6 +1544,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1577,6 +1606,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1648,7 +1678,9 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
-			WantUpdates: []clientgotesting.UpdateActionImpl{},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
@@ -1696,6 +1728,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdate(ConfigMapFinalizerName),
 				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
@@ -1727,6 +1760,40 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 	useTable(t, table, &env)
 }
 
+func ConfigMapFinalizerUpdate(name string, opts ...CMOption) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "ConfigMap",
+		},
+		ConfigMapNamespace,
+		BrokerConfig(bootstrapServers, 20, 5,
+			append(
+				[]CMOption{BrokerConfigFinalizer(name)},
+				opts...,
+			)...,
+		),
+	)
+}
+
+func ConfigMapFinalizerUpdateRemove(name string, opts ...CMOption) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "ConfigMap",
+		},
+		ConfigMapNamespace,
+		BrokerConfig(bootstrapServers, 20, 5,
+			append(
+				[]CMOption{BrokerConfigFinalizerRemove(name)},
+				opts...,
+			)...,
+		),
+	)
+}
+
 func TestBrokerFinalizer(t *testing.T) {
 	t.Parallel()
 
@@ -1746,7 +1813,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1764,6 +1833,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
+				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 		},
 		{
@@ -1772,7 +1842,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(
 					WithDelivery(),
 				),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1790,13 +1862,16 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 2,
 				}),
+				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 		},
 		{
 			Name: "Failed to delete topic",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1833,10 +1908,15 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(
 					WithDelivery(),
 				),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewService(),
 			},
 			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
+			},
 			WantCreates: []runtime.Object{
 				NewConfigMapWithBinaryData(&env, nil),
 			},
@@ -1846,7 +1926,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - preserve config map previous state",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1875,13 +1957,16 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					},
 					Generation: 6,
 				}),
+				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 		},
 		{
 			Name: "Reconciled normal - topic doesn't exist",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1910,6 +1995,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					},
 					Generation: 6,
 				}),
+				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
 			},
 			OtherTestData: map[string]interface{}{
 				wantErrorOnDeleteTopic: sarama.ErrUnknownTopicOrPartition,
@@ -1919,7 +2005,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no broker found in config map",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
-				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerConfig(bootstrapServers, 20, 5,
+					BrokerConfigFinalizer(ConfigMapFinalizerName),
+				),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1932,6 +2020,9 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}, &env),
 			},
 			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapFinalizerUpdateRemove(ConfigMapFinalizerName),
+			},
 		},
 	}
 
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index df1c10289..4af25d83f 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -23,6 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/sets"
 	clientgotesting "k8s.io/client-go/testing"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -44,6 +45,8 @@ const (
 	BrokerUUID      = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
 	BrokerNamespace = "test-namespace"
 	BrokerName      = "test-broker"
+
+	ConfigMapFinalizerName = "kafka.brokers.eventing.knative.dev/" + BrokerNamespace + "-" + BrokerName
 )
 
 func BrokerTopic() string {
@@ -131,8 +134,9 @@ type CMOption func(cm *corev1.ConfigMap)
 func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int, options ...CMOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Namespace: ConfigMapNamespace,
-			Name:      ConfigMapName,
+			Namespace:  ConfigMapNamespace,
+			Name:       ConfigMapName,
+			Finalizers: []string{"finalizer-that-we-should-not-drop"},
 		},
 		Data: map[string]string{
 			kafka.BootstrapServersConfigMapKey:              bootstrapServers,
@@ -146,6 +150,18 @@ func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int,
 	return cm
 }
 
+func BrokerConfigFinalizer(finalizerName string) CMOption {
+	return func(cm *corev1.ConfigMap) {
+		cm.Finalizers = append(cm.Finalizers, finalizerName)
+	}
+}
+
+func BrokerConfigFinalizerRemove(finalizerName string) CMOption {
+	return func(cm *corev1.ConfigMap) {
+		cm.Finalizers = sets.NewString(cm.Finalizers...).Delete(finalizerName).List()
+	}
+}
+
 func BrokerAuthConfig(name string) CMOption {
 	return func(cm *corev1.ConfigMap) {
 		if cm.Data == nil {
diff --git a/test/e2e_new/broker_test.go b/test/e2e_new/broker_test.go
index 6b4d59042..70ae883dd 100644
--- a/test/e2e_new/broker_test.go
+++ b/test/e2e_new/broker_test.go
@@ -43,3 +43,17 @@ func TestBrokerDeletedRecreated(t *testing.T) {
 
 	env.Test(ctx, t, features.BrokerDeletedRecreated())
 }
+
+func TestBrokerConfigMapDeletedFirst(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.BrokerConfigMapDeletedFirst())
+}
diff --git a/test/e2e_new/features/broker_deleted_recreated.go b/test/e2e_new/features/broker_deleted_recreated.go
index 95aa5656b..e54458c15 100644
--- a/test/e2e_new/features/broker_deleted_recreated.go
+++ b/test/e2e_new/features/broker_deleted_recreated.go
@@ -17,9 +17,16 @@
 package features
 
 import (
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/resources/svc"
 
 	"knative.dev/eventing-kafka-broker/test/e2e_new/features/featuressteps"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/resources/configmap"
 )
 
 func BrokerDeletedRecreated() *feature.Feature {
@@ -34,3 +41,28 @@ func BrokerDeletedRecreated() *feature.Feature {
 
 	return f
 }
+
+func BrokerConfigMapDeletedFirst() *feature.Feature {
+	f := feature.NewFeatureNamed("delete broker ConfigMap first")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	cmName := feature.MakeRandomK8sName("cm-deleted-first")
+	sink := feature.MakeRandomK8sName("sink")
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+	f.Setup("install config", configmap.Copy(
+		types.NamespacedName{Namespace: system.Namespace(), Name: "kafka-broker-config"},
+		cmName,
+	))
+	f.Setup("install broker", broker.Install(brokerName, append(broker.WithEnvConfig(), broker.WithConfig(cmName))...))
+	f.Setup("install trigger", trigger.Install(triggerName, brokerName,
+		trigger.WithSubscriber(svc.AsKReference(sink), "")),
+	)
+
+	f.Requirement("delete Broker ConfigMap", featuressteps.DeleteConfigMap(cmName))
+
+	f.Assert("delete broker", featuressteps.DeleteBroker(brokerName))
+
+	return f
+}
diff --git a/test/e2e_new/features/featuressteps/broker.go b/test/e2e_new/features/featuressteps/broker.go
index c156fd93a..6be502fc8 100644
--- a/test/e2e_new/features/featuressteps/broker.go
+++ b/test/e2e_new/features/featuressteps/broker.go
@@ -21,10 +21,14 @@ import (
 
 	cetest "github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
@@ -80,3 +84,29 @@ func BrokerSmokeTest(brokerName, triggerName string) feature.StepFn {
 func DeleteResources(f *feature.Feature) feature.StepFn {
 	return compose(f.DeleteResources)
 }
+
+func DeleteConfigMap(name string) feature.StepFn {
+	foreground := metav1.DeletePropagationForeground
+	return func(ctx context.Context, t feature.T) {
+		ns := environment.FromContext(ctx).Namespace()
+		err := kubeclient.Get(ctx).CoreV1().
+			ConfigMaps(ns).
+			Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &foreground})
+		if err != nil {
+			t.Fatalf("Failed to delete ConfigMap %s/%s: %w\n", ns, name, err)
+		}
+	}
+}
+
+func DeleteBroker(name string) feature.StepFn {
+	foreground := metav1.DeletePropagationForeground
+	return func(ctx context.Context, t feature.T) {
+		ns := environment.FromContext(ctx).Namespace()
+		err := eventingclient.Get(ctx).EventingV1().
+			Brokers(ns).
+			Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &foreground})
+		if err != nil {
+			t.Fatalf("Failed to delete Broker %s/%s: %w\n", ns, name, err)
+		}
+	}
+}
diff --git a/test/e2e_new/resources/configmap/copy.go b/test/e2e_new/resources/configmap/copy.go
new file mode 100644
index 000000000..f8787bb16
--- /dev/null
+++ b/test/e2e_new/resources/configmap/copy.go
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2021 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 configmap
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+func Copy(from types.NamespacedName, toName string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		cm, err := kubeclient.Get(ctx).CoreV1().
+			ConfigMaps(from.Namespace).
+			Get(ctx, from.Name, metav1.GetOptions{})
+		require.Nil(t, err)
+
+		ns := environment.FromContext(ctx).Namespace()
+		toCm := &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: ns,
+				Name:      toName,
+			},
+			Data:       cm.Data,
+			BinaryData: cm.BinaryData,
+		}
+
+		_, err = kubeclient.Get(ctx).CoreV1().
+			ConfigMaps(toCm.GetNamespace()).
+			Create(ctx, toCm, metav1.CreateOptions{})
+		if apierrors.IsAlreadyExists(err) {
+			return
+		}
+		require.Nil(t, err)
+	}
+}
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index b66adc2f8..8718f4d55 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20211212-58afd241e"
+    kafka.eventing.knative.dev/release: "v20211214-f1bb19a50"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crd.yaml b/third_party/eventing-kafka-latest/source-crd.yaml
index 9d9e3a338..9dcfc66f3 100644
--- a/third_party/eventing-kafka-latest/source-crd.yaml
+++ b/third_party/eventing-kafka-latest/source-crd.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20211212-58afd241e"
+    kafka.eventing.knative.dev/release: "v20211214-f1bb19a50"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"