connect

---
name: connect
description: >-
  Teaches how to build streaming data pipelines with Redpanda Connect (formerly
  Benthos): declarative YAML configs with the input/pipeline/output model, processors,
  and Bloblang mappings for transforming messages. Use when creating or editing a
  Connect pipeline config, wiring inputs and outputs (Kafka topics, S3, HTTP,
  databases), writing or debugging Bloblang, or running pipelines with `rpk connect
  run`. For diagnosing a failing pipeline (lint errors, metrics, tracing, dry runs),
  see `/redpanda:connect-debugging`; for change-data-capture connectors, see the
  per-database `/redpanda:connect-cdc-*` skills.
---

# Redpanda Connect

Redpanda Connect (formerly Benthos) is a declarative stream processor: you write a YAML config that specifies an `input`, an optional `pipeline` of processors, and an `output`. Connect reads from the input, runs each record through the processors, and writes to the output — with at-least-once delivery guarantees. It ships as the `redpanda-connect` binary and is also accessible via `rpk connect`.

The component library is large (hundreds of inputs, processors, outputs, caches, buffers, rate limits). This skill teaches the config model, how to run pipelines, how to discover components, Bloblang for transformations, error handling, and several canonical patterns. For CDC-specific pipelines see the `connect-cdc-*` skills; for debugging see `connect-debugging`.

## Quickstart

### 1. Minimal kafka-to-kafka pipeline with a Bloblang transform

```yaml
# my-pipeline.yaml
input:
  redpanda:
    seed_brokers: ["localhost:9092"]
    topics: ["raw-events"]
    consumer_group: connect-demo
    start_offset: earliest

pipeline:
  processors:
    - mapping: |
        root = this
        root.processed_at = now()
        root.event_type = this.type.uppercase()

output:
  redpanda:
    seed_brokers: ["localhost:9092"]
    topic: processed-events
    key: ${! json("id") }
```

Run it:

```bash
# With rpk (recommended)
rpk connect run my-pipeline.yaml

# Or directly with the binary
redpanda-connect run my-pipeline.yaml

# Override a field on the command line (-s flag)
rpk connect run -s input.redpanda.seed_brokers='["broker:9092"]' my-pipeline.yaml

# Load env vars from a .env file
rpk connect run --env-file .env my-pipeline.yaml
```

### 2. Discover available components

```bash
# List all inputs, processors, outputs, etc.
rpk connect list inputs
rpk connect list outputs
rpk connect list processors
rpk connect list caches

# Get the full config schema for a specific component
rpk connect create redpanda          # shows full redpanda input YAML with defaults
rpk connect create //redpanda        # redpanda output template (empty input/processors)
rpk connect create stdin/mapping/stdout  # pipeline with stdin input, mapping processor, stdout output
```

### 3. Lint before deploying

```bash
rpk connect lint my-pipeline.yaml
rpk connect lint --deprecated --labels my-pipeline.yaml
```

### 4. Dry-run (test connections without processing data)

```bash
rpk connect dry-run my-pipeline.yaml
rpk connect dry-run --verbose my-pipeline.yaml
```

## Config Structure

A Connect config has the following top-level keys:

```yaml
# Minimum required keys
input:   { ... }
output:  { ... }

# Optional pipeline of processors
pipeline:
  threads: 1            # parallelism (default 1)
  processors:
    - mapping: "..."

# Optional extras
logger:  { ... }
metrics: { ... }
tracer:  { ... }
buffer:  { ... }

# Named resources — referenced by name from inputs/outputs/processors
cache_resources:      []
rate_limit_resources: []
input_resources:      []
output_resources:     []
processor_resources:  []

# Global Redpanda connection block (shared by redpanda input/output)
redpanda:
  seed_brokers: []
  tls:    { ... }
  sasl:   []
```

Config files are read from well-known paths if no file is given: `redpanda-connect.yaml`, `/redpanda-connect.yaml`, `/etc/redpanda-connect/config.yaml`, `/etc/redpanda-connect.yaml`, `connect.yaml`, `/connect.yaml`, `/etc/connect/config.yaml`, `/etc/connect.yaml`.

## Bloblang Basics

Bloblang is Connect's built-in mapping language used in `mapping`, `mutation`, and `bloblang` processors, and in interpolation strings (`${! ... }`).

```yaml
pipeline:
  processors:
    - mapping: |
        # root = output document, this = input document
        root = this
        root.id = uuid_v4()
        root.ts = now()
        # Delete a field
        root.internal = deleted()
        # Conditional
        root.tier = if this.score > 100 { "premium" } else { "standard" }
        # Array filter
        root.active_users = this.users.filter(u -> u.active == true)
        # Coalesce / fallback
        root.name = this.display_name | this.username | "unknown"
        # Access Kafka metadata
        root.source_topic = meta("kafka_topic")
```

The `mutation` processor mutates messages in-place (more efficient when the output shape is similar to the input). The `mapping` processor creates a fresh output document (safer when the shape changes dramatically). The older name `bloblang` is equivalent to `mapping` and will eventually be deprecated.

## Error Handling

```yaml
# Dead-letter queue using fallback output
output:
  fallback:
    - redpanda:
        seed_brokers: ["localhost:9092"]
        topic: orders
    - retry:
        output:
          redpanda:
            seed_brokers: ["localhost:9092"]
            topic: orders-dlq

# Catch processor errors
pipeline:
  processors:
    - try:
        - mapping: 'root = this.merge({"parsed": this.body.parse_json()})'
    - catch:
        - mapping: 'root.error = error(); root.original = content().string()'
```

## Key Component Groups

| Group | Notable members |
|---|---|
| **Inputs** | `redpanda`, `generate`, `http_server`, `http_client`, `file`, `aws_s3`, `aws_sqs`, `aws_kinesis`, `gcp_pubsub`, `gcp_cloud_storage`, `azure_blob_storage`, `kafka` (community), `sql_select`, `mongodb`, `redis_*`, `nats*`, `amqp_*` |
| **Processors** | `mapping`/`mutation`/`bloblang`, `branch`, `catch`, `try`, `for_each`, `dedupe`, `batch`, `compress`, `decompress`, `schema_registry_decode`, `schema_registry_encode`, `protobuf`, `sql_*`, `cache`, `rate_limit`, `archive`, `split`, `parallel` |
| **Outputs** | `redpanda`, `http_client`, `file`, `broker`, `fallback`, `switch`, `drop`, `cache`, `aws_s3`, `aws_sqs`, `aws_kinesis_firehose`, `gcp_pubsub`, `gcp_bigquery`, `azure_blob_storage`, `opensearch`, `elasticsearch_v8`, `sql_insert` |
| **Caches** | `memory`, `redis`, `ristretto`, `ttlru`, `lru`, `sql`, `aws_dynamodb`, `memcached`, `file` |
| **Buffers** | `memory`, `sqlite`, `system_window`, `none` |
| **Rate Limits** | `local`, `redis` |

Enterprise-only components (require a Redpanda license) include the CDC inputs (`postgres_cdc`, `mysql_cdc`, `mongodb_cdc`, `microsoft_sql_server_cdc`, `oracledb_cdc`, `gcp_spanner_cdc`, `aws_dynamodb_cdc`, `salesforce_cdc`) plus the Snowflake, BigQuery write, Iceberg, Splunk, OpenTelemetry, Slack, Google Drive, and Salesforce families — the complete tiered list is in [Connector Catalog](references/connector-catalog.md). AI/ML processors are certified tier (no license required). Provide a license via `--redpanda-license` flag, `REDPANDA_LICENSE` env var, `REDPANDA_LICENSE_FILEPATH`, or the file `/etc/redpanda/redpanda.license`. One CDC input is *not* enterprise: `tigerbeetle_cdc` is a certified community component (no license needed), but it requires a CGO-enabled Connect build — the `rpk connect` managed plugin and the standard Docker image don't include it.

## Enterprise Features

Redpanda Connect's enterprise (RCL-licensed) features in this domain:

- **Enterprise connectors** — the CDC inputs above plus the Snowflake, BigQuery write, Iceberg, Splunk, OTLP, Slack, Google Drive, and Salesforce families (full list: [Connector Catalog](references/connector-catalog.md)). **Require a license**; blocked after the 30-day trial expires. Each CDC input has nested config (e.g. `oracledb_cdc.logminer{}`, `checkpoint_cache`, `stream_snapshot`). AI/ML processors (`openai_*`, `aws_bedrock_*`, `cohere_*`, `gcp_vertex_ai_*`, `ollama_*`) are certified tier — no license.
- **Allow or deny lists** — restrict which components a pipeline may use via `/etc/redpanda/connector_list.yaml` (`allow:` or `deny:` arrays, mutually exclusive).
- **Secrets management** — resolve secrets from remote systems with the `--secrets` flag (URN schemes `env:`, `redis://`, `aws://`, `gcp://`, `az://`, `none:`).
- **FIPS compliance** — run a FIPS-compliant build of `rpk`/Connect.
- **Configuration service** — the global `redpanda` block can ship Connect's own logs and status events to a Redpanda topic (`logs_topic`, `status_topic`, `pipeline_id`).

Allow/deny lists, secrets management, FIPS, and the configuration service are enterprise features but are **not disabled when a license expires** — only the enterprise connectors are hard-gated. See [Enterprise Features](references/enterprise.md) for every config key.

## Reference Directory

- [Config Structure](references/config-structure.md): Complete pipeline YAML structure, the `redpanda` global block, running pipelines (`rpk connect run`, Docker, `-s` overrides, `--env-file`), and default config file paths.
- [Components](references/components.md): The component model — inputs, processors, outputs, caches, buffers, rate limits — the most-used ones, how to discover and read a component's config schema, enterprise vs community licensing.
- [Bloblang](references/bloblang.md): Bloblang mapping language essentials: `root`/`this`/`meta`, assignment, deletion, variables, conditionals, array methods, functions, `mapping` vs `mutation` vs `bloblang` processor names, and practical transform examples.
- [Patterns](references/patterns.md): Canonical pipelines: kafka→kafka with transform, http_server→kafka ingest, batching, fallback/dead-letter outputs, retries, and windowed aggregation. Runnable YAML.
- [Connector Catalog](references/connector-catalog.md): The tiered component catalog — every enterprise family at the current stable release (Snowflake, BigQuery write, Iceberg, Splunk, OTLP, Slack, Google Drive, Salesforce, CDC), the certified AI/ML + vector-store processors, notable certified families (MQTT, NATS, AMQP, Azure storage, aws_lambda, redpanda_data_transform), tier-vs-license-gate caveats, and live-discovery commands.
- [Migration](references/migration.md): Kafka→Redpanda migration with the unified `redpanda_migrator` input/output pair (Connect 4.67.5+): what it migrates (topic data, topic configs, schemas, ACLs, consumer group offsets), unified vs the removed legacy bundle components, the end-to-end workflow (ACLs → run → monitor lag → verify → cut over), label pairing, sync schedules and guarantees, when to use it vs a plain kafka→redpanda pipeline.
- [Streams Mode](references/streams-mode.md): Running many isolated streams in one Connect process (`rpk connect streams`): per-stream config file layout, the `-o/--observability` and `-r/--resources` flags, shared resources, stream-prefixed HTTP endpoints, the `stream` metrics label, and the `/streams` REST API — which is **disabled by default** and enabled with `--bind-http` (create/read/update/patch/delete streams, `/resources/{type}/{id}`, `?chilled=true`; `GET /ready` stays available either way).
- [Enterprise Features](references/enterprise.md): Enterprise (license-gated) Connect features and their exact config keys — supplying a license (`--redpanda-license`, `REDPANDA_LICENSE`, `/etc/redpanda/redpanda.license`); all CDC inputs with nested blocks (`postgres_cdc`, `mysql_cdc`, `mongodb_cdc`, `oracledb_cdc` `logminer{}`, `microsoft_sql_server_cdc`, `gcp_spanner_cdc`, `aws_dynamodb_cdc`, `salesforce_cdc`); AI/ML processors; allow/deny lists (`connector_list.yaml`); secrets management (`--secrets` URNs); FIPS compliance; and the configuration service (`logs_topic`/`status_topic`). Notes which features require a license vs which survive expiry.

• SKILL.md 12 KB
• references/SOURCES.md 34 KB
• references/bloblang.md 8 KB
• references/components.md 14 KB
• references/config-structure.md 9 KB
• references/connector-catalog.md 9 KB
• references/enterprise.md 19 KB
• references/migration.md 12 KB
• references/patterns.md 10 KB
• references/streams-mode.md 8 KB