Sink Connector Configuration

The AMQP sink connector writes Kafka messages to an AMQP queue. It formats the messages as described in the architecture documentation.

NOTE: At this time Aiven sink connector for AMQP on Apache Kafka only supports queue type messages. This means that the messages will be distributed across all the tasks. In effect the maximum number of Kafka Connect tasks becomes the number of simultaneous messages that can be processed. Defining more than one task has a high probability of changing the order in which the messages are processed.

Configuration

Apache Kafka has produced a users guide that describes how to run Kafka Connect. This documentation describes the Connect workers configuration. They have also produced a good description of how Kafka Connect resumes after a failure or stoppage.

Configuration options

All the configuration options are listed below in alphabetical order.

amqp.address

Configuration option: amqp.address
Since: 0.1.0
Default value: none
Type: STRING
Valid values: non-empty string without ISO control characters
Importance: MEDIUM

The address (topic) to listend to.

amqp.format

Configuration option: amqp.format
Since: 0.2.0
Default value: none
Type: STRING
Valid values: (case insensitive) [RAW, BODY, NOT_AMQP]
Importance: MEDIUM

The format for the AMQP message IO. RAW - AMQP message is encoded into its native byte buffer format and passed to Kafka as a byte[], the Key is the key specified by the AMQP message or an ULID string if none was provided All properties, annotations, and footers are copied into Kafka message headers with values prefixed by ‘amqp.’ BODY - The body of the AMQP message is extracted and placed into the Kafka value, the Kafka value schema is set appropriately for the value, the Key is the key specified by the AMQP message or an ULID string if none was provided. All properties, annotations, and footers are copied into Kafka message headers with values prefixed by ‘amqp.’. NOT_AMQP - (Only valid for sink) The Kafka message did not an originate as an AMQP message. The AMQP body will be set with the value from the Kafka message, the AMQP key will be set with the key from the Kafka message. All Kafka headers that are prefixed with “amqp.” and are associated with AMQP properties, annotations, or footers will be used to set the appropriate values in the AMQP message.

amqp.host

Configuration option: amqp.host
Since: 0.1.0
Default value: none
Type: STRING
Importance: MEDIUM

The host address for the AMQP service

amqp.password

Configuration option: amqp.password
Since: 0.1.0
Default value: none
Type: PASSWORD
Valid values: non-empty string without ISO control characters
Importance: MEDIUM

The password for the user to log into the AMQP server.

amqp.port

Configuration option: amqp.port
Since: 0.1.0
Default value: 5672
Type: INT
Valid values: [1,…,65534]
Importance: MEDIUM

The port for the AMQP server.

amqp.strategy

Configuration option: amqp.strategy
Since: 0.2.0
Default value: none
Type: STRING
Valid values: (case insensitive) [RAW, BODY]
Importance: MEDIUM

The strategy for building the AMQP message.

amqp.user

Configuration option: amqp.user
Since: 0.1.0
Default value: none
Type: STRING
Valid values: non-empty string without ISO control characters
Importance: MEDIUM

The user to log into the AMQP server.

data.compression.type

Configuration option: data.compression.type
Since: 0.1.0
Default value: NONE
Type: STRING
Valid values: (case insensitive) [ZSTD, GZIP, NONE, SNAPPY]
Importance: MEDIUM

The compression to use to read/write data streams from the data storage. Note: different backends define the data stream differently. Please check the documentation

Error Tolerance

Configuration option: errors.tolerance
Default value: none
Type: STRING
Valid values: [none, all]
Importance: MEDIUM

Behavior for tolerating errors during connector operation. ‘none’ is the default value and signals that any error will result in an immediate connector task failure; ‘all’ changes the behavior to skip over problematic records.

kafka.retry.backoff.ms

Configuration option: kafka.retry.backoff.ms
Since: 0.1.0
Default value: none
Type: LONG
Valid values: [0 Milliseconds,…,1.0 Days]
Importance: MEDIUM

The retry backoff in milliseconds. This config is used to notify Kafka Connect to retry delivering a message batch or performing recovery in case of transient exceptions. Maximum value is 1.0 Days (86400000 Milliseconds)

key.converter.schema.registry.enable

Configuration option: key.converter.schema.registry.enable
Since: 0.1.0
Default value: none
Type: BOOLEAN
Importance: MEDIUM

The key converter schema registry enablement flag. If ‘true’ the key converter schema registry will be enabled, if ‘false’ the registry will not be enabled. If not set the key converter schema registry will be enabled if the schema.registry.enable parameter enabled.

key.converter.schema.registry.url

Configuration option: key.converter.schema.registry.url
Since: 0.1.0
Default value: none
Type: STRING
Valid values: A valid URL. URL scheme must be: ‘https’.
Importance: MEDIUM

Schema registry URL for key converters. If not specified the ‘schema.registry.url’ value will be used.

schema.registry.enable

Configuration option: schema.registry.enable
Since: 0.1.0
Default value: none
Type: BOOLEAN
Importance: MEDIUM

The schema registry enablement flag. If ‘true’ the schema registry will be enabled, if ‘false’ the registry will not be enabled. If not set the registry will be enabled if the schema.registry.url parameter is set.

schema.registry.url

Configuration option: schema.registry.url
Since: 0.1.0
Default value: none
Type: STRING
Valid values: A valid URL. URL scheme must be: ‘https’.
Importance: MEDIUM

The default schema registry URL.

Tasks max

Configuration option: tasks.max
Default value: 1
Type: INT
Valid values: [1,…]
Importance: HIGH

Maximum number of tasks to use for this connector.

value.converter.schema.registry.enable

Configuration option: value.converter.schema.registry.enable
Since: 0.1.0
Default value: none
Type: BOOLEAN
Importance: MEDIUM

The value converter schema registry enablement flag. If ‘true’ the value converter schema registry will be enabled, if ‘false’ the registry will not be enabled. If not set the value converter schema registry will be enabled if the schema.registry.enable parameter enabled.

value.converter.schema.registry.url

Configuration option: value.converter.schema.registry.url
Since: 0.1.0
Default value: none
Type: STRING
Valid values: A valid URL. URL scheme must be: ‘https’.
Importance: MEDIUM

Schema registry URL for value converters. If not specified the ‘schema.registry.url’ value will be used.