redpanda

The Redpanda Connect configuration service allows you to:

Configure Redpanda cluster credentials in a single configuration block, which is referenced by multiple components in data pipeline. For more information, see the Pipeline example.
Send logs and status updates to topics on a Redpanda cluster, in addition to the default logger.

The redpanda namespace contains the configuration of this service.

Common
Advanced

# Common configuration fields, showing default values
redpanda:
  seed_brokers: [] # No default (optional)
  pipeline_id: ""
  logs_topic: ""
  logs_level: info
  status_topic: ""

yml

# All configuration fields, showing default values
redpanda:
  seed_brokers: [] # No default (optional)
  client_id: benthos
  tls:
    enabled: false
    skip_cert_verify: false
    enable_renegotiation: false
    root_cas: ""
    root_cas_file: ""
    client_certs: []
  sasl: [] # No default (optional)
  metadata_max_age: 5m
  request_timeout_overhead: 10s
  conn_idle_timeout: 20s
  pipeline_id: ""
  logs_topic: ""
  logs_level: info
  status_topic: ""
  partitioner: "" # No default (optional)
  idempotent_write: true
  compression: "" # No default (optional)
  timeout: 10s
  max_message_bytes: 1MB
  broker_write_max_bytes: 100MB
  allow_auto_topic_creation: true

yml

Pipeline example

This data pipeline reads data from topic_A and topic_B on a Redpanda cluster, and then writes the data to topic_C on the same cluster. The cluster details are configured within the redpanda configuration block, so you only need to configure them once. This is a useful feature when you have multiple inputs and outputs in the same data pipeline that need to connect to the same cluster.

input:
  redpanda_common:
    topics: [ topic_A, topic_B ]

output:
  redpanda_common:
    topic: topic_C
    key: ${! @id }

redpanda:
  seed_brokers: [ "127.0.0.1:9092" ]
  tls:
    enabled: true
  sasl:
    - mechanism: SCRAM-SHA-512
      password: bar
      username: foo

Fields

`seed_brokers`

A list of broker addresses to connect to in order. Use commas to separate multiple addresses in a single list item.

Type: array

# Examples

seed_brokers:
  - localhost:9092

seed_brokers:
  - foo:9092
  - bar:9092

seed_brokers:
  - foo:9092,bar:9092

yml

`client_id`

An identifier for the client connection.

Type: string

Default: benthos

`tls`

Override system defaults with custom TLS settings.

Type: object

`tls.enabled`

Whether custom TLS settings are enabled.

Type: bool

Default: false

`tls.skip_cert_verify`

Whether to skip server-side certificate verification.

Type: bool

Default: false

`tls.enable_renegotiation`

Whether to allow the remote server to request renegotiation. Enable this option if you’re seeing the error message local error: tls: no renegotiation.

Type: bool

Default: false

`tls.root_cas`

Specify a root certificate authority to use (optional). This is a string that represents a certificate chain from the parent trusted root certificate, through possible intermediate signing certificates, to the host certificate.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

# Example

root_cas: |-
  -----BEGIN CERTIFICATE-----
  ...
  -----END CERTIFICATE-----

yml

`tls.root_cas_file`

Specify the path to a root certificate authority file (optional). This is a file, often with a .pem extension, which contains a certificate chain from the parent trusted root certificate, through possible intermediate signing certificates, to the host certificate.

Type: string

Default: ""

# Example

root_cas_file: ./root_cas.pem

yml

`tls.client_certs`

A list of client certificates to use. For each certificate, specify either the fields cert and key or cert_file and key_file.

Type: array

Default: []

# Examples

client_certs:
  - cert: foo
    key: bar

client_certs:
  - cert_file: ./example.pem
    key_file: ./example.key

yml

`tls.client_certs[].cert`

The plain text certificate to use.

Type: string

Default: ""

`tls.client_certs[].key`

The plain text certificate key to use.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`tls.client_certs[].cert_file`

The path of a certificate to use.

Type: string

Default: ""

`tls.client_certs[].key_file`

The path of a certificate key to use.

Type: string

Default: ""

`tls.client_certs[].password`

The plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete pbeWithMD5AndDES-CBC algorithm is not supported for the PKCS#8 format.

The pbeWithMD5AndDES-CBC algorithm does not authenticate ciphertext, and is vulnerable to padding oracle attacks which may allow an attacker to recover the plain text password.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

# Examples

password: foo

password: ${KEY_PASSWORD}

yml

`sasl`

Specify one or more methods or mechanisms of SASL authentication. They are tried in order. If the broker supports the first SASL mechanism, all connections use it. If the first mechanism fails, the client picks the first supported mechanism. If the broker does not support any client mechanisms, all connections fail.

Type: array

# Example

sasl:
  - mechanism: SCRAM-SHA-512
    password: bar
    username: foo

yml

`sasl[].mechanism`

The SASL mechanism to use.

Type: string

Option

Summary

AWS_MSK_IAM

AWS IAM-based authentication as specified by the aws-msk-iam-auth Java library.

OAUTHBEARER

OAuth Bearer-based authentication.

PLAIN

Plain text authentication.

SCRAM-SHA-256

SCRAM-based authentication as specified in RFC5802.

SCRAM-SHA-512

SCRAM-based authentication as specified in RFC5802.

none

Disable SASL authentication

`sasl[].username`

A username for PLAIN or SCRAM-* authentication.

Type: string

Default: ""

`sasl[].password`

A password for PLAIN or SCRAM-* authentication.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`sasl[].token`

The token to use for a single session’s OAUTHBEARER authentication.

Type: string

Default: ""

`sasl[].extensions`

Key/value pairs to add to OAUTHBEARER authentication requests.

Type: object

`sasl[].aws`

AWS specific fields for when the mechanism is set to AWS_MSK_IAM.

Type: object

`sasl[].aws.region`

The AWS region to target.

Type: string

Default: ""

`sasl[].aws.endpoint`

Specify a custom endpoint for the AWS API.

Type: string

Default: ""

`sasl[].aws.credentials`

Manually configure the AWS credentials to use (optional). For more information, see the Amazon Web Services guide.

Type: object

`sasl[].aws.credentials.profile`

The profile from ~/.aws/credentials to use.

Type: string

Default: ""

`sasl[].aws.credentials.id`

The ID of the AWS credentials to use.

Type: string

Default: ""

`sasl[].aws.credentials.secret`

The secret for the AWS credentials in use.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`sasl[].aws.credentials.token`

The token for the AWS credentials in use. This is a required value for short-term credentials.

Type: string

Default: ""

`sasl[].aws.credentials.from_ec2_role`

Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance.

Type: bool

Default: false

`sasl[].aws.credentials.role`

The role ARN to assume.

Type: string

Default: ""

`sasl[].aws.credentials.role_external_id`

An external ID to use when assuming a role.

Type: string

Default: ""

`metadata_max_age`

The maximum period of time after which metadata is refreshed.

Type: string

Default: 5m

`request_timeout_overhead`

Grants an additional buffer or overhead to requests that have timeout fields defined. This field is based on the behavior of Apache Kafka’s request.timeout.ms parameter, but with the option to extend the timeout deadline.

Type: string

Default: 10s

`conn_idle_timeout`

Define how long connections can remain idle before they are closed.

Type: string

`pipeline_id`

The ID of a Redpanda Connect data pipeline (optional). When specified, the pipeline ID is written to all logs and status updates sent to the configured topics.

Type: string

Default: ""

`logs_topic`

The topic that logs are sent to.

Type: string

Default: ""

# Example

logs_topic: __redpanda.connect.logs

yml

`logs_level`

The logging level of logs sent to Redpanda.

Type: string

Default: info

Options: debug, info, warn, error

`status_topic`

The topic that status updates are sent to. For full details of the schema for status updates, see the object specification.

Type: string

Default: ""

# Example

status_topic: __redpanda.connect.status

yml

`partitioner`

Override the default murmur2 hashing partitioner.

Type: string

Option Summary

Option	Summary
`least_backup`	Chooses the least backed up partition. The partition with the fewest buffered records. Partitions are selected per batch.
`manual`	Manually select a partition for each message. You must also specify a value for the `partition` field.
`murmur2_hash`	Kafka’s default hash algorithm that uses a 32-bit murmur2 hash of the key to compute the partition for the record.
`round_robin`	Does a round robin of messages through all available partitions. This algorithm has lower throughput and causes higher CPU load on brokers, but is useful if you want to ensure an even distribution of records to partitions.

least_backup

Chooses the least backed up partition. The partition with the fewest buffered records. Partitions are selected per batch.

manual

Manually select a partition for each message. You must also specify a value for the partition field.

murmur2_hash

Kafka’s default hash algorithm that uses a 32-bit murmur2 hash of the key to compute the partition for the record.

round_robin

Does a round robin of messages through all available partitions. This algorithm has lower throughput and causes higher CPU load on brokers, but is useful if you want to ensure an even distribution of records to partitions.

`idempotent_write`

Enable the idempotent write producer option. This requires the IDEMPOTENT_WRITE permission on CLUSTER. Disable this option if the IDEMPOTENT_WRITE permission is not available.

Type: bool

Default: true

`compression`

Set an explicit compression type (optional). The default preference is to use snappy when the broker supports it. Otherwise, use none.

Type: string

Options: lz4 , snappy , gzip , none , zstd

`timeout`

The maximum period of time allowed for sending log or status update messages before a request is abandoned and a retry attempted.

Type: string

Default: 10s

`max_message_bytes`

The maximum size of an individual message in bytes. Messages larger than this value are rejected. This field is equivalent to Kafka’s max.message.bytes.

Type: string

Default: 1MB

# Examples

max_message_bytes: 100MB

max_message_bytes: 50mib

yml

`broker_write_max_bytes`

The upper bound for the number of bytes written to a broker connection in a single write. This field corresponds to Kafka’s socket.request.max.bytes.

Type: string

Default: "100MB"

# Examples

broker_write_max_bytes: 128MB

broker_write_max_bytes: 50mib

yml

Was this helpful?

group Ask in the community

mail Share your feedback

group_add Make a contribution

What do you think of this page?

Let us know more:

Let us contact you about your feedback:

redpanda

Pipeline example

Fields

seed_brokers

client_id

tls

tls.enabled

tls.skip_cert_verify

tls.enable_renegotiation

tls.root_cas

tls.root_cas_file

tls.client_certs

tls.client_certs[].cert

tls.client_certs[].key

tls.client_certs[].cert_file

tls.client_certs[].key_file

tls.client_certs[].password

sasl

sasl[].mechanism

sasl[].username

sasl[].password

sasl[].token

sasl[].extensions

sasl[].aws

sasl[].aws.region

sasl[].aws.endpoint

sasl[].aws.credentials

sasl[].aws.credentials.profile

sasl[].aws.credentials.id

sasl[].aws.credentials.secret

sasl[].aws.credentials.token

sasl[].aws.credentials.from_ec2_role

sasl[].aws.credentials.role

sasl[].aws.credentials.role_external_id

metadata_max_age

request_timeout_overhead

conn_idle_timeout

pipeline_id

logs_topic

logs_level

status_topic

partitioner

idempotent_write

compression

timeout

max_message_bytes

broker_write_max_bytes

Simple online edits

Contribution guide

`seed_brokers`

`client_id`

`tls`

`tls.enabled`

`tls.skip_cert_verify`

`tls.enable_renegotiation`

`tls.root_cas`

`tls.root_cas_file`

`tls.client_certs`

`tls.client_certs[].cert`

`tls.client_certs[].key`

`tls.client_certs[].cert_file`

`tls.client_certs[].key_file`

`tls.client_certs[].password`

`sasl`

`sasl[].mechanism`

`sasl[].username`

`sasl[].password`

`sasl[].token`

`sasl[].extensions`

`sasl[].aws`

`sasl[].aws.region`

`sasl[].aws.endpoint`

`sasl[].aws.credentials`

`sasl[].aws.credentials.profile`

`sasl[].aws.credentials.id`

`sasl[].aws.credentials.secret`

`sasl[].aws.credentials.token`

`sasl[].aws.credentials.from_ec2_role`

`sasl[].aws.credentials.role`

`sasl[].aws.credentials.role_external_id`

`metadata_max_age`

`request_timeout_overhead`

`conn_idle_timeout`

`pipeline_id`

`logs_topic`

`logs_level`

`status_topic`

`partitioner`

`idempotent_write`

`compression`

`timeout`

`max_message_bytes`

`broker_write_max_bytes`