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

Issue:
Kafka Broker finalizer stuck if "unable to build topic config from configmap: error validating topic config from configmap invalid configuration"
Creating a Kafka broker with an existing ConfigMap that does not have valid configuration, .e.g accidentally pointing to a MTBroker ConfigMap from a Kafka-classed Broker, leads to a Broker that cannot be deleted - the finalizer becomes stuck

to reproduce:

Config:
```yaml
apiVersion: v1
data:
  default-br-config: |
    clusterDefault:
      brokerClass: MTChannelBasedBroker
      apiVersion: v1
      kind: ConfigMap
      name: config-br-default-channel
      namespace: knative-eventing
      delivery:
        retry: 10
        backoffPolicy: exponential
        backoffDelay: PT0.2S
kind: ConfigMap
```

Create a broker like:

```yaml
apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: broker
  annotations:
    eventing.knative.dev/broker.class: Kafka
```

It is stuck as:

```bash
k get brokers.eventing.knative.dev -A 
NAMESPACE   NAME     URL   AGE     READY   REASON
default     broker         3m22s   False   unable to build topic config from configmap: error validating topic config from configmap invalid configuration - numPartitions: 0 - replicationFactor: 0 - bootstrapServers: [] - ConfigMap data: map[channel-template-spec:apiVersion: messaging.knative.dev/v1...
```

Deletion of that above broker is stuck at:

```
broker.eventing.knative.dev "broker" deleted
```

Pull request:
Fix finalization problem when broker has incorrect topic config
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #2815

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- If an invalid topic config is present for a broker we do not block finalization

<!--
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
If an invalid topic config is present for a broker we do not block finalization
```

**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 e55473786..03e598d81 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -360,7 +360,15 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	if !externalTopic {
 		topicConfig, err := r.topicConfig(logger, broker, brokerConfig)
 		if err != nil {
-			return fmt.Errorf("failed to resolve broker config: %w", err)
+
+			// On finalize, we fail to get a valid config, we can safely ignore and return nil
+			// no further actions are needed since we are also not putting the finalizer on given secret
+			// if we are not having a valid topic config
+			if strings.Contains(err.Error(), "validating topic config") {
+				return nil
+			} else {
+				return fmt.Errorf("failed to resolve broker config: %w", err)
+			}
 		}
 
 		// get security option for Sarama with secret info in it


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 553d1102d..37a504140 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -2201,7 +2201,50 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
-
+		{
+			Name: "Reconciled normal - with broken config",
+			Objects: []runtime.Object{
+				NewDeletedBroker(
+					WithBrokerConfig(KReference(BogusBrokerConfig())),
+				),
+				BogusBrokerConfig(),
+				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),
+				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.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &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",
+				}),
+			},
+			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 e5d38029d..188b27a23 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -243,6 +243,19 @@ func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int,
 	return cm
 }
 
+func BogusBrokerConfig() *corev1.ConfigMap {
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ConfigMapNamespace,
+			Name:      ConfigMapName,
+		},
+		Data: map[string]string{
+			"foo": "bar",
+		},
+	}
+	return cm
+}
+
 func BrokerAuthConfig(name string) CMOption {
 	return func(cm *corev1.ConfigMap) {
 		if cm.Data == nil {
diff --git a/test/e2e_new/bogus_config/bogus_config.yaml b/test/e2e_new/bogus_config/bogus_config.yaml
new file mode 100644
index 000000000..174c28e38
--- /dev/null
+++ b/test/e2e_new/bogus_config/bogus_config.yaml
@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ .name }}
+  namespace: {{ .namespace }}
+data:
+  channel-template-spec: |
+    apiVersion: messaging.knative.dev/v1
+    kind: InMemoryChannel
+  auth.secret.ref.name: strimzi-sasl-secret
diff --git a/test/e2e_new/bogus_config/install.go b/test/e2e_new/bogus_config/install.go
new file mode 100644
index 000000000..2239775cd
--- /dev/null
+++ b/test/e2e_new/bogus_config/install.go
@@ -0,0 +1,38 @@
+/*
+ * 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 bogus_config
+
+import (
+	"context"
+	"embed"
+
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
+)
+
+const ConfigMapName = "broken-kafka-broker-config"
+
+//go:embed *.yaml
+var yamls embed.FS
+
+func Install(ctx context.Context, t feature.T) {
+	if _, err := manifest.InstallYamlFS(ctx, yamls, map[string]interface{}{
+		"name": ConfigMapName,
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/test/e2e_new/broker_test.go b/test/e2e_new/broker_test.go
index 029d2b84a..f72e30b52 100644
--- a/test/e2e_new/broker_test.go
+++ b/test/e2e_new/broker_test.go
@@ -165,6 +165,22 @@ func TestTriggerLatestOffset(t *testing.T) {
 	env.Test(ctx, t, features.TriggerLatestOffset())
 }
 
+func TestBrokerWithBogusConfig(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.BrokerWithBogusConfig())
+}
+
 func TestBrokerCannotReachKafkaCluster(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 f89afc18c..0386a82fb 100644
--- a/test/rekt/features/broker_deleted_recreated.go
+++ b/test/rekt/features/broker_deleted_recreated.go
@@ -19,6 +19,7 @@ package features
 import (
 	"k8s.io/apimachinery/pkg/types"
 	kafkabroker "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	"knative.dev/eventing-kafka-broker/test/e2e_new/bogus_config"
 	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
 
 	"knative.dev/pkg/system"
@@ -214,6 +215,25 @@ func BrokerAuthSecretForInternalTopicDoesNotExist() *feature.Feature {
 	return f
 }
 
+// BrokerWithBogusConfig tests that a broker can be deleted even when it has a bogus config attached.
+func BrokerWithBogusConfig() *feature.Feature {
+	f := feature.NewFeatureNamed("delete broker with bogus config")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	f.Setup("install bogus configuration", bogus_config.Install)
+
+	f.Requirement("Create SASL secret", featuressteps.CopySecretInTestNamespace(system.Namespace(), "strimzi-sasl-secret"))
+
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		broker.WithConfig(bogus_config.ConfigMapName),
+	))
+	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")