Redpanda Connect Documentation

The Redpanda Unified Migrator is a comprehensive data migration system designed for Apache Kafka and Redpanda clusters. It orchestrates three specialized sub-migrators to provide complete cluster-to-cluster migration, covering topics, schemas, and consumer groups.

Core Components

• Migrator: The central coordinator managing input/output lifecycles, message transformations, and timing of sub-migrator operations.
• topicMigrator: Handles topic infrastructure, including name interpolation, partition mirroring, configuration copying, and optional ACL replication.
• schemaRegistryMigrator: Synchronizes schemas and compatibility settings, supporting regex-based filtering and various ID translation modes.
• groupsMigrator: Manages consumer group offset translation using timestamp correlation and embedded offset headers.

Use the mqtt input component to subscribe to topics on MQTT brokers. This component adds specific metadata to each message which can be accessed via Bloblang interpolation.

Metadata fields added to each message:

• mqtt_duplicate
• mqtt_qos
• mqtt_retained
• mqtt_topic
• mqtt_message_id

The redpanda_migrator input is a Kafka consumer designed for migration pipelines. It consumes records from a source cluster and forwards them downstream.

IMPORTANT: This input requires a corresponding redpanda_migrator output in the same pipeline. The input itself does not handle topic, schema, or group synchronization; all migration coordination logic resides in the paired output component. Introduced in version 4.67.0.

The otlp_http input component exposes an OpenTelemetry Collector HTTP receiver that accepts traces, logs, and metrics via the OTLP/HTTP protocol. It converts incoming OTLP data (protobuf or JSON) into individual Redpanda OTEL v1 messages. Each signal (span, log record, or metric) is unbatched into a separate message with embedded Resource and Scope metadata.

Endpoints:

• /v1/traces - OpenTelemetry traces
• /v1/logs - OpenTelemetry logs
• /v1/metrics - OpenTelemetry metrics

Supported Protocols:

• application/x-protobuf
• application/json

Output Format Metadata: Each message includes:

• otel_signal_type: "trace", "log", or "metric"
• otel_encoding: "json" or "protobuf"

The Couchbase processor performs operations against a Couchbase cluster for each message, allowing you to store or retrieve data within message payloads.

Note: When performing insert, replace, or upsert operations, the content property must be set.

Introduced in version 4.11.0.

The cached processor caches the result of applying one or more processors to messages identified by a key. If the key exists in the cache, the message content is replaced with the cached result, bypassing the processors. This is ideal for reducing the cost of expensive operations (like HTTP requests) that produce consistent results for specific inputs.

Note: The cache uses a custom versioned schema. You cannot point this processor to a cache pre-populated with data from other sources.

pipeline:
  processors:
    - cached:
        key: my_key
        cache: my_cache
        processors:
          - http:
              url: http://example.com/api
              verb: GET

The Parquet input reads and decodes Apache Parquet files into a stream of structured messages. It was introduced in version 4.8.0.

By default, BYTE_ARRAY or FIXED_LEN_BYTE_ARRAY values are extracted as byte slices ([]byte). If the logical type is UTF8, they are extracted as string. When these byte slices are later serialized to JSON, they are base64 encoded by default. You can convert these to strings or other types using Bloblang transformations (e.g., root.foo = this.foo.string()).