# amqp_0_9

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [cloud-data-platform-full.txt](https://docs.redpanda.com/cloud-data-platform-full.txt)

---
title: amqp_0_9
latest-operator-version: v26.1.4
latest-console-tag: v3.7.3
latest-connect-version: 4.93.0
latest-redpanda-tag: v26.1.9
docname: connect/components/inputs/amqp_0_9
page-component-name: cloud-data-platform
page-version: master
page-component-version: master
page-component-title: Cloud
page-relative-src-path: connect/components/inputs/amqp_0_9.adoc
page-edit-url: https://github.com/redpanda-data/cloud-docs/edit/main/modules/develop/pages/connect/components/inputs/amqp_0_9.adoc
page-git-created-date: "2024-09-09"
page-git-modified-date: "2026-05-26"
---

<!-- Source: https://docs.redpanda.com/cloud-data-platform/develop/connect/components/inputs/amqp_0_9.md -->

**Type:** Input ▼

[Input](https://docs.redpanda.com/cloud-data-platform/develop/connect/components/inputs/amqp_0_9/)[Output](https://docs.redpanda.com/cloud-data-platform/develop/connect/components/outputs/amqp_0_9/)

**Available in:** Cloud, [Self-Managed](https://docs.redpanda.com/connect/components/inputs/amqp_0_9/%20%22View%20the%20Self-Managed%20version%20of%20this%20component%22)

Connects to an AMQP (0.91) queue. AMQP is a messaging protocol used by various message brokers, including RabbitMQ.

#### Common

```yml
inputs:
  label: ""
  amqp_0_9:
    urls: [] # No default (required)
    queue: "" # No default (required)
    consumer_tag: ""
    prefetch_count: 10
```

#### Advanced

```yml
inputs:
  label: ""
  amqp_0_9:
    urls: [] # No default (required)
    queue: "" # No default (required)
    queue_declare:
      enabled: false
      durable: true
      auto_delete: false
      arguments: "" # No default (optional)
    bindings_declare: [] # No default (optional)
    consumer_tag: ""
    auto_ack: false
    nack_reject_patterns: []
    prefetch_count: 10
    prefetch_size: 0
    tls:
      enabled: false
      skip_cert_verify: false
      enable_renegotiation: false
      root_cas: ""
      root_cas_file: ""
      client_certs: []
```

TLS is automatically enabled when connecting to an `amqps` URL. However, you can customize [TLS settings](#tls) if required.

## [](#metadata)Metadata

This input adds the following metadata fields to each message:

-   `amqp_content_type`

-   `amqp_content_encoding`

-   `amqp_delivery_mode`

-   `amqp_priority`

-   `amqp_correlation_id`

-   `amqp_reply_to`

-   `amqp_expiration`

-   `amqp_message_id`

-   `amqp_timestamp`

-   `amqp_type`

-   `amqp_user_id`

-   `amqp_app_id`

-   `amqp_consumer_tag`

-   `amqp_delivery_tag`

-   `amqp_redelivered`

-   `amqp_exchange`

-   `amqp_routing_key`

-   All existing message headers, including nested headers prefixed with the key of their respective parent.


You can access these metadata fields using [function interpolations](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/interpolation/#bloblang-queries).

## [](#fields)Fields

### [](#auto_ack)`auto_ack`

Set to `true` to automatically acknowledge messages as soon as they are consumed rather than waiting for acknowledgments from downstream. This can improve throughput and prevent the pipeline from becoming blocked, but delivery guarantees are lost.

**Type**: `bool`

**Default**: `false`

### [](#bindings_declare)`bindings_declare[]`

Passively declares the bindings of the target queue to make sure they exist and are configured correctly. If the bindings exist, then the passive declaration verifies that fields specified in this object match them.

**Type**: `object`

```yaml
# Examples:
bindings_declare:
  - exchange: foo
    key: bar
```

### [](#bindings_declare-exchange)`bindings_declare[].exchange`

The exchange of the declared binding.

**Type**: `string`

**Default**: `""`

### [](#bindings_declare-key)`bindings_declare[].key`

The key of the declared binding.

**Type**: `string`

**Default**: `""`

### [](#consumer_tag)`consumer_tag`

A consumer tag to uniquely identify the consumer.

**Type**: `string`

**Default**: `""`

### [](#nack_reject_patterns)`nack_reject_patterns[]`

A list of regular expression patterns to match against errors in messages that Redpanda Connect fails to deliver. When a message has an error that matches a pattern, it is dropped or delivered to a dead-letter queue (if a queue has been configured).

By default, failed messages are negatively acknowledged (nacked) and requeued.

**Type**: `array`

**Default**: `[]`

```yaml
# Examples:
nack_reject_patterns:
  - "^reject me please:.+$"
```

### [](#prefetch_count)`prefetch_count`

The maximum number of pending messages at a given time.

**Type**: `int`

**Default**: `10`

### [](#prefetch_size)`prefetch_size`

The maximum size of pending messages (in bytes) at a given time.

**Type**: `int`

**Default**: `0`

### [](#queue)`queue`

An AMQP queue to consume from.

**Type**: `string`

### [](#queue_declare)`queue_declare`

Passively declares the [target queue](#queue) to make sure a queue with the specified name exists and is configured correctly. If the queue exists, then the passive declaration verifies that fields specified in this object match the its properties.

**Type**: `object`

### [](#queue_declare-arguments)`queue_declare.arguments`

Arguments for server-specific implementations of the queue (optional). You can use arguments to configure additional parameters for queue types that require them. For more information about available arguments, see the [RabbitMQ Client Library](https://github.com/rabbitmq/amqp091-go/blob/b3d409fe92c34bea04d8123a136384c85e8dc431/types.go#L282-L362).

| Argument | Description | Accepted values |
| --- | --- | --- |
| x-queue-type | Declares the type of queue. | Options: classic (default), quorum, stream, drop-head, reject-publish, and reject-publish-dlx. |
| x-max-length | The maximum number of messages in the queue. | A non-negative integer. |
| x-max-length-bytes | The maximum size of messages (in bytes) in the queue. | A non-negative integer. |
| x-overflow | Sets the queue’s overflow behavior. | Options: drop-head (default), reject-publish, reject-publish-dlx. |
| x-message-ttl | The duration (in milliseconds) that messages remain in the queue before they expire and are discarded. | A string that represents the number of milliseconds. For example, 60000 retains messages for one minute. |
| x-expires | The duration after which the queue automatically expires. | A positive integer. |
| x-max-age | The duration (in configurable units) that streamed messages are retained on disk before they are discarded. | Options: Y, M, D, h, m, s. For example, 7D retains messages for a week. |
| x-stream-max-segment-size-bytes | The maximum size (in bytes) of the segment files held on disk. | A positive integer. Default: 500000000 (approximately 500 MB). |
| x-queue-version | The version of the classic queue to use. | Options: 1 or 2. |
| x-consumer-timeout | The duration (in milliseconds) that a consumer can remain idle before it is automatically canceled. | A positive integer that represents the number of milliseconds. For example, 60000 sets a timeout duration of one minute. |
| x-single-active-consumer | When set to true, a single consumer receives messages from the queue even when multiple consumers are subscribed to it. | A boolean. |

**Type**: `object`

```yaml
# Examples:
arguments:
  x-max-length: 1000
  x-max-length-bytes: 4096
  x-queue-type: quorum
```

### [](#queue_declare-auto_delete)`queue_declare.auto_delete`

Whether the declared queue auto-deletes when there are no active consumers.

**Type**: `bool`

**Default**: `false`

### [](#queue_declare-durable)`queue_declare.durable`

Whether the declared queue is durable.

**Type**: `bool`

**Default**: `true`

### [](#queue_declare-enabled)`queue_declare.enabled`

Whether to enable queue declaration.

**Type**: `bool`

**Default**: `false`

### [](#tls)`tls`

Configure Transport Layer Security (TLS) settings to secure network connections. This includes options for standard TLS as well as mutual TLS (mTLS) authentication where both client and server authenticate each other using certificates. Key configuration options include `enabled` to enable TLS, `client_certs` for mTLS authentication, `root_cas`/`root_cas_file` for custom certificate authorities, and `skip_cert_verify` for development environments.

**Type**: `object`

### [](#tls-client_certs)`tls.client_certs[]`

A list of client certificates for mutual TLS (mTLS) authentication. Configure this field to enable mTLS, authenticating the client to the server with these certificates.

You must set `tls.enabled: true` for the client certificates to take effect.

**Certificate pairing rules**: For each certificate item, provide either:

-   Inline PEM data using both `cert` **and** `key` or

-   File paths using both `cert_file` **and** `key_file`.


Mixing inline and file-based values within the same item is not supported.

**Type**: `object`

**Default**: `[]`

```yaml
# Examples:
client_certs:
  - cert: foo
    key: bar

# ---

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

### [](#tls-client_certs-cert)`tls.client_certs[].cert`

A plain text certificate to use.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-cert_file)`tls.client_certs[].cert_file`

The path of a certificate to use.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-key)`tls.client_certs[].key`

A plain text certificate key to use.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-key_file)`tls.client_certs[].key_file`

The path of a certificate key to use.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-password)`tls.client_certs[].password`

A 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.

Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
password: foo

# ---

password: ${KEY_PASSWORD}
```

### [](#tls-enable_renegotiation)`tls.enable_renegotiation`

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

**Type**: `bool`

**Default**: `false`

### [](#tls-enabled)`tls.enabled`

Whether custom TLS settings are enabled.

**Type**: `bool`

**Default**: `false`

### [](#tls-root_cas)`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. Use either this field for inline certificate data or `root_cas_file` for file-based certificate loading.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
root_cas: |-
  -----BEGIN CERTIFICATE-----
  ...
  -----END CERTIFICATE-----
```

### [](#tls-root_cas_file)`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. Use either this field for file-based certificate loading or `root_cas` for inline certificate data.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
root_cas_file: ./root_cas.pem
```

### [](#tls-skip_cert_verify)`tls.skip_cert_verify`

Whether to skip server-side certificate verification. Set to `true` only for testing environments as this reduces security by disabling certificate validation. When using self-signed certificates or in development, this may be necessary, but should never be used in production. Consider using `root_cas` or `root_cas_file` to specify trusted certificates instead of disabling verification entirely.

**Type**: `bool`

**Default**: `false`

### [](#urls)`urls[]`

A list of URLs to connect to. This input attempts to connect to each URL in the list, in order, until a successful connection is established. It then continues to use that URL until the connection is closed.

If an item in the list contains commas, it is split into multiple URLs.

**Type**: `array`

```yaml
# Examples:
urls:
  - "amqp://guest:guest@127.0.0.1:5672/"

# ---

urls:
  - "amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"

# ---

urls:
  - "amqp://127.0.0.1:5672/"
  - "amqp://127.0.0.2:5672/"
```