topic: add annotation to allow deletion with missing credentials

When a namespace is deleted, Kubernetes may delete the credentials
Secret before the Topic CR. The Topic controller then fails to create
a Kafka client, returns an error, and never removes its finalizer.
The namespace gets stuck in Terminating state indefinitely.

Add opt-in annotation to handle this:

  operator.redpanda.com/allow-deletion-with-missing-credentials: "true"

When set, the controller allows finalizer removal during deletion if
it cannot connect due to missing credentials. The underlying Kafka
topic may be orphaned, but that beats a stuck namespace.

The annotation is off by default to preserve existing behavior. Error
detection covers both Kafka protocol errors (via ignoreAllConnectionErrors)
and K8s NotFound errors (via new isK8sNotFoundInChain helper).

Integration tests verify both behaviors:
- Without annotation: deletion fails when credentials missing
- With annotation: finalizer removed, deletion proceeds

Author: birdayz

.changes/unreleased/operator-Fixed-20260122-100940.yaml

@@ -0,0 +1,4 @@
+project: operator
+kind: Fixed
+body: Add annotation to allow Topic finalizer removal when credentials Secret is missing during deletion, preventing stuck namespaces.
+time: 2026-01-22T10:00:00.000000+01:00

operator/internal/controller/redpanda/topic_controller.go

@@ -44,10 +44,16 @@ import (
 
 const (
 	NoneConstantString = "none"
+
+	// AllowDeletionWithMissingCredentialsAnnotation when set to "true" allows
+	// Topic finalizer removal even if the credentials Secret is missing during
+	// deletion. This prevents namespaces from getting stuck in Terminating state
+	// when the Secret is deleted before the Topic.
+	AllowDeletionWithMissingCredentialsAnnotation = "operator.redpanda.com/allow-deletion-with-missing-credentials"
 )
 
 var (
-	ErrScaleDownPartitionCount     = errors.New("unable to scale down number of partition in topic")
+	ErrScaleDownPartitionCount = errors.New("unable to scale down number of partition in topic")
 	ErrEmptyTopicConfigDescription = errors.New("topic config description response is empty")
 	ErrEmptyMetadataTopic          = errors.New("metadata topic response is empty")
 	ErrWrongCreateTopicResponse    = errors.New("requested topic was not part of create topic response")
@@ -185,6 +191,20 @@ func (r *TopicReconciler) reconcile(ctx context.Context, recorder record.EventRe
 
 	kafkaClient, err := r.createKafkaClient(ctx, topic, l)
 	if err != nil {
+		// If topic is being deleted and the AllowDeletionWithMissingCredentials
+		// annotation is set, allow finalizer removal when credentials are missing.
+		// This prevents namespaces from getting stuck in Terminating state.
+		//
+		// We check both ignoreAllConnectionErrors (Kafka protocol/config errors)
+		// and K8s NotFound (missing secrets/configmaps) since the latter isn't
+		// covered by the general helper.
+		if !topic.ObjectMeta.DeletionTimestamp.IsZero() {
+			allowMissing := topic.GetAnnotations()[AllowDeletionWithMissingCredentialsAnnotation] == "true"
+			if allowMissing && (ignoreAllConnectionErrors(l, err) == nil || isK8sNotFoundInChain(err)) {
+				l.V(2).Info("Ignoring client error during deletion (annotation enabled)", "error", err)
+				return redpandav1alpha2.TopicReady(topic), ctrl.Result{}, nil
+			}
+		}
 		return redpandav1alpha2.TopicFailed(topic), ctrl.Result{}, err
 	}
 	defer kafkaClient.Close()
@@ -602,3 +622,16 @@ func (r *TopicReconciler) recordErrorEvent(err error, recorder record.EventRecor
 	args = append(args, err)
 	return fmt.Errorf(message+": %w", args...) // nolint:goerr113 // That is not dynamic error
 }
+
+// isK8sNotFoundInChain walks the error chain to check if a K8s NotFound error
+// is wrapped anywhere. This is needed because secret loading wraps the underlying
+// K8s error, and apierrors.IsNotFound() doesn't traverse wrapped errors.
+func isK8sNotFoundInChain(err error) bool {
+	for err != nil {
+		if apierrors.IsNotFound(err) {
+			return true
+		}
+		err = errors.Unwrap(err)
+	}
+	return false
+}

operator/internal/controller/redpanda/topic_controller_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/twmb/franz-go/pkg/kadm"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/kmsg"
+	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"
@@ -823,4 +824,109 @@ func TestReconcile(t *testing.T) { // nolint:funlen // These tests have clear su
 
 		assert.Equal(t, time.Duration(0), result.RequeueAfter)
 	})
+	t.Run("delete_topic_with_missing_credentials_errors_without_annotation", func(t *testing.T) {
+		// Without the AllowDeletionWithMissingCredentials annotation, deletion
+		// should fail when the credentials Secret is missing. This is the default
+		// behavior to prevent accidental data loss.
+		topicName := "delete-topic-missing-secret-no-annotation"
+
+		topic := redpandav1alpha2.Topic{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:       topicName,
+				Namespace:  testNamespace,
+				Finalizers: []string{FinalizerKey},
+			},
+			Spec: redpandav1alpha2.TopicSpec{
+				Partitions:        ptr.To(1),
+				ReplicationFactor: ptr.To(1),
+				KafkaAPISpec: &redpandav1alpha2.KafkaAPISpec{
+					Brokers: []string{seedBroker},
+					SASL: &redpandav1alpha2.KafkaSASL{
+						Username: "testuser",
+						Password: &redpandav1alpha2.ValueSource{
+							SecretKeyRef: &corev1.SecretKeySelector{
+								LocalObjectReference: corev1.LocalObjectReference{
+									Name: "non-existent-secret",
+								},
+								Key: "password",
+							},
+						},
+						Mechanism: redpandav1alpha2.SASLMechanismScramSHA256,
+					},
+				},
+			},
+		}
+
+		err := c.Create(ctx, &topic)
+		require.NoError(t, err)
+
+		err = c.Delete(ctx, &topic)
+		require.NoError(t, err)
+
+		key := types.NamespacedName{Name: topicName, Namespace: testNamespace}
+		req := mcreconcile.Request{Request: ctrl.Request{NamespacedName: key}, ClusterName: mcmanager.LocalCluster}
+
+		// Without annotation, reconcile should return an error
+		_, err = tr.Reconcile(ctx, req)
+		assert.Error(t, err, "reconciler should error when credentials secret is missing without annotation")
+		assert.Contains(t, err.Error(), "non-existent-secret")
+
+		// Topic should still exist (finalizer not removed)
+		err = c.Get(ctx, key, &topic)
+		assert.NoError(t, err, "topic should still exist when deletion fails")
+	})
+	t.Run("delete_topic_with_missing_credentials_succeeds_with_annotation", func(t *testing.T) {
+		// With the AllowDeletionWithMissingCredentials annotation set to "true",
+		// deletion should succeed even when the credentials Secret is missing.
+		// This prevents namespaces from getting stuck in Terminating state.
+		topicName := "delete-topic-missing-secret-with-annotation"
+
+		topic := redpandav1alpha2.Topic{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:       topicName,
+				Namespace:  testNamespace,
+				Finalizers: []string{FinalizerKey},
+				Annotations: map[string]string{
+					AllowDeletionWithMissingCredentialsAnnotation: "true",
+				},
+			},
+			Spec: redpandav1alpha2.TopicSpec{
+				Partitions:        ptr.To(1),
+				ReplicationFactor: ptr.To(1),
+				KafkaAPISpec: &redpandav1alpha2.KafkaAPISpec{
+					Brokers: []string{seedBroker},
+					SASL: &redpandav1alpha2.KafkaSASL{
+						Username: "testuser",
+						Password: &redpandav1alpha2.ValueSource{
+							SecretKeyRef: &corev1.SecretKeySelector{
+								LocalObjectReference: corev1.LocalObjectReference{
+									Name: "non-existent-secret",
+								},
+								Key: "password",
+							},
+						},
+						Mechanism: redpandav1alpha2.SASLMechanismScramSHA256,
+					},
+				},
+			},
+		}
+
+		err := c.Create(ctx, &topic)
+		require.NoError(t, err)
+
+		err = c.Delete(ctx, &topic)
+		require.NoError(t, err)
+
+		key := types.NamespacedName{Name: topicName, Namespace: testNamespace}
+		req := mcreconcile.Request{Request: ctrl.Request{NamespacedName: key}, ClusterName: mcmanager.LocalCluster}
+
+		// With annotation, reconcile should succeed
+		result, err := tr.Reconcile(ctx, req)
+		assert.NoError(t, err, "reconciler should not error when annotation is set")
+		assert.Equal(t, time.Duration(0), result.RequeueAfter)
+
+		// Topic should be deleted (finalizer removed)
+		err = c.Get(ctx, key, &topic)
+		assert.True(t, apierrors.IsNotFound(err), "topic should be deleted after finalizer removal")
+	})
 }