Streaming
beta

Run Redpanda Connect Pipelines in Kubernetes

Run Redpanda Connect pipelines as Kubernetes resources managed by the Redpanda Operator. You define each pipeline in a Pipeline custom resource, and the operator deploys it, validates its configuration before startup, rolls it on every change, and reports its health in the resource status. Pipelines follow the same declarative workflow as your Topic and User resources, so you can build and operate streaming data pipelines with the same GitOps tooling you already use for cluster resources.

After reading this page, you will be able to:

  • Enable the Connect controller on the Redpanda Operator

  • Connect pipelines to Redpanda clusters with scoped credentials

  • Inject secrets and cloud IAM identities into pipeline workloads

This feature is in beta and requires Redpanda Operator 26.2 or later. Beta features are not recommended for production environments. To give feedback, reach out to the Redpanda team in Redpanda Community Slack.

This feature requires an enterprise license. To get a trial license key or extend your trial period, generate a new trial license key. To purchase a license, contact Redpanda Sales.

If Redpanda has enterprise features enabled and it cannot find a valid license, restrictions apply.

Your license must include the Redpanda Connect product. The operator treats a license without this entitlement like a missing license and does not deploy Pipeline resources. See License behavior.

When to use the Pipeline resource

You don’t install a separate operator. The Connect controller is part of the Redpanda Operator, and you enable it with a single Helm setting on your existing installation. See Enable the Connect controller.

Use the Pipeline resource when you want:

  • Declarative, GitOps-friendly management: Pipelines are plain Kubernetes manifests that you can version, review, and deploy with tools such as Argo CD or Flux.

  • Validated rollouts: The operator lints every configuration before the pipeline starts, and rolls the workload automatically when the configuration changes.

  • Cluster integration: A pipeline can reference a Redpanda resource in the same namespace, and the operator injects the connection details, TLS material, and credentials for you.

  • Built-in connectors: Redpanda Connect ships hundreds of connectors and processors, including AI integrations, without the JVM or plugin management that Kafka Connect requires.

For Kafka Connect, which is community-supported with Redpanda, see Create and Manage Kafka Connect Connectors in Kubernetes. To run Redpanda Connect on Kubernetes without the operator, see Install or Upgrade with the Helm Chart.

Prerequisites

You must have the following:

  • Kubectl: Ensure you have the kubectl command-line tool installed and configured to communicate with your cluster.

  • Redpanda Operator: Ensure you have the Redpanda Operator deployed in your Kubernetes cluster.

  • Enterprise Edition license: The Connect controller is an enterprise feature and requires a license that includes the Redpanda Connect product. Without one, pipelines are reconciled but never deployed. See License behavior.

  • Image access: Your cluster must be able to pull the Redpanda Connect image (docker.redpanda.com/redpandadata/connect by default).

Enable the Connect controller

The Connect controller ships inside the Redpanda Operator binary and is disabled by default. Enable it through the operator Helm chart, which configures both the controller and the enterprise license:

  1. Create a Secret that holds your Enterprise Edition license:

    kubectl create secret generic redpanda-license \
  --from-file=license=<path-to-license-file> \
  --namespace <namespace>

  2. Install or upgrade the operator with the Connect controller enabled:

    helm upgrade --install redpanda-operator redpanda/operator \
  --namespace <namespace> \
  --set connectController.enabled=true \
  --set enterprise.licenseSecretRef.name=redpanda-license \
  --set enterprise.licenseSecretRef.key=license

    connectController.enabled: true adds the --enable-connect flag to the operator Deployment, and enterprise.licenseSecretRef mounts the license Secret into the operator Pod. To also create a PodMonitor for each pipeline, for example to scrape pipeline metrics with Prometheus Operator, set connectController.monitoring.enabled=true.

  3. Verify that the Pipeline CRD is installed:

    kubectl get crd pipelines.cluster.redpanda.com

Deploy your first pipeline

You define a pipeline entirely in the spec.configYaml field, which holds a standard Redpanda Connect configuration. This minimal example generates data and writes it to the Pod’s standard output, so you can try the workflow without a Redpanda cluster:

  1. Create a file called hello-pipeline.yaml:

    hello-pipeline.yaml
    apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: hello
  namespace: <namespace>
spec:
  configYaml: |
    input:
      generate:
        interval: 1s
        mapping: 'root.msg = "hello from connect"'
    output:
      stdout: {}

  2. Apply the manifest and watch the pipeline come up:

    kubectl apply -f hello-pipeline.yaml
kubectl get pipeline hello --namespace <namespace> --watch

    Example output:

    NAME    READY   PHASE     REPLICAS   AVAILABLE   AGE
hello   True    Running   1          1           20s

  3. Inspect the pipeline’s output:

    kubectl logs deploy/hello -c connect --namespace <namespace> -f

When you apply a Pipeline resource, the operator:

  1. Renders a ConfigMap that holds the final connect.yaml.

  2. Renders a Deployment whose lint init container runs redpanda-connect lint against the configuration. The pipeline only starts if the configuration passes lint. A failing configuration surfaces as ConfigValid: False in the resource status.

  3. Stamps the Pod template with a cluster.redpanda.com/config-checksum annotation, so any later configuration change automatically rolls the Deployment. A second annotation, cluster.redpanda.com/credentials-checksum, tracks every referenced Secret and ConfigMap, so credential rotations also roll the pipeline.

To stop a pipeline without deleting it, pause it. The workload scales to zero and the phase changes to Stopped:

kubectl patch pipeline hello --namespace <namespace> --type merge -p '{"spec":{"paused":true}}'

Connect a pipeline to a Redpanda cluster

Production pipelines read from or write to a Redpanda cluster. Bind the pipeline to a Redpanda resource with spec.cluster.clusterRef, and give it SASL credentials with a User resource referenced through spec.userRef. The operator resolves the broker addresses, TLS material, and SASL credentials from the referenced resources and injects them into the rendered configuration, so your configYaml contains only the pipeline logic.

This example deploys a pipeline that reads the orders topic and archives each order to S3.

  1. Create a User resource for the pipeline. The user owns the pipeline’s SASL identity, and its ACLs scope what the pipeline can access:

    orders-to-warehouse-user.yaml
    apiVersion: cluster.redpanda.com/v1alpha2
kind: User
metadata:
  name: orders-to-warehouse
  namespace: <namespace>
spec:
  cluster:
    clusterRef:
      name: <cluster-name>
  authentication:
    type: scram-sha-512
    password:
      valueFrom:
        secretKeyRef:
          name: orders-to-warehouse-password
          key: password
  authorization:
    acls:
      - type: allow
        resource:
          type: topic
          name: orders
          patternType: literal
        operations: [Read, Describe]
      - type: allow
        resource:
          type: group
          name: orders-to-warehouse-ingest
          patternType: literal
        operations: [Read, Describe]

    Literal ACLs fit a single, static pipeline like this one. Pipelines whose configuration creates topics (for example, change data capture producing mysql.* topics) or consumes with dynamic consumer groups need patternType: prefixed grants instead. With a non-superuser pipeline user, a missing group ACL surfaces only as a consumer authorization error at runtime, so scope these grants up front.

  2. Create the Pipeline resource, referencing both the cluster and the user:

    orders-to-warehouse-pipeline.yaml
    apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: orders-to-warehouse
  namespace: <namespace>
spec:
  cluster:
    clusterRef:
      name: <cluster-name>       # The operator resolves brokers and TLS.
  userRef:
    name: orders-to-warehouse    # The operator resolves the password Secret and SASL mechanism.
  valueSources:
    - name: S3_SECRET_KEY
      source:
        secretKeyRef:
          name: s3-creds
          key: secret_access_key
  configYaml: |
    input:
      redpanda:
        # Only the pipeline logic. The operator injects seed_brokers,
        # tls, and sasl from clusterRef and userRef.
        topics: [orders]
        consumer_group: orders-to-warehouse-ingest
    pipeline:
      processors:
        - mapping: |
            root = this
            root.ingested_at = now()
    output:
      aws_s3:
        bucket: warehouse-orders
        region: us-east-2
        credentials:
          secret: ${S3_SECRET_KEY}

  3. Apply both manifests and verify the pipeline:

    kubectl apply -f orders-to-warehouse-user.yaml -f orders-to-warehouse-pipeline.yaml
kubectl get pipeline orders-to-warehouse --namespace <namespace>

What the operator injects

The operator merges the resolved connection into every input.redpanda and output.redpanda block in your configuration, and additionally renders the connection as the top-level redpanda block (the shared client), so redpanda_common components also work without inline credentials. The rendered configuration for the example above looks like this:

input:
  redpanda:
    seed_brokers:                                      # Injected
      - redpanda-0.redpanda.<namespace>.svc.cluster.local.:9093
      - redpanda-1.redpanda.<namespace>.svc.cluster.local.:9093
      - redpanda-2.redpanda.<namespace>.svc.cluster.local.:9093
    tls:                                               # Injected
      enabled: true
      root_cas_file: /etc/tls/certs/ca/ca.crt
    sasl:                                              # Injected
      - mechanism: SCRAM-SHA-512
        username: ${REDPANDA_SASL_USERNAME}
        password: ${REDPANDA_SASL_PASSWORD}
    topics: [orders]                                   # User-supplied
    consumer_group: orders-to-warehouse-ingest         # User-supplied
# ...plus the same connection rendered as the top-level redpanda block,
# and your pipeline and output sections, unchanged.

The SASL environment variables resolve from the User resource: the username is the User’s name, the mechanism comes from spec.authentication.type, and the password is projected from the User’s password Secret.

Keys that you set yourself always win. For example, if you write seed_brokers inside input.redpanda, the operator leaves your value untouched and injects only the missing keys. This is the escape hatch for pipelines that point one input or output at a different cluster. Blocks for other connectors (aws_s3, mongodb, and so on) pass through untouched, so new Connect connectors work without operator changes.

If you omit userRef, the operator falls back to the SASL identity of the cluster source (for clusterRef, the cluster’s bootstrap user). Use a dedicated User with scoped ACLs in production so each pipeline authenticates with least-privilege credentials.

Connect to an external cluster

To target a cluster that isn’t managed by a Redpanda resource in the same namespace, such as a Redpanda Cloud cluster, a self-managed deployment outside Kubernetes, or another Kafka-compatible cluster, replace clusterRef with cluster.staticConfiguration and declare the connection inline. The injection contract is identical to clusterRef. Only the source of the connection details differs.

cross-region-mirror.yaml
apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: cross-region-mirror
  namespace: <namespace>
spec:
  cluster:
    staticConfiguration:
      kafka:
        brokers:
          - kafka.us-east.example.com:9094
        tls:
          caCert:                      # Omit for publicly-issued certificates.
            secretKeyRef:
              name: external-kafka-ca
              key: ca.crt
        sasl:
          mechanism: SCRAM-SHA-512     # PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512
          username: pipeline-mirror-svc
          password:
            secretKeyRef:
              name: external-kafka-creds
              key: password
  configYaml: |
    input:
      redpanda:
        topics: [orders]
    output:
      redpanda:
        topic: orders.mirrored

Note the following:

  • staticConfiguration and userRef are mutually exclusive: the static configuration carries its own SASL identity, so the API server rejects a Pipeline that sets both. userRef also requires clusterRef, because without a referenced cluster there is no cluster to authenticate against.

  • The CA certificate can come from a Secret (secretKeyRef) or a ConfigMap (configMapKeyRef). For mutual TLS listeners, additionally set tls.cert and tls.key to present a client certificate.

  • Setting any TLS certificate material enables TLS. tls: {enabled: true} alone requests TLS with publicly-issued certificates, and omitting the tls block connects without TLS.

  • The SASL password also accepts an inline value. Avoid inline passwords outside development: the value is stored in plain text in the Pipeline spec.

Give a pipeline a cloud IAM identity

Some connectors call cloud APIs directly, for example to read from Amazon RDS with Identity and Access Management (IAM) database authentication, or to write to S3 without static keys. Set spec.serviceAccountName to run the pipeline Pod under a dedicated Kubernetes ServiceAccount whose cloud IAM trust is scoped to that one pipeline: IAM Roles for Service Accounts (IRSA) on EKS, Workload Identity on GKE, or Pod Identity on AKS. The operator does not create the ServiceAccount. Provision it, with its cloud IAM annotations, alongside your other infrastructure.

The pipeline’s cluster credentials (cluster and userRef) and its cloud identity (serviceAccountName) are independent trust boundaries: either can be set without the other.

This example runs MySQL change data capture against an Amazon RDS database using IAM authentication, so no database password exists anywhere in the Pipeline, the Pod, or a Secret:

  1. Create the ServiceAccount with the IAM role annotation. The IAM role’s trust policy must be scoped to this exact ServiceAccount, and its policy must grant rds-db:connect:

    mysql-cdc-orders-sa.yaml
    apiVersion: v1
kind: ServiceAccount
metadata:
  name: mysql-cdc-orders-rds
  namespace: <namespace>
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::<account-id>:role/mysql-cdc-orders-rds

  2. Create the Pipeline with serviceAccountName set:

    mysql-cdc-orders.yaml
    apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: mysql-cdc-orders
  namespace: <namespace>
spec:
  cluster:
    clusterRef:
      name: <cluster-name>
  userRef:
    name: mysql-cdc-orders-svc               # SASL identity for the Redpanda output
  serviceAccountName: mysql-cdc-orders-rds   # Cloud IAM boundary for AWS calls
  valueSources:
    # No MYSQL_PASSWORD: IAM authentication supplies a short-lived token.
    - name: MYSQL_HOST
      source:
        secretKeyRef:
          name: mysql-cdc-creds
          key: host
    - name: MYSQL_USER
      source:
        secretKeyRef:
          name: mysql-cdc-creds
          key: username
  configYaml: |
    input:
      mysql_cdc:
        # Connect calls rds:GenerateDBAuthToken with the Pod's IAM role and
        # uses the 15-minute token as the MySQL password.
        dsn: "${MYSQL_USER}@tcp(${MYSQL_HOST}:3306)/shop?tls=skip-verify&allowCleartextPasswords=1"
        aws:
          enabled: true
          region: us-east-2
          endpoint: "${MYSQL_HOST}:3306"
        stream_snapshot: true
        tables: [orders]
        flavor: mysql
    output:
      redpanda:
        topic: mysql.shop.orders
        key: '${! @table }'

The Pod assumes only the annotated role, through the projected ServiceAccount token. userRef still gates access to Redpanda, while the ServiceAccount gates access to AWS.

Inject values and secrets with valueSources

Use spec.valueSources to project named values into the pipeline Pod as environment variables, which your configuration references with ${NAME} interpolation. Each entry is a single, typed pull from one of four sources, and you can mix sources freely in one list:

  • inline: A literal value in the Pipeline spec.

  • secretKeyRef: One key from a Kubernetes Secret.

  • configMapKeyRef: One key from a ConfigMap.

  • externalSecretRef: A secret resolved through the External Secrets Operator, backed by a system such as Vault or 1Password.

Because every value is a named, per-key pull, unused keys in a Secret never leak into the Pod environment, the environment variable name can differ from the Secret’s key, and multiple pipelines can pull different keys from the same Secret.

This example fans one topic out to MongoDB, Snowflake, and MySQL, drawing credentials from all four source types:

orders-fanout.yaml
apiVersion: cluster.redpanda.com/v1alpha2
kind: Pipeline
metadata:
  name: orders-fanout
  namespace: <namespace>
spec:
  cluster:
    clusterRef:
      name: <cluster-name>
  userRef:
    name: orders-fanout-svc
  valueSources:
    # MongoDB Atlas: full connection URI from a Secret.
    - name: MONGO_URI
      source:
        secretKeyRef:
          name: mongo-orders-atlas
          key: connection_uri
    # Snowflake key-pair auth: account and user inline, private key from a
    # Secret, passphrase from an External Secrets Operator resource.
    - name: SNOWFLAKE_ACCOUNT
      source:
        inline: "ab12345.us-east-1"
    - name: SNOWFLAKE_USER
      source:
        inline: "PIPELINE_SVC"
    - name: SNOWFLAKE_PRIVATE_KEY
      source:
        secretKeyRef:
          name: snowflake-pipeline-svc
          key: rsa_key.p8
    - name: SNOWFLAKE_PRIVATE_KEY_PASSPHRASE
      source:
        externalSecretRef:
          name: snowflake-pipeline-svc-passphrase
    # MySQL: host from a ConfigMap so the endpoint can rotate without
    # touching the Pipeline resource.
    - name: MYSQL_USER
      source:
        inline: "warehouse_writer"
    - name: MYSQL_PASSWORD
      source:
        secretKeyRef:
          name: mysql-warehouse-creds
          key: password
    - name: MYSQL_HOST
      source:
        configMapKeyRef:
          name: warehouse-env
          key: mysql_replica_host
  configYaml: |
    input:
      redpanda:
        topics: [orders]
        consumer_group: orders-fanout
    output:
      broker:
        outputs:
          - mongodb:
              url: ${MONGO_URI}
              database: orders
              collection: ingested
              operation: insert-one
          - snowflake_put:
              account: ${SNOWFLAKE_ACCOUNT}
              user: ${SNOWFLAKE_USER}
              private_key: ${SNOWFLAKE_PRIVATE_KEY}
              private_key_pass: ${SNOWFLAKE_PRIVATE_KEY_PASSPHRASE}
              database: WAREHOUSE
              schema: PUBLIC
              stage: "@PIPELINE_STAGE"
          - sql_insert:
              driver: mysql
              dsn: "${MYSQL_USER}:${MYSQL_PASSWORD}@tcp(${MYSQL_HOST}:3306)/warehouse?parseTime=true"
              table: orders
              columns: [order_id, ingested_at, payload]
              args_mapping: |
                root = [ this.id, this.ingested_at, this.format_json() ]

Resolution of every entry is reported by the ValueSourcesResolved status condition.

Customize the pipeline workload

The Pipeline spec exposes standard workload settings:

spec:
  replicas: 3                        # Default 1. Autoscalable through the scale subresource.
  paused: true                       # Scale to zero without deleting the resource.
  image: docker.redpanda.com/redpandadata/connect:<version>   # Override the Connect image.
  resources:
    requests:
      cpu: 100m
      memory: 256Mi
    limits:
      cpu: "1"
      memory: 1Gi
  serviceAccountName: my-pipeline-sa # Pod identity. See the cloud IAM section.
  nodeSelector:
    redpanda.com/node-pool: connect  # Pin pipelines to a dedicated node pool.
  tolerations:
    - key: dedicated
      operator: Equal
      value: connect
      effect: NoSchedule
  affinity: {}                       # Standard Kubernetes affinity, merged with zones.
  zones: ["us-east-2a", "us-east-2b"]  # Spread Pods across availability zones.
  budget:
    maxUnavailable: 1                # Creates a PodDisruptionBudget.

The Connect image is resolved with the following precedence: spec.image on the Pipeline, then the operator chart’s connectController.image values, then the default image baked into the operator binary.

If a pipeline needs setup work before it starts, such as fetching certificates or waiting for a dependency, add extraInitContainers. They run to completion, in order, before the operator’s built-in lint init container, so anything they stage into a shared volume is visible to both lint and the Connect runtime. Declare the backing volume in extraVolumes and mount it into the lint and Connect containers with extraVolumeMounts:

spec:
  extraVolumes:
    - name: shared
      emptyDir: {}
  extraVolumeMounts:                 # Mounted into the lint and connect containers.
    - name: shared
      mountPath: /shared
      readOnly: true
  extraInitContainers:
    - name: fetch-certs
      image: curlimages/curl:8.11.0
      command: ["sh", "-c", "curl -fsSL $CERT_URL -o /shared/ca.pem"]
      volumeMounts:
        - name: shared
          mountPath: /shared

The volume names config, cluster-tls-ca, and cluster-tls-client are reserved by the operator.

Autoscale pipelines

The Pipeline resource implements the Kubernetes scale subresource, so kubectl scale, a HorizontalPodAutoscaler, or a KEDA ScaledObject can drive the replica count directly. For step-by-step HPA and KEDA examples, see Autoscale Redpanda Connect Pipelines.

Monitor pipelines

Redpanda Connect serves its metrics on each pipeline Pod’s http port (4195) at /metrics. If you run Prometheus Operator, set connectController.monitoring.enabled=true in the operator chart to render a PodMonitor for every pipeline, so metrics such as input_received and output_sent are scraped automatically. These metrics can also drive autoscaling. See Autoscale Redpanda Connect Pipelines.

License behavior

The Connect controller requires a valid Enterprise Edition license that includes the Redpanda Connect product to create or update pipeline workloads:

  • No license, expired license, or a license without the Redpanda Connect product: New Pipeline resources are reconciled but get no workload. The resource reports License: False in its status conditions and never reaches the Running phase. Pipelines that were already running keep processing data, but the operator stops applying spec updates to them until you configure a valid license.

  • License restored: Reconciliation resumes automatically. You do not need to recreate Pipeline resources.

The operator mirrors the license into each pipeline’s namespace as a Pipeline-owned Secret named <pipeline-name>-license, injected into the Pod as REDPANDA_LICENSE, so the Connect runtime’s own license checks pass without configuring the license twice. Anyone with permission to read Secrets in a namespace that hosts pipelines can read the license, so scope namespace RBAC accordingly.

For how licensing works across Redpanda products, see Redpanda Licenses and Enterprise Features.

Check pipeline status

Inspect a pipeline’s phase and conditions with the following command:

kubectl get pipeline <pipeline-name> --namespace <namespace> -o yaml

The .status.phase field summarizes the pipeline state:

Phase Meaning

Pending

The resource is accepted but not yet acted on.

Provisioning

The Deployment is created, but Pods are not yet ready.

Running

The desired number of replicas is ready.

Stopped

The pipeline is paused or scaled to zero.

Unknown

The state could not be determined.

The .status.conditions array reports each aspect of reconciliation:

Condition True means

Ready

The pipeline workload is healthy. This is the headline condition.

ConfigValid

The Connect configuration passed lint.

ClusterRef

The cluster source (clusterRef or staticConfiguration) was resolved.

UserRef

The referenced User’s credentials, including its password Secret, were resolved.

ValueSourcesResolved

Every valueSources entry resolved to an existing object and key.

License

The operator’s enterprise license is valid and includes Redpanda Connect.

Two operational guarantees are worth knowing:

  • Pipeline Pods do not depend on the operator process. The Deployment is owned by the Pipeline resource, not the operator, so operator restarts and upgrades never interrupt running pipelines.

  • Failures never tear down a running workload. If a referenced Redpanda resource, User, Secret, ConfigMap, or the license becomes unresolvable, the controller reports the failing condition on the status and stops syncing, but the last-known-good Deployment keeps running. Only deleting the Pipeline resource removes its workload.

Troubleshoot

Use this table to diagnose common failure conditions:

Symptom Cause and resolution

License: False

No valid license, or the license does not include the Redpanda Connect product. Configure enterprise.licenseSecretRef in the operator chart. See License behavior.

ConfigValid: False

The lint init container rejected the configuration. Run kubectl logs <pod> -c lint to see the lint error. The condition message carries a truncated copy.

ClusterRef: False

The referenced Redpanda resource doesn’t exist in the pipeline’s namespace, or its CRD isn’t installed. A clusterRef cannot point to another namespace.

Pod stuck in Init or ImagePullBackOff

The cluster can’t pull the Connect image or an extraInitContainers image. Check the image reference and pull secrets.

redpanda_common errors with shared client not found

The pipeline isn’t bound to a cluster. Set cluster.clusterRef (with userRef for SASL clusters) so the operator renders the top-level redpanda block.

Ready: False with reason NameConflict

Another workload already owns an object with this pipeline’s name, such as a ConfigMap or Deployment. Rename the pipeline or remove the conflicting object. The operator refuses to adopt objects it does not own.

Next steps

Suggested labs

Search all labs