Apache Kafka Connector 4.9 Reference - Mule 4

Array of String

List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

String

Default group ID for all the Kafka consumers that use this configuration.

Number

Determines the number of consumers the connection initially creates.

1

Number

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before the expiration of this timeout, then the consumer is considered failed and the group rebalances to reassign the partitions to another member. For consumers using a non-null group.instance.id which reaches this timeout, partitions are not immediately reassigned. Instead, the consumer stops sending heartbeats and partitions are reassigned after the expiration of session.timeout.ms.

60

Enumeration, one of:

Time unit for the Maximum polling interval field. You can override this parameter at the source level.

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in the READ_COMMITTED mode, consumer.poll() returns only messages up to the last stable offset (LSO), which is one less than the offset of the first open transaction.

In particular, any messages appearing after messages belonging to ongoing transactions are withheld until the relevant transaction is completed. As a result, READ_COMMITTED consumers are not able to read up to the high watermark when there are in-flight transactions. Furthermore, when in READ_COMMITTED, the seekToEnd method returns the LSO.

READ_UNCOMMITTED

Boolean

Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.

true

Enumeration, one of:

Determines what to do when there is no initial offset in Kafka or if the current offset no longer exists on the server, for example, because the data was deleted.

LATEST

Number

Amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.

100

Enumeration, one of:

Time unit for the Retry Backoff Timeout field.

MILLISECONDS

Boolean

Automatically check the CRC32 of the records consumed. This ensures that no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so in situations that require extremely high performance, this can be disabled.

true

Number

Size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this parameter at the source level.

64

Enumeration, one of:

Unit of measure for the Default receive buffer size field. You can override this parameter at the source level.

KB

Number

Size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default is used. You can override this parameter at the source level.

128

Enumeration, one of:

Unit of measure for the Default send buffer size field. You can override this parameter at the source level.

KB

Number

Maximum amount of time the client waits for the response of a request. If the response is not received before the timeout elapses, the client resends the request if necessary, or fails the request if the retries are exhausted. You can override this parameter at the source level.

30

Enumeration, one of:

Time unit for the Request Timeout field. You can override this parameter at the source level.

SECONDS

Number

Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.

500

Enumeration, one of:

Controls how the client uses DNS lookups. If set to USE_ALL_DNS_IPS then, when the lookup returns multiple IP addresses for a hostname, they attempt to connect before failing the connection. Applies to both bootstrap and advertised servers. If the value is RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY each entry is resolved and expanded into a list of canonical names.

USE_ALL_DNS_IPS

Number

Expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

3

Enumeration, one of:

Time unit for the Heartbeat interval field.

SECONDS

Number

Timeout used to detect consumer failures when using Kafka’s group management facility. The consumer sends periodic heartbeats to indicate its aliveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker removes this consumer from the group and initiates a rebalance. The value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

10

Enumeration, one of:

Time unit for the Session Timeout field.

SECONDS

Number

Close idle connections after the number of milliseconds specified by this configuration.

540

Enumeration, one of:

Time unit for the Connection maximum idle time field.

SECONDS

Object

Additional properties used to customize the Kafka connection.

For example:

TLS

Defines a TLS configuration used by both clients and server sides to secure the communication for the Mule app. The connector automatically sets the security.protocol to use for the communication. The valid values are PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. The default value is PLAINTEXT or SASL_PLAINTEXT for SASL authentication - kerberos/scram/plain. If the broker configures SSL as the protocol, then configure at least the keystore in the tls:context child element of the configuration and the connector automatically uses SSL (or SASL_SSL for SASL authentication) as the security.protocol.

String

Endpoint identification algorithm used by clients to validate the server hostname. The default value is an empty string, which means the endpoint identification algorithm is disabled. Clients, including client connections created by the broker for inter-broker communication, verify that the broker host name matches the host name in the brokers certificate.

Array of String

List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.

Array of Topic Partition

List of topic-partition pairs to assign. Consumers are not automatically rebalanced.

Number

Minimum amount of data the server returns for a fetch request. If insufficient data is available, the request waits for the specified minimum amount of data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. If you set this to a value greater than 1, the server waits for larger amounts of data to accumulate, which can improve server throughput slightly at the cost of some additional latency. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch minimum size field. You can override this parameter at the source level.

BYTE

Number

Maximum amount of data the server returns for a fetch request. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch is still returned to ensure that the consumer can make progress. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). The consumer performs multiple fetches in parallel. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch maximum size field. You can override this parameter at the source level.

MB

Number

Maximum amount of data per partition that the server returns. The consumer fetches records in batches. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch is still returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). See fetch.max.bytes for limiting the consumer request size. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default maximum partition fetch size field. You can override this parameter at the source level.

MB

Number

Maximum amount of time the server blocks before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement specified by fetch.min.bytes.

500

Enumeration, one of:

Time unit for the Fetch Maximum Wait Timeout field.

MILLISECONDS

Reconnection

Configures a reconnection strategy to use when a connector operation fails to connect to an external server.

Array of String

List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

String

Default group ID for all the Kafka consumers that use this configuration.

Number

Determines the number of consumers the connection initially creates.

1

Number

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before the expiration of this timeout, then the consumer is considered failed and the group rebalances to reassign the partitions to another member. For consumers using a non-null group.instance.id which reaches this timeout, partitions are not immediately reassigned. Instead, the consumer stops sending heartbeats and partitions are reassigned after the expiration of session.timeout.ms.

60

Enumeration, one of:

Time unit for the Maximum polling interval field. You can override this parameter at the source level.

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in the READ_COMMITTED mode, consumer.poll() returns only messages up to the last stable offset (LSO), which is one less than the offset of the first open transaction.

In particular, any messages appearing after messages belonging to ongoing transactions are withheld until the relevant transaction is completed. As a result, READ_COMMITTED consumers are not able to read up to the high watermark when there are in-flight transactions. Furthermore, when in READ_COMMITTED, the seekToEnd method returns the LSO.

READ_UNCOMMITTED

Boolean

Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.

true

Enumeration, one of:

Determines what to do when there is no initial offset in Kafka or if the current offset no longer exists on the server, for example, because the data was deleted.

LATEST

Number

Amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.

100

Enumeration, one of:

Time unit for the Retry Backoff Timeout field.

MILLISECONDS

Boolean

Automatically check the CRC32 of the records consumed. This ensures that no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so in situations that require extremely high performance, this can be disabled.

true

Number

Size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this parameter at the source level.

64

Enumeration, one of:

Unit of measure for the Default receive buffer size field. You can override this parameter at the source level.

KB

Number

Size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default is used. You can override this parameter at the source level.

128

Enumeration, one of:

Unit of measure for the Default send buffer size field. You can override this parameter at the source level.

KB

Number

Maximum amount of time the client waits for the response of a request. If the response is not received before the timeout elapses, the client resends the request if necessary, or fails the request if the retries are exhausted. You can override this parameter at the source level.

30

Enumeration, one of:

Time unit for the Request Timeout field. You can override this parameter at the source level.

SECONDS

Number

Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.

500

Enumeration, one of:

Controls how the client uses DNS lookups. If set to USE_ALL_DNS_IPS then, when the lookup returns multiple IP addresses for a hostname, they attempt to connect before failing the connection. Applies to both bootstrap and advertised servers. If the value is RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY each entry is resolved and expanded into a list of canonical names.

USE_ALL_DNS_IPS

Number

Expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

3

Enumeration, one of:

Time unit for the Heartbeat interval field.

SECONDS

Number

Timeout used to detect consumer failures when using Kafka’s group management facility. The consumer sends periodic heartbeats to indicate its aliveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker removes this consumer from the group and initiates a rebalance. The value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

10

Enumeration, one of:

Time unit for the Session Timeout field.

SECONDS

Number

Close idle connections after the number of milliseconds specified by this configuration.

540

Enumeration, one of:

Time unit for the Connection maximum idle time field.

SECONDS

Object

Additional properties used to customize the Kafka connection.

For example:

TLS

Defines a TLS configuration used by both clients and server sides to secure the communication for the Mule app. The connector automatically sets the security.protocol to use for the communication. The valid values are PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. The default value is PLAINTEXT or SASL_PLAINTEXT for SASL authentication - kerberos/scram/plain. If the broker configures SSL as the protocol, then configure at least the keystore in the tls:context child element of the configuration and the connector automatically uses SSL (or SASL_SSL for SASL authentication) as the security.protocol.

String

Endpoint identification algorithm used by clients to validate the server hostname. The default value is an empty string, which means the endpoint identification algorithm is disabled. Clients, including client connections created by the broker for inter-broker communication, verify that the broker host name matches the host name in the brokers certificate.

Array of String

List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.

Array of Topic Partition

List of topic-partition pairs to assign. Consumers are not automatically rebalanced.

Number

Minimum amount of data the server returns for a fetch request. If insufficient data is available, the request waits for the specified minimum amount of data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. If you set this to a value greater than 1, the server waits for larger amounts of data to accumulate, which can improve server throughput slightly at the cost of some additional latency. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch minimum size field. You can override this parameter at the source level.

BYTE

Number

Maximum amount of data the server returns for a fetch request. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch is still returned to ensure that the consumer can make progress. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). The consumer performs multiple fetches in parallel. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch maximum size field. You can override this parameter at the source level.

MB

Number

Maximum amount of data per partition that the server returns. The consumer fetches records in batches. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch is still returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). See fetch.max.bytes for limiting the consumer request size. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default maximum partition fetch size field. You can override this parameter at the source level.

MB

Number

Maximum amount of time the server blocks before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement specified by fetch.min.bytes.

500

Enumeration, one of:

Time unit for the Fetch Maximum Wait Timeout field.

MILLISECONDS

String

Entity that is authenticated by a computer system or a network. Principals can be individual people, computers, services, or computational entities such as processes and threads.

x

String

Kerberos principal name that Kafka runs as.

x

String

Path to the krb5.conf file, which contains Kerberos configuration information. This information includes the locations of KDCs and admin servers for the Kerberos realms of interest, defaults for the current realm, defaults for Kerberos applications, and the mappings of hostnames to Kerberos realms.

Boolean

Set this option to true to obtain the ticket-granting ticket (TGT) from the ticket cache. Set this option to false if you do not want to use the ticket cache. The connector searches for the ticket cache as follows:

false

String

Name of the ticket cache that contains the user’s ticket-granting ticket (TGT). If this value is set, Use ticket cache must also be set to true. Otherwise, a configuration error is returned.

Boolean

Set this option to true if you want the connector to obtain the principal’s key from the keytab. If you don’t set this value, the connector locates the keytab by using the Kerberos configuration file. If the keytab is not specified in the Kerberos configuration file, the connector looks for the {user.home}{file.separator}krb5.keytab file.

false

String

Set this option to the file name of the keytab to obtain the principal’s secret key.

Boolean

Set this option to true to store the principal’s subject private credentials.

false

Reconnection

Configures a reconnection strategy to use when a connector operation fails to connect to an external server.

Array of String

The list of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

String

The endpoint identification algorithm used by clients to validate server host name. The default value is an empty string, which means it is disabled. Clients including client connections created by the broker for inter-broker communication verify that the broker host name matches the host name in the brokers certificate.

String

Default Group ID for all the Kafka Consumers that use this configuration.

Number

Determines the number of consumers the connection initially creates.

1

Number

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before the expiration of this timeout, then the consumer is considered failed and the group rebalances to reassign the partitions to another member. For consumers using a non-null group.instance.id which reaches this timeout, partitions are not immediately reassigned. Instead, the consumer stops sending heartbeats and partitions are reassigned after the expiration of session.timeout.ms.

60

Enumeration, one of:

Determines the time unit for request timeout scalar. This parameter can be overridden at source level.

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in the READ_COMMITTED mode, consumer.poll() returns only messages up to the last stable offset (LSO), which is one less than the offset of the first open transaction.

In particular, any messages appearing after messages belonging to ongoing transactions are withheld until the relevant transaction is completed. As a result, READ_COMMITTED consumers are not able to read up to the high watermark when there are in-flight transactions. Furthermore, when in READ_COMMITTED, the seekToEnd method returns the LSO.

READ_UNCOMMITTED

Boolean

Whether internal topics matching a subscribed pattern is excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.

true

Enumeration, one of:

Determines what to do when there is no initial offset in Kafka or if the current offset no longer exists on the server, for example, because the data was deleted.

LATEST

Number

The amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.

100

Enumeration, one of:

Determines the time unit for the reconnect backoff timeout scalar.

MILLISECONDS

Boolean

Automatically check the CRC32 of the records consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance.

true

Number

The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this field at the source level.

64

Enumeration, one of:

The unit of measure for the receive buffer size scalar. You can override this field at the source level.

KB

Number

The size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default is used. You can override this field at the source level.

128

Enumeration, one of:

The unit of measure for the send buffer size scalar. This parameter can be overridden at source level.

KB

Number

The configuration controls the maximum amount of time the client waits for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted. This parameter can be overridden at source level.

30

Enumeration, one of:

Determines the time unit for request timeout scalar. This parameter can be overridden at source level.

SECONDS

Number

The maximum number of records returned on a poll call to the Kafka cluster. This parameter can be overridden at source level.

500

Enumeration, one of:

Controls how the client uses DNS lookups. If set to use_all_dns_ips then, when the lookup returns multiple IP addresses for a hostname, they will all be attempted to connect to before failing the connection. Applies to both bootstrap and advertised servers. If the value is resolve_canonical_bootstrap_servers_only each entry will be resolved and expanded into a list of canonical names.

USE_ALL_DNS_IPS

Number

The expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

3

Enumeration, one of:

Determines the time unit for fetch heartbeat interval time scalar.

SECONDS

Number

The timeout used to detect consumer failures when using Kafka’s group management facility. The consumer sends periodic heartbeats to indicate its liveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker will remove this consumer from the group and initiate a rebalance. Note that the value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

10

Enumeration, one of:

Determines the time unit for session timeout scalar.

SECONDS

Number

Close idle connections after the number of milliseconds specified by this config.

540

Enumeration, one of:

Determines the time unit for connections maximum idle time scalar.

SECONDS

Object

Additional properties used to customize the Kafka connection.

For example:

TLS

Defines a TLS configuration, which can be used from both the client and server sides to secure communication for the Mule app. The connector will automatically set the 'security.protocol' to use for communication. Valid values are PLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL. Default value when no configuration has been provided is PLAINTEXT(or SASL_PLAINTEXT for SASL authentication - kerberos/scram/plain/token). If SSL was configured as protocol on the broker side then the user needs to configure at least the keystore in the 'tls:context' child element of this config and the connector will automatically use SSL(or SASL_SSL for SASL authentication) as 'security.protocol'.

Array of String

The list of subscription regular expressions to subscribe to. This topics will be automatically rebalanced between the amount of consumers of the topic.

Array of Topic Partition

The list of topic-partition pairs to assign. Note that there will be no automatic rebalance of the consumers

Number

The minimum amount of data the server should return for a fetch request. If insufficient data is available the request waits for that much data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. Setting this to something greater than 1 will cause the server to wait for larger amounts of data to accumulate which can improve server throughput a bit at the cost of some additional latency. This parameter can be overridden at source level.

1

Enumeration, one of:

BYTE

Number

The maximum amount of data the server should return for a fetch request. Records are fetched in batches by the consumer, and if the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch will still be returned to ensure that the consumer can make progress. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config). Note that the consumer performs multiple fetches in parallel. This parameter can be overridden at source level.

1

Enumeration, one of:

The unit of measure for the maximum partition fetch size scalar. This parameter can be overridden at source level.

MB

Number

The maximum amount of data per-partition the server will return. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch will still be returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (topic config). See fetch.max.bytes for limiting the consumer request size.This parameter can be overridden at source level.

1

Enumeration, one of:

The unit of measure for the maximum partition fetch size scalar. This parameter can be overridden at source level.

MB

Number

The maximum amount of time the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement given by fetch.min.bytes.

500

Enumeration, one of:

Determines the time unit for fetch maximum wait timeout scalar.

MILLISECONDS

String

Specifies the configuration key used to define the OAuth client ID within the JAAS configuration.

clientId

String

The value for the Client ID which refers to the unique identifier assigned to the OAuth client, facilitating secure authentication and authorization processes.

x

String

Specifies the configuration key used to define the OAuth client secret within the JAAS configuration.

clientSecret

String

The value for the client secret which refers to the confidential authentication credential assigned to the OAuth client.

x

String

Specifies the configuration key used to define the OAuth token endpoint within the JAAS configuration.

sasl.oauthbearer.token.endpoint.url

String

Describes the URL where clients can request access tokens from the authentication server.

x

String

Specifies the configuration key used to define the OAuth scope within the JAAS configuration.

scope

String

Defines the specific permissions and access rights granted to the client application by the authorization server.

String

Specifies the configuration key used to define the OAuth audience within the JAAS configuration.

audience

String

Comma separated list of one or more target audiences for the access token.

Number

Sets the maximum duration, in seconds, for which an access token remains valid.

String

Unique identifier associated with an authenticated user, often used for authorization decisions and resource access.

Enumeration, one of:

Defines whether to include client credentials in the basic authentication header or in the body of the authentication request.

BASIC_AUTHENTICATION_HEADER

Boolean

Indicates whether this login module must be successfully authenticated for Kafka clients to establish a connection using OAuth 2.0 bearer tokens.

true

Boolean

Whether to include 'Accept-application/json' header in the authentication request.

true

Object

Represents a mapping of key-value pairs containing extensions for custom broker OAuth implementations. If you specify a key for the OAuth Extensions field, ensure that the input contains only alphabetic characters and is at least one character long. If you specify a value for OAuth Extensions field, ensure the input contains only printable ASCII characters or whitespace (including space, tab, carriage return, and newline), and the value is at least one character long.

Number

Time for which the login refresh thread waits until a certain portion of the credential’s lifespan passes and attempts to refresh it. Acceptable values range from (0.5) 50% to (1.0) 100%, with a default of (0.8) 80% if unspecified.

0.8

Number

The maximum random jitter added to the login refresh thread’s sleep time relative to the credential’s lifespan is between (0) 0% and (0.25) 25%, with a default of (0.05) 5% if unspecified.

0.05

Number

The desired minimum time in seconds for the login refresh thread to wait before refreshing a credential. Legal values are between 0 and 900 (15 minutes).

60

Number

The amount of buffer time in seconds before credential expiration to maintain when refreshing a credential. Legal values are between 0 and 3600 (1 hour). A default value of 300 (5 minutes) is used if no value is specified.

300

Boolean

Indicates whether to use persistent connections:

Mule uses persistent connections.

Mule closes the connection after the first request completes.

true

Number

Maximum number of connections to open to the backend. HTTP requests are sent in parallel over multiple connections. Setting this value too high can impact latency and consume additional resources without increasing throughput. By default, the number of connections is unlimited.

-1

Number

The number of milliseconds that a connection can remain idle before it is closed. The value of this attribute is only used when persistent connections are enabled.

30000

Boolean

Whether received responses should be streamed, meaning processing continues as soon as all headers are parsed and the body streamed as it arrives. When enabled, the response MUST be eventually read since depending on the configured buffer size it may not fit into memory and processing will stop until space is available.

false

Number

Size of the buffer that stores the HTTP response, in bytes. By default, the space is not limited.

-1

Tcp Client Socket Params

Encapsulates the configurable properties for TCP client socket connections.

Proxy Config Params

Configures a proxy for outbound connections.

Boolean

Specifies whether to follow redirects or not.

false

Number

Maximum time that the request element will block the execution of the flow waiting for the HTTP response.

60000

Reconnection

When the application is deployed, a connectivity test is performed on all connectors. If set to true, deployment fails if the test doesn’t pass after exhausting the associated reconnection strategy

Array of String

List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

String

Default group ID for all the Kafka consumers that use this configuration.

Number

Determines the number of consumers the connection initially creates.

1

Number

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before the expiration of this timeout, then the consumer is considered failed and the group rebalances to reassign the partitions to another member. For consumers using a non-null group.instance.id which reaches this timeout, partitions are not immediately reassigned. Instead, the consumer stops sending heartbeats and partitions are reassigned after the expiration of session.timeout.ms.

60

Enumeration, one of:

Time unit for the Maximum polling interval field. You can override this parameter at the source level.

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in the READ_COMMITTED mode, consumer.poll() returns only messages up to the last stable offset (LSO), which is one less than the offset of the first open transaction.

In particular, any messages appearing after messages belonging to ongoing transactions are withheld until the relevant transaction is completed. As a result, READ_COMMITTED consumers are not able to read up to the high watermark when there are in-flight transactions. Furthermore, when in READ_COMMITTED, the seekToEnd method returns the LSO.

READ_UNCOMMITTED

Boolean

Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.

true

Enumeration, one of:

Determines what to do when there is no initial offset in Kafka or if the current offset no longer exists on the server, for example, because the data was deleted.

LATEST

Number

Amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.

100

Enumeration, one of:

Time unit for the Retry Backoff Timeout field.

MILLISECONDS

Boolean

Automatically check the CRC32 of the records consumed. This ensures that no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so in situations that require extremely high performance, this can be disabled.

true

Number

Size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this parameter at the source level.

64

Enumeration, one of:

Unit of measure for the Default receive buffer size field. You can override this parameter at the source level.

KB

Number

Size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default is used. You can override this parameter at the source level.

128

Enumeration, one of:

Unit of measure for the Default send buffer size field. You can override this parameter at the source level.

KB

Number

Maximum amount of time the client waits for the response of a request. If the response is not received before the timeout elapses, the client resends the request if necessary, or fails the request if the retries are exhausted. You can override this parameter at the source level.

30

Enumeration, one of:

Time unit for the Request Timeout field. You can override this parameter at the source level.

SECONDS

Number

Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.

500

Enumeration, one of:

Controls how the client uses DNS lookups. If set to USE_ALL_DNS_IPS then, when the lookup returns multiple IP addresses for a hostname, they attempt to connect before failing the connection. Applies to both bootstrap and advertised servers. If the value is RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY each entry is resolved and expanded into a list of canonical names.

USE_ALL_DNS_IPS

Number

Expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

3

Enumeration, one of:

Time unit for the Heartbeat interval field.

SECONDS

Number

Timeout used to detect consumer failures when using Kafka’s group management facility. The consumer sends periodic heartbeats to indicate its aliveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker removes this consumer from the group and initiates a rebalance. The value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

10

Enumeration, one of:

Time unit for the Session Timeout field.

SECONDS

Number

Close idle connections after the number of milliseconds specified by this configuration.

540

Enumeration, one of:

Time unit for the Connection maximum idle time field.

SECONDS

Object

Additional properties used to customize the Kafka connection.

For example:

TLS

Defines a TLS configuration used by both clients and server sides to secure the communication for the Mule app. The connector automatically sets the security.protocol to use for the communication. The valid values are PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. The default value is PLAINTEXT or SASL_PLAINTEXT for SASL authentication - kerberos/scram/plain. If the broker configures SSL as the protocol, then configure at least the keystore in the tls:context child element of the configuration and the connector automatically uses SSL (or SASL_SSL for SASL authentication) as the security.protocol.

String

Endpoint identification algorithm used by clients to validate the server hostname. The default value is an empty string, which means the endpoint identification algorithm is disabled. Clients, including client connections created by the broker for inter-broker communication, verify that the broker host name matches the host name in the brokers certificate.

Array of String

List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.

Array of Topic Partition

List of topic-partition pairs to assign. Consumers are not automatically rebalanced.

Number

Minimum amount of data the server returns for a fetch request. If insufficient data is available, the request waits for the specified minimum amount of data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. If you set this to a value greater than 1, the server waits for larger amounts of data to accumulate, which can improve server throughput slightly at the cost of some additional latency. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch minimum size field. You can override this parameter at the source level.

BYTE

Number

Maximum amount of data the server returns for a fetch request. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch is still returned to ensure that the consumer can make progress. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). The consumer performs multiple fetches in parallel. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch maximum size field. You can override this parameter at the source level.

MB

Number

Maximum amount of data per partition that the server returns. The consumer fetches records in batches. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch is still returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). See fetch.max.bytes for limiting the consumer request size. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default maximum partition fetch size field. You can override this parameter at the source level.

MB

Number

Maximum amount of time the server blocks before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement specified by fetch.min.bytes.

500

Enumeration, one of:

Time unit for the Fetch Maximum Wait Timeout field.

MILLISECONDS

String

Username with which to login.

x

String

Password with which to login.

x

Enumeration, one of:

Encryption algorithm used by SCRAM.

x

Reconnection

Configures a reconnection strategy to use when a connector operation fails to connect to an external server.

Array of String

List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

String

Default group ID for all the Kafka consumers that use this configuration.

Number

Determines the number of consumers the connection initially creates.

1

Number

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before the expiration of this timeout, then the consumer is considered failed and the group rebalances to reassign the partitions to another member. For consumers using a non-null group.instance.id which reaches this timeout, partitions are not immediately reassigned. Instead, the consumer stops sending heartbeats and partitions are reassigned after the expiration of session.timeout.ms.

60

Enumeration, one of:

Time unit for the Maximum polling interval field. You can override this parameter at the source level.

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in the READ_COMMITTED mode, consumer.poll() returns only messages up to the last stable offset (LSO), which is one less than the offset of the first open transaction.

In particular, any messages appearing after messages belonging to ongoing transactions are withheld until the relevant transaction is completed. As a result, READ_COMMITTED consumers are not able to read up to the high watermark when there are in-flight transactions. Furthermore, when in READ_COMMITTED, the seekToEnd method returns the LSO.

READ_UNCOMMITTED

Boolean

Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.

true

Enumeration, one of:

Determines what to do when there is no initial offset in Kafka or if the current offset no longer exists on the server, for example, because the data was deleted.

LATEST

Number

Amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.

100

Enumeration, one of:

Time unit for the Retry Backoff Timeout field.

MILLISECONDS

Boolean

Automatically check the CRC32 of the records consumed. This ensures that no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so in situations that require extremely high performance, this can be disabled.

true

Number

Size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this parameter at the source level.

64

Enumeration, one of:

Unit of measure for the Default receive buffer size field. You can override this parameter at the source level.

KB

Number

Size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default is used. You can override this parameter at the source level.

128

Enumeration, one of:

Unit of measure for the Default send buffer size field. You can override this parameter at the source level.

KB

Number

Maximum amount of time the client waits for the response of a request. If the response is not received before the timeout elapses, the client resends the request if necessary, or fails the request if the retries are exhausted. You can override this parameter at the source level.

30

Enumeration, one of:

Time unit for the Request Timeout field. You can override this parameter at the source level.

SECONDS

Number

Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.

500

Enumeration, one of:

Controls how the client uses DNS lookups. If set to USE_ALL_DNS_IPS then, when the lookup returns multiple IP addresses for a hostname, they attempt to connect before failing the connection. Applies to both bootstrap and advertised servers. If the value is RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY each entry is resolved and expanded into a list of canonical names.

USE_ALL_DNS_IPS

Number

Expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

3

Enumeration, one of:

Time unit for the Heartbeat interval field.

SECONDS

Number

Timeout used to detect consumer failures when using Kafka’s group management facility. The consumer sends periodic heartbeats to indicate its aliveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker removes this consumer from the group and initiates a rebalance. The value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

10

Enumeration, one of:

Time unit for the Session Timeout field.

SECONDS

Number

Close idle connections after the number of milliseconds specified by this configuration.

540

Enumeration, one of:

Time unit for the Connection maximum idle time field.

SECONDS

Object

Additional properties used to customize the Kafka connection.

For example:

TLS

Defines a TLS configuration used by both clients and server sides to secure the communication for the Mule app. The connector automatically sets the security.protocol to use for the communication. The valid values are PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. The default value is PLAINTEXT or SASL_PLAINTEXT for SASL authentication - kerberos/scram/plain. If the broker configures SSL as the protocol, then configure at least the keystore in the tls:context child element of the configuration and the connector automatically uses SSL (or SASL_SSL for SASL authentication) as the security.protocol.

String

Endpoint identification algorithm used by clients to validate the server hostname. The default value is an empty string, which means the endpoint identification algorithm is disabled. Clients, including client connections created by the broker for inter-broker communication, verify that the broker host name matches the host name in the brokers certificate.

Array of String

List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.

Array of Topic Partition

List of topic-partition pairs to assign. Consumers are not automatically rebalanced.

Number

Minimum amount of data the server returns for a fetch request. If insufficient data is available, the request waits for the specified minimum amount of data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. If you set this to a value greater than 1, the server waits for larger amounts of data to accumulate, which can improve server throughput slightly at the cost of some additional latency. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch minimum size field. You can override this parameter at the source level.

BYTE

Number

Maximum amount of data the server returns for a fetch request. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch is still returned to ensure that the consumer can make progress. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). The consumer performs multiple fetches in parallel. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch maximum size field. You can override this parameter at the source level.

MB

Number

Maximum amount of data per partition that the server returns. The consumer fetches records in batches. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch is still returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). See fetch.max.bytes for limiting the consumer request size. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default maximum partition fetch size field. You can override this parameter at the source level.

MB

Number

Maximum amount of time the server blocks before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement specified by fetch.min.bytes.

500

Enumeration, one of:

Time unit for the Fetch Maximum Wait Timeout field.

MILLISECONDS

String

User used by the client to connect to the Kafka broker.

x

String

Password used by the client to connect to the Kafka broker.

x

Reconnection

Configures a reconnection strategy to use when a connector operation fails to connect to an external server.

Array of String

List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

String

Default group ID for all the Kafka consumers that use this configuration.

Number

Determines the number of consumers the connection initially creates.

1

Number

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before the expiration of this timeout, then the consumer is considered failed and the group rebalances to reassign the partitions to another member. For consumers using a non-null group.instance.id which reaches this timeout, partitions are not immediately reassigned. Instead, the consumer stops sending heartbeats and partitions are reassigned after the expiration of session.timeout.ms.

60

Enumeration, one of:

Time unit for the Maximum polling interval field. You can override this parameter at the source level.

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in the READ_COMMITTED mode, consumer.poll() returns only messages up to the last stable offset (LSO), which is one less than the offset of the first open transaction.

In particular, any messages appearing after messages belonging to ongoing transactions are withheld until the relevant transaction is completed. As a result, READ_COMMITTED consumers are not able to read up to the high watermark when there are in-flight transactions. Furthermore, when in READ_COMMITTED, the seekToEnd method returns the LSO.

READ_UNCOMMITTED

Boolean

Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.

true

Enumeration, one of:

Determines what to do when there is no initial offset in Kafka or if the current offset no longer exists on the server, for example, because the data was deleted.

LATEST

Number

Amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios.

100

Enumeration, one of:

Time unit for the Retry Backoff Timeout field.

MILLISECONDS

Boolean

Automatically check the CRC32 of the records consumed. This ensures that no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so in situations that require extremely high performance, this can be disabled.

true

Number

Size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this parameter at the source level.

64

Enumeration, one of:

Unit of measure for the Default receive buffer size field. You can override this parameter at the source level.

KB

Number

Size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default is used. You can override this parameter at the source level.

128

Enumeration, one of:

Unit of measure for the Default send buffer size field. You can override this parameter at the source level.

KB

Number

Maximum amount of time the client waits for the response of a request. If the response is not received before the timeout elapses, the client resends the request if necessary, or fails the request if the retries are exhausted. You can override this parameter at the source level.

30

Enumeration, one of:

Time unit for the Request Timeout field. You can override this parameter at the source level.

SECONDS

Number

Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.

500

Enumeration, one of:

Controls how the client uses DNS lookups. If set to USE_ALL_DNS_IPS then, when the lookup returns multiple IP addresses for a hostname, they attempt to connect before failing the connection. Applies to both bootstrap and advertised servers. If the value is RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY each entry is resolved and expanded into a list of canonical names.

USE_ALL_DNS_IPS

Number

Expected time between heartbeats to the consumer coordinator when using Kafka’s group management facilities. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

3

Enumeration, one of:

Time unit for the Heartbeat interval field.

SECONDS

Number

Timeout used to detect consumer failures when using Kafka’s group management facility. The consumer sends periodic heartbeats to indicate its aliveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker removes this consumer from the group and initiates a rebalance. The value must be in the allowable range as configured in the broker configuration by group.min.session.timeout.ms and group.max.session.timeout.ms.

10

Enumeration, one of:

Time unit for the Session Timeout field.

SECONDS

Number

Close idle connections after the number of milliseconds specified by this configuration.

540

Enumeration, one of:

Time unit for the Connection maximum idle time field.

SECONDS

Object

Additional properties used to customize the Kafka connection.

For example:

TLS

Defines a TLS configuration used by both clients and server sides to secure the communication for the Mule app. The connector automatically sets the security.protocol to use for the communication. The valid values are PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. The default value is PLAINTEXT or SASL_PLAINTEXT for SASL authentication - kerberos/scram/plain. If the broker configures SSL as the protocol, then configure at least the keystore in the tls:context child element of the configuration and the connector automatically uses SSL (or SASL_SSL for SASL authentication) as the security.protocol.

String

Endpoint identification algorithm used by clients to validate the server hostname. The default value is an empty string, which means the endpoint identification algorithm is disabled. Clients, including client connections created by the broker for inter-broker communication, verify that the broker host name matches the host name in the brokers certificate.

Array of String

List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.

Array of Topic Partition

List of topic-partition pairs to assign. Consumers are not automatically rebalanced.

Number

Minimum amount of data the server returns for a fetch request. If insufficient data is available, the request waits for the specified minimum amount of data to accumulate before answering the request. The default setting of 1 byte means that fetch requests are answered as soon as a single byte of data is available or the fetch request times out waiting for data to arrive. If you set this to a value greater than 1, the server waits for larger amounts of data to accumulate, which can improve server throughput slightly at the cost of some additional latency. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch minimum size field. You can override this parameter at the source level.

BYTE

Number

Maximum amount of data the server returns for a fetch request. Records are fetched in batches by the consumer. If the first record batch in the first non-empty partition of the fetch is larger than this value, the record batch is still returned to ensure that the consumer can make progress. As such, this is not an absolute maximum. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). The consumer performs multiple fetches in parallel. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default fetch maximum size field. You can override this parameter at the source level.

MB

Number

Maximum amount of data per partition that the server returns. The consumer fetches records in batches. If the first record batch in the first non-empty partition of the fetch is larger than this limit, the batch is still returned to ensure that the consumer can make progress. The maximum record batch size accepted by the broker is defined using message.max.bytes (broker configuration) or max.message.bytes (topic configuration). See fetch.max.bytes for limiting the consumer request size. You can override this parameter at the source level.

1

Enumeration, one of:

Unit of measure for the Default maximum partition fetch size field. You can override this parameter at the source level.

MB

Number

Maximum amount of time the server blocks before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement specified by fetch.min.bytes.

500

Enumeration, one of:

Time unit for the Fetch Maximum Wait Timeout field.

MILLISECONDS

String

ID of the token.

x

String

Token HMAC.

x

Enumeration, one of:

Encryption algorithm used by SCRAM.

x

Reconnection

Configures a reconnection strategy to use when a connector operation fails to connect to an external server.

Array of String

List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers.

x

Number

Producer attempts to batch records together into fewer requests whenever multiple records are sent to the same partition. This helps performance on both the client and the server. This configuration controls the default batch size in bytes. No attempt is made to batch records larger than this size. Requests sent to brokers contain multiple batches, one for each partition with the data that is available to send. A small batch size makes batching less common and can reduce throughput (a batch size of 0 disables batching entirely). A very large batch size can result in more wasteful use of memory as a buffer of the specified batch size is always allocated in anticipation of additional records.

16

Enumeration, one of:

Unit of measure for the Batch size field.

KB

Number

Total bytes of memory the producer uses to buffer records waiting to send to the server. If records are sent faster than they are delivered to the server, the producer blocks for max.block.ms, after which it throws an exception. This setting generally corresponds to the total memory the producer uses, but is not exact because the memory used by the producer is not all used for buffering. Some additional memory is used for compression (if compression is enabled), as well as for maintaining in-flight requests. The default value in the Apache Kafka documentation is 33554432 (32MB), but this should be capped to align with expected values for Mule instances in CloudHub (v0.1 core).

1

Enumeration, one of:

Unit of measure for the Buffer size field.

KB

Enumeration, one of:

Controls how the client uses DNS lookups. If set to USE_ALL_DNS_IPS then, when the lookup returns multiple IP addresses for a hostname, they attempt to connect before failing the connection. Applies to both bootstrap and advertised servers. If the value is RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY each entry is resolved and expanded into a list of canonical names..

USE_ALL_DNS_IPS

Enumeration, one of:

Compression type for all data generated by the producer. The default is NONE (no compression). Compression works on full batches of data, so the efficacy of batching also impacts the compression ratio. More batching means better compression.

NONE

Number

Close idle connections after the specified time is reached.

540

Enumeration, one of:

Time unit for the Connections maximum idle time field.

SECONDS

Number

Upper limit on the time to report success or failure after a call to send() returns. This limits the total time that a record is delayed prior to sending, the time to await acknowledgment from the broker (if expected), and the time allowed for retrying send failures. The producer might report failure to send a record earlier than this configuration if either an unrecoverable error is encountered, the retries have been exhausted, or the record is added to a batch that reached an earlier delivery expiration deadline. The value of this configuration must be greater than or equal to the sum of request.timeout.ms and linger.ms.

120

Enumeration, one of:

Time unit for the Delivery Timeout field.

SECONDS

Boolean

When set to true, the producer ensures that exactly one copy of each message is written in the stream. If false, the producer retries due to broker failures and might write duplicates of the retried message in the stream. Enabling idempotence requires max.in.flight.requests.per.connection to be less than or equal to 5, retries to be greater than 0. and acks to be all. If these values are not explicitly set by the user, suitable values are chosen. If incompatible values are set, a ConnectionException is thrown.

false

Number

Producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they are sent out. However, in some circumstances the client might want to reduce the number of requests, even under moderate load.

This setting accomplishes this by adding a small amount of artificial delay (rather than immediately sending out a record the producer waits for up to the given delay to allow other records to be sent so that the sends can be batched together). This is analogous to Nagle’s algorithm in TCP. This setting gives the upper bound on the delay for batching.

After the specified batch.size worth of records is received for a partition, it is sent immediately regardless of this setting, however if fewer than the specified number of bytes accumulated for this partition is received, the producer "lingers" for the specified time waiting for more records to arrive.

This setting defaults to 0 (no delay). Setting linger.ms=5, for example, has the effect of reducing the number of requests sent but would add up to 5ms of latency to records sent in the absence of load.

0

Enumeration, one of:

Time unit for the Linger time field.

SECONDS

Number

Controls for how long KafkaProducer.send() and KafkaProducer.partitionsFor() are blocked. These methods can be blocked either because the buffer is full or the metadata unavailable.Blocking in the user-supplied serializers or partitioner is not counted against this timeout.

60

Enumeration, one of:

Time unit for the Maximum block time field.

SECONDS

Number

Maximum number of unacknowledged requests the client sends on a single connection before blocking. If the value is greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (if retries are enabled).

5

Number

Maximum size of a request in bytes. This setting limits the number of record batches the producer sends in a single request to avoid sending huge requests. This is also effectively a cap on the maximum record batch size. Note that the server has its own cap on record batch size, which might be different from this.

1

Enumeration, one of:

Unit of measure for the Maximum request size field.

MB

Enumeration, one of:

Number of acknowledgments the producer requires the leader to receive before considering a request complete. This controls the durability of records that are sent.

NONE

Number

Size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. You can override this parameter at the source level.

64