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

Issue:
Broker referencing a Secret finalizer not deleted when "the object has been modified" occurs
**Describe the bug**
Deleting a Broker referencing a Secret sometimes leads to a Broker stuck with the "brokers.eventing.knative.dev" finalizer.

Aparently this occurs when both the Broker and the Secret are being deleted at around the same time, and an error occurs which prevents the deletion of the Broker's finalizer:

```
2023-01-12T13:12:00.355354032Z {"level":"error","ts":"2023-01-12T13:12:00.354Z","logger":"kafka-broker-controller","caller":"controller/controller.go:566","msg":"Reconcile error","commit":"24e536e-dirty","knative.dev/pod":"kafka-controller-9b87fdb6b-plh4j","knative.dev/controller":"knative.dev.eventing-kafka-broker.control-plane.pkg.reconciler.broker.Reconciler","knative.dev/kind":"eventing.knative.dev.Broker","knative.dev/traceid":"09266b22-4b65-4b12-b799-f9ace7b2d5a3","knative.dev/key":"kb-dn-saslplain-4/sender3","duration":0.26126783,"error":"failed to clear finalizers: Operation cannot be fulfilled on brokers.eventing.knative.dev \"sender3\": the object has been modified; please apply your changes to the latest version and try again","stacktrace":"knative.dev/pkg/controller.(*Impl).handleErr\n\t/go/src/github.com/openshift/origin/vendor/knative.dev/pkg/controller/controller.go:566\nknative.dev/pkg/controller.(*Impl).processNextWorkItem\n\t/go/src/github.com/openshift/origin/vendor/knative.dev/pkg/controller/controller.go:543\nknative.dev/pkg/controller.(*Impl).RunContext.func3\n\t/go/src/github.com/openshift/origin/vendor/knative.dev/pkg/controller/controller.go:491"}
```

 A subsequent reconciliation of the Broker fails due to the Secret no longer existing:

```
2023-01-12T13:12:00.414276190Z {"level":"error","ts":"2023-01-12T13:12:00.414Z","logger":"kafka-broker-controller","caller":"broker/reconciler.go:313","msg":"Returned an error","commit":"24e536e-dirty","knative.dev/pod":"kafka-controller-9b87fdb6b-plh4j","knative.dev/controller":"knative.dev.eventing-kafka-broker.control-plane.pkg.reconciler.broker.Reconciler","knative.dev/kind":"eventing.knative.dev.Broker","knative.dev/traceid":"8afb6abc-6f3f-4a18-8918-046ab9e69c92","knative.dev/key":"kb-dn-saslplain-4/sender3","targetMethod":"FinalizeKind","error":"failed to get secret: failed to get secret kb-dn-saslplain-4/sender3: secrets \"sender3\" not found","stacktrace":"knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker.(*reconcilerImpl).Reconcile\n\t/go/src/github.com/openshift/origin/vendor/knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker/reconciler.go:313\nknative.dev/pkg/controller.(*Impl).processNextWorkItem\n\t/go/src/github.com/openshift/origin/vendor/knative.dev/pkg/controller/controller.go:542\nknative.dev/pkg/controller.(*Impl).RunContext.func3\n\t/go/src/github.com/openshift/origin/vendor/knative.dev/pkg/controller/controller.go:491"}
```

 **Expected behavior**
It should be possible to delete the Broker when the referenced secret's finalizer was already successfully removed by the kafka broker controller.

 **To Reproduce**
no specific reproducer, noticed on a long running test cluster

 **Knative release version**
1.6

 **Additional context**
running on OpenShift, deployed via a CI build of Serverless operator 1.27 https://github.com/openshift-knative/serverless-operator/tree/release-1.27 

Pull request description:
Improve finalization on missing secret ref
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #2885

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- No different treatment on internal or external topic for finalization
- If secret is missing, we try a number of times, otherwise we let go and do not block finalization
- added `rekt` test for "non existing secret" for "internal topic

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
No different treatment on internal or external topic for finalization, and we no longer block if we can not resolve the secret reference for a number of times.
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 05ea829c9..e55473786 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -335,15 +335,13 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return nil
 	}
 
-	_, externalTopic := isExternalTopic(broker)
 	secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: brokerConfig, UseNamespaceInConfigmap: false}, r.SecretProviderFunc())
 	if err != nil {
-
-		if externalTopic {
-			// we don't care
-			return nil
-		} else {
-			return fmt.Errorf("failed to get secret: %w", err)
+		// If we can not get the referenced secret,
+		// let us try for a bit before we give up.
+		brokerUUID := string(broker.GetUID())
+		if r.Counter.Inc(brokerUUID) <= 5 {
+			return controller.NewRequeueAfter(5 * time.Second)
 		}
 	}
 
@@ -358,6 +356,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 
 	// External topics are not managed by the broker,
 	// therefore we do not delete them
+	_, externalTopic := isExternalTopic(broker)
 	if !externalTopic {
 		topicConfig, err := r.topicConfig(logger, broker, brokerConfig)
 		if err != nil {


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index fae9a678e..553d1102d 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"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/counter"
+
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"k8s.io/utils/pointer"
@@ -2179,7 +2181,8 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
-			Key: testKey,
+			Key:     testKey,
+			WantErr: true,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
 					Resources:  []*contract.Resource{},
@@ -2427,8 +2430,9 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 					T:                                      t,
 				}, nil
 			},
-			Env:    env,
-			Prober: proberMock,
+			Env:     env,
+			Prober:  proberMock,
+			Counter: counter.NewExpiringCounter(ctx),
 		}
 
 		reconciler.Tracker = &FakeTracker{}
diff --git a/test/e2e_new/broker_test.go b/test/e2e_new/broker_test.go
index 6e24ffe38..029d2b84a 100644
--- a/test/e2e_new/broker_test.go
+++ b/test/e2e_new/broker_test.go
@@ -133,6 +133,22 @@ func TestBrokerExternalTopicAuthSecretDoesNotExist(t *testing.T) {
 	env.Test(ctx, t, features.BrokerExternalTopicAuthSecretDoesNotExist())
 }
 
+func TestBrokerAuthSecretForInternalTopicDoesNotExist(t *testing.T) {
+	// this test is observed to flake more when it is parallel
+	// t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.WithPollTimings(PollInterval, PollTimeout),
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.BrokerAuthSecretForInternalTopicDoesNotExist())
+}
+
 func TestTriggerLatestOffset(t *testing.T) {
 	// this test is observed to flake more when it is parallel
 	// t.Parallel()
diff --git a/test/rekt/features/broker_deleted_recreated.go b/test/rekt/features/broker_deleted_recreated.go
index 01e1ab486..f89afc18c 100644
--- a/test/rekt/features/broker_deleted_recreated.go
+++ b/test/rekt/features/broker_deleted_recreated.go
@@ -188,6 +188,32 @@ func BrokerExternalTopicAuthSecretDoesNotExist() *feature.Feature {
 	return f
 }
 
+// BrokerAuthSecretForInternalTopicDoesNotExist tests that a broker can be deleted without the Secret.
+func BrokerAuthSecretForInternalTopicDoesNotExist() *feature.Feature {
+	f := feature.NewFeatureNamed("delete broker with non existing Secret or Topic")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	configName := feature.MakeRandomK8sName("config")
+
+	f.Setup("create broker config", brokerconfigmap.Install(
+		configName,
+		brokerconfigmap.WithBootstrapServer(testpkg.BootstrapServersSsl),
+		brokerconfigmap.WithAuthSecret("does-not-exist"),
+		brokerconfigmap.WithNumPartitions(1),
+		brokerconfigmap.WithReplicationFactor(1),
+	))
+
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		append(
+			broker.WithEnvConfig(),
+			broker.WithConfig(configName))...))
+
+	f.Assert("delete broker", featuressteps.DeleteBroker(brokerName))
+
+	return f
+}
+
 // BrokerCannotReachKafkaCluster tests that a broker can be deleted even when KafkaCluster is unreachable.
 func BrokerCannotReachKafkaCluster() *feature.Feature {
 	f := feature.NewFeatureNamed("delete broker with unreachable Kafka cluster")