Apache Kafka Connector Reference - Mule 4

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

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

Number

Determines the number of consumers the connection will initially create.

1

Number

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 resends the request, or the request fails if the specified number of retries are exhausted. This parameter can be overridden at the source level.

300

Enumeration, one of:

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

SECONDS

Enumeration, one of:

Controls how to read messages that are written transactionally.

If set to read_committed, consumer.poll() only transactional messages that have been committed are returned. If set to read_uncommitted’ (default), `consumer.poll() all messages are returned, even transactional messages that were aborted. Non-transactional messages are returned unconditionally in either mode. Messages are always returned in offset order. Hence, in 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 should be 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 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

The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. This parameter can be overridden at the source level.

64

Enumeration, one of:

The unit of measure for the receive buffer size scalar. This parameter can be overridden 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. This parameter can be overridden at the source level.

128

Enumeration, one of:

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

KB

Number

The configuration controls the maximum amount of time the client will wait 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. This parameter can be overridden at the source level.

30

Enumeration, one of:

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

SECONDS

Number

The maximum number of records returned on a poll call to the Kafka cluster. This parameter can be overridden 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, a connection is attempted to all of the IP addresses before the connection fails. 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.

DEFAULT

Number

The expected time between heartbeats to the consumer coordinator when using Apache 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 fetching the heartbeat interval time scalar.

SECONDS

Number

The timeout used to detect consumer failures when using Apache 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:

Determines the time unit for session timeout scalar.

SECONDS

Number

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

540

Enumeration, one of:

Determines the time unit for connections maximum idle time scalar.

SECONDS

TLS

Protocol to use for communication. Valid values are HTTP (default) and HTTPS. When using HTTPS the HTTP communication is secured using TLS or SSL. If HTTPS is configured as the protocol then the user needs to configure at least the keystore in the tls:context child element of the listener-config.

HTTP

Array of String

The list of subscription regular expressions to subscribe to. Topics are automatically rebalanced between the amount of consumers of the topic.

Array of Topic Partition

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

Number

The minimum amount of data the server should return 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. Setting this to something greater than 1 causes the server to wait for larger amounts of data to accumulate, which can improve server throughput slightly, at the cost of some additional latency. This parameter can be overridden at the 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 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. This parameter can be overridden at the source level.

1

Enumeration, one of:

The unit of measure for the maximum partition fetch size scalar. This parameter can be overridden at the 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 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. This parameter can be overridden at the source level.

1

Enumeration, one of:

The unit of measure for the maximum partition fetch size scalar. This parameter can be overridden at the 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 specified by fetch.min.bytes.

500

Enumeration, one of:

Determines the time unit for fetch maximum wait timeout scalar.

MILLISECONDS

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

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

x

Number

The 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 will 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 zero 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:

The unit of measure for the batch size scalar.

KB

Number

The total bytes of memory the producer can use to buffer records waiting to be sent to the server. If records are sent faster than they can be delivered to the server, the producer will block for max.block.ms after which it throws an exception. This setting should generally correspond to the total memory the producer will use, but is not exact because not all memory the producer uses is 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:

The unit of measure for the max request size scalar.

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, the connection is attempted on all of the IP addresses 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.

DEFAULT

Enumeration, one of:

The compression type for all data generated by the producer. The default is none (no compression). Valid values are none, gzip, snappy, lz4, or zstd. Compression is performed 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:

Determines the time unit for the connections maximum idle scalar.

SECONDS

Number

An upper limit on the time to report success or failure after a call to send() returns. This limits the total time that a record will be 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 should be greater than or equal to the sum of request.timeout.ms and linger.ms.

120

Enumeration, one of:

Determines the time unit for the delivery timeout scalar.

SECONDS

Boolean

When set to true, the producer will ensure that exactly one copy of each message is written in the stream. If false, the producer retries due to broker failures, and so on, and may 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 must be all. If these values are not explicitly set by the user, suitable values will be chosen. If incompatible values are set, a Connection Exception is thrown.

false

Number

The 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 can be sent out. However, in some circumstances the client may 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 will wait 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:

Determines the time unit for the linger time scalar.

SECONDS

Number

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

60

Enumeration, one of:

Determines the time unit for the maximum block time scalar.

SECONDS

Number

The maximum number of unacknowledged requests the client will send on a single connection before blocking. If this setting is set to be greater than 1 and there are failed sends, there is a risk of message reordering due to retries (if retries are enabled).

5

Number

The maximum size of a request in bytes. This setting will limit the number of record batches the producer will send 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 may be different from this.

1

Enumeration, one of:

The unit of measure for the max request size scalar.

MB

Enumeration, one of:

The 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

The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default is used. This parameter can be overridden at the source level.

64

Enumeration, one of:

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

KB

Number

Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error. Allowing retries without setting max.in.flight.requests.per.connection to 1 will potentially change the ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second succeeds, then the records in the second batch may appear first. Additionally, produce requests will be failed before the number of retries has been exhausted if the timeout configured by delivery.timeout.ms expires before successful acknowledgment. It is recommended that users leave this configuration unset and instead use delivery.timeout.ms to control retry behavior.

1

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.

1000

Enumeration, one of:

Determines the time unit for the reconnect backoff timeout scalar.

MILLISECONDS

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. This parameter can be overridden 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 will wait 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 should be larger than replica.lag.time.max.ms (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.

30

Enumeration, one of:

The request timeout time unit.

SECONDS

TLS

Protocol to use for communication. Valid values are HTTP (default) and HTTPS. When using HTTPS, the HTTP communication is secured using TLS or SSL. If HTTPS was configured as the protocol, then the user needs to configure at least the keystore in the tls:context child element of this listener-config.

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.