Add Pipeline CRD for Redpanda Connect pipeline management

.changes/unreleased/operator-Added-20260323-connect-crd.yaml

@@ -0,0 +1,2 @@
+kind: Added
+body: Add Pipeline CRD and controller for managing Redpanda Connect pipelines via the operator, gated by enterprise license validation.

.changes/unreleased/operator-Changed-20260323-connect-default.yaml

@@ -0,0 +1,17 @@
+project: operator
+kind: Changed
+body: |
+  The Redpanda Connect controller can now be enabled via the operator Helm chart
+  by setting `connectController.enabled: true`. The controller is disabled by default.
+  The enterprise license is configured at the operator level via
+  `enterprise.licenseSecretRef` in the Helm chart values.
+
+  To enable the Connect controller:
+
+  ```
+  helm install redpanda-operator redpanda/operator \
+    --set connectController.enabled=true \
+    --set enterprise.licenseSecretRef.name=redpanda-license \
+    --set enterprise.licenseSecretRef.key=license
+  ```
+time: 2026-03-23T15:00:00.000000-04:00

acceptance/features/console-upgrades.feature

@@ -2,7 +2,7 @@
 Feature: Upgrading the operator with Console installed
   @skip:gke @skip:aks @skip:eks
   Scenario: Console v2 to v3 no warnings
-    Given I helm install "redpanda-operator" "redpanda/operator" --version v25.1.3 with values:
+    Given I helm install "redpanda-operator" "redpanda/operator" --version v25.3.1 with values:
     """
     image:
       repository: redpandadata/redpanda-operator
@@ -49,7 +49,7 @@ Feature: Upgrading the operator with Console installed
 
   @skip:gke @skip:aks @skip:eks
   Scenario: Console v2 to v3 with warnings
-    Given I helm install "redpanda-operator" "redpanda/operator" --version v25.1.3 with values:
+    Given I helm install "redpanda-operator" "redpanda/operator" --version v25.3.1 with values:
     """
     image:
       repository: redpandadata/redpanda-operator

acceptance/features/operator-upgrades.feature

@@ -1,8 +1,8 @@
 @vcluster
 Feature: Upgrading the operator
   @skip:gke @skip:aks @skip:eks
-  Scenario: Operator upgrade from 25.2.2
-    Given I helm install "redpanda-operator" "redpanda/operator" --version v25.2.2 with values:
+  Scenario: Operator upgrade from 25.3.1
+    Given I helm install "redpanda-operator" "redpanda/operator" --version v25.3.1 with values:
     """
     image:
       repository: redpandadata/redpanda-operator

acceptance/features/pipeline-crds.feature

@@ -0,0 +1,263 @@
+@cluster:basic
+Feature: Pipeline CRDs
+  Background: Cluster available
+    Given cluster "basic" is available
+
+  Scenario: Create and run a Pipeline
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: demo-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello world"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    Then pipeline "demo-pipeline" is successfully running
+
+  Scenario: Delete a Pipeline
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: delete-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    And pipeline "delete-pipeline" is successfully running
+    When I delete the CRD pipeline "delete-pipeline"
+    Then pipeline "delete-pipeline" does not exist
+
+  Scenario: Update a Pipeline config
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: update-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "original"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    And pipeline "update-pipeline" is successfully running
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: update-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "updated"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    Then pipeline "update-pipeline" is successfully running
+
+  Scenario: Stop a Pipeline
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: stop-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    And pipeline "stop-pipeline" is successfully running
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: stop-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+      paused: true
+    """
+    Then pipeline "stop-pipeline" is stopped
+
+  Scenario: Resume a stopped Pipeline
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: resume-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    And pipeline "resume-pipeline" is successfully running
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: resume-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+      paused: true
+    """
+    And pipeline "resume-pipeline" is stopped
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: resume-pipeline
+    spec:
+      configYaml: |
+        input:
+          generate:
+            mapping: 'root.message = "hello"'
+            interval: "5s"
+        output:
+          stdout: {}
+      replicas: 1
+      paused: false
+    """
+    Then pipeline "resume-pipeline" is successfully running
+
+  Scenario: Invalid Pipeline config detected by lint
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: invalid-pipeline
+    spec:
+      configYaml: |
+        input:
+          not_a_real_input:
+            mapping: 'root = "broken"'
+        output:
+          stdout: {}
+      replicas: 1
+    """
+    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:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: producer-pipeline
+    spec:
+      cluster:
+        clusterRef:
+          name: basic
+      configYaml: |
+        input:
+          generate:
+            count: 0
+            interval: "1s"
+            mapping: 'root.message = "hello from pipeline"'
+        output:
+          redpanda:
+            seed_brokers:
+              - "${RPK_BROKERS}"
+            tls:
+              enabled: ${RPK_TLS_ENABLED}
+              root_cas_file: "${RPK_TLS_ROOT_CAS_FILE}"
+            topic: "pipeline-produce-test"
+      replicas: 1
+    """
+    Then pipeline "producer-pipeline" is successfully running
+    And topic "pipeline-produce-test" has messages in cluster "basic"
+
+  Scenario: Pipeline reads from Redpanda via clusterRef
+    Given I create topic "pipeline-consume-test" in cluster "basic"
+    And I produce messages to "pipeline-consume-test" in cluster "basic"
+    When I apply Kubernetes manifest:
+    """
+    ---
+    apiVersion: cluster.redpanda.com/v1alpha2
+    kind: Pipeline
+    metadata:
+      name: consumer-pipeline
+    spec:
+      cluster:
+        clusterRef:
+          name: basic
+      configYaml: |
+        input:
+          redpanda:
+            seed_brokers:
+              - "${RPK_BROKERS}"
+            tls:
+              enabled: ${RPK_TLS_ENABLED}
+              root_cas_file: "${RPK_TLS_ROOT_CAS_FILE}"
+            topics:
+              - "pipeline-consume-test"
+            consumer_group: "pipeline-consumer-group"
+        output:
+          drop: {}
+      replicas: 1
+    """
+    Then pipeline "consumer-pipeline" is successfully running

acceptance/go.sum

@@ -604,6 +604,8 @@ github.com/redpanda-data/common-go/goldenfile v0.0.0-20260109170727-1dd9f5d22ee1
 github.com/redpanda-data/common-go/goldenfile v0.0.0-20260109170727-1dd9f5d22ee1/go.mod h1:V3OBV2kcF/BDDytUZuKvIygbaXoGPT5VO3KmMAz+mBM=
 github.com/redpanda-data/common-go/kube v0.0.0-20260408144400-efba9928bb27 h1:735zfoMDegKzXO+mipLeEJhjvoboMiOwlZiEfWTs9IY=
 github.com/redpanda-data/common-go/kube v0.0.0-20260408144400-efba9928bb27/go.mod h1:87/jKBvBse9m7PBwCxzISdxOpHblNKTqxZZNa1U1utM=
+github.com/redpanda-data/common-go/license v0.0.0-20260120073450-935d3dd3d6c1 h1:6aPxMthcrAljux5bgqU78yHxM8BK1ITqh9G9H+s707U=
+github.com/redpanda-data/common-go/license v0.0.0-20260120073450-935d3dd3d6c1/go.mod h1:F1fp8xVNS2UwWFosOjJ9+5jaEZnXSjB9AdHk2R9XlpI=
 github.com/redpanda-data/common-go/net v0.1.1-0.20240429123545-4da3d2b371f7 h1:MXLdjFdFjOtyuUR4TdVVsqFP8xnru2YDwzH9bJTUr1M=
 github.com/redpanda-data/common-go/net v0.1.1-0.20240429123545-4da3d2b371f7/go.mod h1:UJIi/yUxGOBYXUrfUsOkxfYxcb/ll7mZrwae/i+U2kc=
 github.com/redpanda-data/common-go/otelutil v0.0.0-20260109170727-1dd9f5d22ee1 h1:fLBU7IJZq8PR88AkEdI5G+yyBsftionGapvAOMPW9hg=