AMQP 1.0 Source Connectors

No authentication (anonymous)

Core connector (io.axual.connect.plugins.amqp.source.AmqpSourceConnector).
Leave amqp.auth.username and amqp.auth.password unset to log in anonymously, or set amqp.auth.enabled=false to skip the SASL exchange altogether.

Username and password (SASL PLAIN)

Core connector. Set amqp.auth.username and amqp.auth.password.

OAuth2 (client credentials or private key JWT)

OAuth2 connector (io.axual.connect.plugins.amqp.oauth2.source.OAuth2AmqpSourceConnector). Configure amqp.oauth. properties.
The connector sets amqp.auth. itself from the acquired token, so configuring them here has no effect.

client_secret_basic

Client ID and secret sent via HTTP Basic.

client_secret_post

Client ID and secret sent in the token request body.

private_key_jwt

Signed client assertion using a private key.
Provide the key as JWK JSON (amqp.oauth.private.key.jwk.json) or PEM (amqp.oauth.private.key.pem.content).

amqp.host

The hostname where the AMQP server can be reached.

string

non-empty string

high

amqp.port

The port number where the AMQP server listens.

int

5672

[1,…​]

medium

amqp.tls.enabled

Enable TLS connection to the AMQP server.

boolean

false

medium

amqp.auth.enabled

When set to false, no authentication is attempted. Default is true.

boolean

true

high

amqp.auth.username

Username for connecting to the AMQP server. Uses anonymous authentication if not set.

string

null

high

amqp.auth.password

Password for connecting to the AMQP server.

password

null

high

amqp.source

The AMQP queue or topic address to consume messages from. This corresponds to the queue name in the broker.

string

non-empty string

high

amqp.consumer.qos

Quality of Service level: AT_MOST_ONCE (fire and forget), AT_LEAST_ONCE (guaranteed, may duplicate), EXACTLY_ONCE (requires broker support).

string

AT_LEAST_ONCE

AT_MOST_ONCE, AT_LEAST_ONCE, EXACTLY_ONCE

medium

amqp.consumer.durable

Use a durable consumer. The AMQP link remains in place after disconnection so messages are not lost. Essential for persistent topic subscriptions.

boolean

false

medium

amqp.consumer.link.name

Explicit link name. Overrides auto-generation. When multiple tasks share the same link name, messages are load-balanced between them.

string

null

low

amqp.consumer.link.name.prefix

Prefix for auto-generated link names. The actual link name will be {prefix}-{task-id}. When unset, the connector name is used as the prefix, so the name becomes {connector-name}-{task-id}.

string

null

medium

amqp.consumer.link.credit

Number of messages prefetched and buffered locally before requiring acknowledgments.

int

100

[1,…​]

medium

amqp.consumer.receive.timeout

Maximum time in milliseconds to wait for a message when polling.

long

1000

[100,…​]

low

amqp.consumer.close.timeout

Maximum time in milliseconds to wait for graceful consumer shutdown.

long

5000

[1000,…​]

low

amqp.consumer.incoming.window

Session-level incoming window size for flow control.

int

2048

[1,…​]

low

amqp.consumer.outgoing.window

Session-level outgoing window size for flow control.

int

2048

[1,…​]

low

amqp.consumer.auto.acknowledge

Automatically acknowledge messages after successful processing.

boolean

false

medium

amqp.consumer.reject.on.error

On processing error: if true, reject the message (no redelivery); if false, release it (can be redelivered).

boolean

false

medium

amqp.idle.timeout

Maximum idle time in milliseconds before the connection is considered dead.

long

2147483647

[1,…​]

medium

amqp.max.frame.size

Maximum size in bytes for individual AMQP frames.

long

2147483647

[1,…​]

medium

amqp.input.buffer.size

Network buffer size in bytes for receiving frames.

int

131072

[1,…​]

medium

amqp.input.buffer.extend.size

Extension size in bytes when the input buffer is insufficient to receive a frame.

int

65536

[1,…​]

medium

amqp.output.buffer.size

Network buffer size in bytes for sending frames.

int

131072

[1,…​]

medium

amqp.output.buffer.extend.size

Extension size in bytes when the output buffer is insufficient to send a frame.

int

65536

[1,…​]

medium

amqp.container.id.prefix

Prefix for auto-generated AMQP container IDs. The actual ID will be {prefix}-{task-id}. When unset, the connector name is used as the prefix, so the ID becomes {connector-name}-{task-id}.

string

null

medium

amqp.container.id

Explicit container ID. Overrides auto-generation.
WARNING: Using the same ID across multiple tasks may cause broker tracking issues.

string

null

low

amqp.key.extractor

Ordered list of key extraction strategies. Each is tried in turn until a key is found.
Options (case insensitive): MESSAGE_ID, CORRELATION_ID, SUBJECT, UUID, NULL.

list

MESSAGE_ID,SUBJECT,NULL

high

amqp.data.mode

How an encoded message body becomes the Kafka record value. See Message body handling.
BASE64 Base64 encodes the Data section bytes and leaves Value bodies rendered as in release 1.0.0.
JSON decodes the body bytes as UTF-8 and publishes the JSON text unchanged. It applies to a Data section and to a Value section holding binary; text in a Value section is published unchanged, because text needs no decoding.
BASE64_JSON reads the body as Base64 text, decodes that first, and then behaves as JSON. It also applies to a Value section holding text.
In both decoding modes, a body that does not hold exactly one valid JSON document is a conversion failure, handled according to amqp.error.handling.

string

BASE64

BASE64, JSON, BASE64_JSON

high

amqp.header.value.format

How the msg. and app. record headers are stored. See Header values.
BYTES stores them as UTF-8 bytes, which the default header converter renders Base64 encoded.
STRING stores them as strings, which the default header converter renders as readable text.

string

BYTES

BYTES, STRING

medium

amqp.error.handling

What happens to a message that could not be converted. See Conversion error handling.
When left empty, the deprecated amqp.unsupported.type.ignore decides: true means DROP, false means FAIL.

string

DROP, BROKER_REJECT, FAIL

medium

amqp.unsupported.type.ignore

Deprecated, superseded by amqp.error.handling, which is used instead whenever it is set. When true, a message that cannot be converted is not written to Kafka; when false, the task fails.

boolean

true

medium

amqp.unsupported.type.log

Log a message that could not be converted.

boolean

true

medium

amqp.unsupported.type.log.name

Logger name to use when logging a message that could not be converted. Empty means the connector’s own logger.

string

low

amqp.unsupported.type.log.include.body

Include the full AMQP message in that log entry.

boolean

false

low

amqp.unsupported.type.log.include.message.id

Include the message id in that log entry. Ignored when the full message is already included.

boolean

false

low

topic

The Kafka topic to write messages to.

string

non-empty string

high

kafka.commit.timeout.ms

How long to wait for a converted message to be confirmed written to Kafka. While messages are waiting, the AMQP consumer is paused.

long

30000

[1,…​]

medium

amqp.validate.connection

Attempt a live broker connection during connector validation.

boolean

false

low

amqp.oauth.username

The AMQP username for SASL authentication. The bearer token is sent as the password, so this is the only credential you set yourself.

string

high

amqp.oauth.token.endpoint

The URL for the access token endpoint.

string

high

amqp.oauth.method

Client authentication method: client_secret_basic, client_secret_post, or private_key_jwt.

string

client_secret_basic

medium

amqp.oauth.client.id

The client identifier registered with the authorization server.

string

high

amqp.oauth.client.secret

The client secret (for client_secret_basic and client_secret_post methods).

password

null

medium

amqp.oauth.scope

Space-separated list of requested permissions (e.g., read_users write_data).

string

null

medium

amqp.oauth.private.key.jwk.json

Private key in JWK JSON format (for private_key_jwt method).

password

null

medium

amqp.oauth.private.key.pem.content

Private key in PEM encoding (alternative to JWK).

password

null

medium

amqp.oauth.private.key.pem.password

Password for an encrypted PEM private key.

password

null

low

amqp.oauth.private.key.jws.algorithm

JWS algorithm used to sign the client assertion. Default is RS256.

string

RS256

medium

amqp.oauth.private.key.claims.issuer

JWT Issuer Claim (iss). Defaults to amqp.oauth.client.id if not set.

string

null

medium

amqp.oauth.private.key.claims.audience

JWT Audience Claim (aud). Defaults to amqp.oauth.token.endpoint if not set.

string

null

medium

amqp.oauth.private.key.claims.expirationSeconds

JWT Expiration Time Claim (exp) in seconds.

long

300

medium

amqp.oauth.private.key.claims.notBeforeSeconds

JWT Not Before Claim (nbf). Seconds added to the issued-at time. Not set by default.

long

null

medium

amqp.oauth.private.key.claims.issuedAt

When true, adds an Issued At (iat) claim to the JWT.

boolean

false

medium

amqp.oauth.private.key.claims.jwtId

When true, adds a random JWT Identifier (jti) claim.

boolean

false

medium

AMQPBinary

"binary : " + Base64

AMQPDecimal128

"decimal128 : " + Base64

AMQPDecimal32 / AMQPDecimal64

Decimal notation, as BigDecimal.toString() renders it

AMQPByte / AMQPUnsignedByte

"0x" + 2-digit hex

AMQPTimestamp

Epoch milliseconds (long as string)

AMQPBoolean

"true" or "false" (lowercase)

AMQPChar

Single-character Unicode string

AMQPUuid

Canonical UUID text, as UUID.toString() renders it

AMQPString

The string itself

Numeric types (AMQPInt, AMQPLong, AMQPShort, AMQPFloat, AMQPDouble and their unsigned variants)

String.valueOf(value)

AMQPArray / AMQPList / AMQPMap

Not supported - results in a conversion error

Any other type

The string form the AMQP type reports for itself

Data section

Base64 text of the bytes

Bytes decoded as UTF-8 and published unchanged

Read as Base64 first, then decoded as UTF-8 and published unchanged

Value section holding binary

binary : <base64>

Bytes decoded as UTF-8 and published unchanged

Read as Base64 first, then decoded as UTF-8 and published unchanged

Value section holding text

Published unchanged

Published unchanged, text needs no decoding

Read as Base64, decoded and published unchanged

Value section, any other type

Rendered per its AMQP type in every mode, see the table above

BYTES (default)

Stored with Schema.OPTIONAL_BYTES_SCHEMA. The release 1.0.0 behaviour. Under SimpleHeaderConverter a consumer sees YXBwbGljYXRpb24vanNvbg==; under ByteArrayConverter, used by the examples below, the raw bytes application/json reach the consumer.

STRING

Stored with Schema.OPTIONAL_STRING_SCHEMA. Under SimpleHeaderConverter a consumer sees application/json.

DROP

Log a warning and accept the message. It leaves the broker queue and is not written to Kafka, so its content is lost. This is what an unset amqp.error.handling resolves to, unless amqp.unsupported.type.ignore was set to false.

BROKER_REJECT

Log a warning and reject the message, handing it back to the broker. A broker with a dead-letter address configured routes it there with the original bytes intact. Without one, the message is discarded exactly as with DROP.

FAIL

Log a warning, reject the message and fail the task. Nothing is lost, but the connector stops until the cause is resolved.

field.path

Dot separated path to the Base64 encoded field, for example data or body.payload.data. Required, there is no default. Every segment must be non-empty. Field names that contain a dot cannot be addressed. When a document repeats a key, the first match is decoded.

string

high