AMQP 1.0 Source Connectors

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,…​]

high

amqp.tls.enabled

Enable TLS connection to the AMQP server.

boolean

false

high

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.consumer.source

The AMQP queue or topic address to consume messages from.

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

high

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

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,…​]

medium

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,…​]

medium

amqp.consumer.outgoing.window

Session-level outgoing window size for flow control.

int

2048

[1,…​]

medium

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

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

topic

The Kafka topic to write messages to.

string

non-empty string

high

validate.connection

Attempt a live broker connection during connector validation.

boolean

false

low

oauth.token.endpoint

The URL for the access token endpoint.

string

high

oauth.method

Client authentication method: client_secret_basic, client_secret_post, or private_key_jwt.

string

client_secret_basic

high

oauth.client.id

The client identifier registered with the authorization server.

string

null

high

oauth.client.secret

The client secret (for client_secret_basic and client_secret_post methods).

password

null

high

oauth.scope

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

string

null

high

oauth.private.key.jwk.json

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

password

null

high

oauth.private.key.pem.content

Private key in PEM encoding (alternative to JWK).

password

null

high

oauth.private.key.pem.password

Password for an encrypted PEM private key.

password

null

high

oauth.private.key.jws.algorithm

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

string

RS256

high

oauth.private.key.claims.issuer

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

string

null

high

oauth.private.key.claims.audience

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

string

null

high

oauth.private.key.claims.expirationSeconds

JWT Expiration Time Claim (exp) in seconds.

int

300

medium

oauth.private.key.claims.notBeforeSeconds

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

int

null

medium

oauth.private.key.claims.issuedAt

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

boolean

false

medium

oauth.private.key.claims.jwtId

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

boolean

false

medium