fix: pre-create topic in produce acceptance test and harden controller

The Pipeline_produces_to_Redpanda_via_clusterRef acceptance test failed
consistently because Redpanda defaults auto_create_topics_enabled to
false. The producer pipeline could not auto-create the target topic, so
no messages were ever delivered.

- Pre-create pipeline-produce-test topic before running the producer
  pipeline, matching the pattern used by the consumer scenario
- Remove misleading "Found topic" logs from ExpectTopic/ExpectNoTopic
  that printed unconditionally even when the topic was not found
- Increase checkTopic timeout from 10s to 30s for CI stability
- Handle NotFound/Conflict errors during finalizer removal to avoid
  noisy UID precondition errors when pipelines are deleted concurrently

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: david-yu

acceptance/features/pipeline-crds.feature

@@ -200,6 +200,7 @@ Feature: Pipeline CRDs
     Then pipeline "invalid-pipeline" has invalid config
 
   Scenario: Pipeline produces to Redpanda via clusterRef
+    Given I create topic "pipeline-produce-test" in cluster "basic"
     When I apply Kubernetes manifest:
     """
     ---

acceptance/steps/helpers.go

@@ -252,15 +252,13 @@ func (c *clusterClients) ExpectTopic(ctx context.Context, topic string) {
 
 	t.Logf("Checking that topic %q exists in cluster %q", topic, c.cluster)
 	c.checkTopic(ctx, topic, true, fmt.Sprintf("Topic %q does not exist in cluster %q", topic, c.cluster))
-	t.Logf("Found topic %q in cluster %q", topic, c.cluster)
 }
 
 func (c *clusterClients) ExpectNoTopic(ctx context.Context, topic string) {
 	t := framework.T(ctx)
 
 	t.Logf("Checking that topic %q does not exist in cluster %q", topic, c.cluster)
 	c.checkTopic(ctx, topic, false, fmt.Sprintf("Topic %q still exists in cluster %q", topic, c.cluster))
-	t.Logf("Found no topic %q in cluster %q", topic, c.cluster)
 }
 
 // Enable experimental feature support.
@@ -314,7 +312,7 @@ func (c *clusterClients) checkTopic(ctx context.Context, topic string, exists bo
 		require.NoError(t, topics.Error())
 
 		return exists == topics.Has(topic)
-	}, 10*time.Second, 1*time.Second, message) {
+	}, 30*time.Second, 2*time.Second, message) {
 		t.Errorf("Final list of topics: %v", topics.Names())
 	}
 }

operator/internal/controller/pipeline/controller.go

@@ -152,6 +152,11 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 			}
 			// NB: Apply can't be used to remove finalizers.
 			if err := c.Ctl.Update(ctx, pipeline); err != nil {
+				// The object may have been fully deleted between Get and
+				// Update. This is benign — the finalizer is already gone.
+				if apierrors.IsNotFound(err) || apierrors.IsConflict(err) {
+					return ctrl.Result{}, nil
+				}
 				return ctrl.Result{}, err
 			}
 		}