Apache Kafka Connector 4.10 Reference | MuleSoft Documentation
# Apache Kafka Connector 4.10 Reference
Anypoint Connector for Apache Kafka (Apache Kafka Connector) enables you to interact with the Apache Kafka messaging system. It provides seamless integration between your Mule app and an Apache Kafka cluster, using Mule runtime engine (Mule).
When a **Message Listener** or a **Batch Message Listener** uses the `MANUAL` acknowledgement mode, you must use a **Commit** operation at the end of the flow when the flow finishes successfully (either because there were no errors or because there is an **On Error Continue** component in the flow).   
---  
Mule server notifications are not supported because the Kafka library used in the connector manages network disruptions internally.   
---  
## Consumer Configuration
Configuration for consumers for Apache Kafka Connector.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
Name for this configuration. Connectors reference the configuration with this name.  
**Connection** | 
  * Consumer Plaintext Connection
  * Consumer Kerberos Connection
  * Consumer SASL/OAUTHBEARER - Client Credentials Connection
  * Consumer SASL/SCRAM Connection
  * Consumer SASL/PLAIN Connection
  * Consumer SASL/TOKEN Connection

|  Connection types for this configuration.  
**Default acknowledgement mode** |  Enumeration, one of
  * MANUAL
  * IMMEDIATE
  * DUPS_OK

|  Defines the way that the Kafka Broker instance is notified of the consumption of messages.
  * AUTO Commits messages automatically if the flow finishes successfully.
  * MANUAL The user must commit messages manually upon receipt by using the **Commit** operation.
  * IMMEDIATE Mule automatically commits the messages upon receipt before triggering the flow.
  * DUPS_OK Works in the same way as MANUAL, but the commits are asynchronous, which can lead to duplicate records.

  
**Default listener poll timeout** |  Time to wait to perform a poll if data is not available in the buffer (fetched). If no value is set, the poll is returned immediately with any records that are currently available in the buffer or else returns empty if there is no data. Must not be negative.  
**Default listener poll timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default listener poll timeout** field. This combines with **Poll Timeout** to define the total timeout for the polling. |  `MILLISECONDS`  
**Default operation poll timeout** |  Time to wait for an operation to finish. If no value is set or a negative value is set, the operation waits forever.  
**Default operation poll timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default operation timeout** field. This combines with **Operation Timeout** to define the total default timeout for the operations that use this configuration.  
Converts the provided timestamps into `ZonedLocalDateTimes` in the results. The default value is provided by the system.  
**Expiration Policy** |  Expiration Policy |  Configures the minimum amount of time that a dynamic configuration instance can remain idle before Mule considers it eligible for expiration.  
### Consumer Plaintext Connection Type
Use an unauthenticated and non-encrypted connection type.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
Default group ID for all the Kafka consumers that use this configuration.  
**Consumer Amount** |  Determines the number of consumers the connection initially creates.  
**Maximum polling interval** |  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`.  
**Maximum Polling Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum polling interval** field. You can override this parameter at the source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED

|  Controls how to read messages that are written transactionally.
  * `READ_COMMITTED` `consumer.poll()` returns only committed transactional messages are returned.
  * `READ_UNCOMMITTED` `consumer.poll()` returns all messages, even transactional messages that were aborted.

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`  
**Exclude internal topics** |  Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.  
**Auto offset reset** |  Enumeration, one of:
  * EARLIEST
  * LATEST
  * ERROR

|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
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.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Request Timeout** |  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.  
**Request Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Request Timeout** field. You can override this parameter at the source level.  
**Default record limit** |  Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY
|  Time unit for the **Request Timeout** field. You can override this parameter at the source level.  
**Default record limit** |  Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Heartbeat interval** |  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.  
**Heartbeat Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Heartbeat interval** field.  
**Session Timeout** |  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`.  
**Session timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`.  
**Endpoint identification algorithm** |  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.  
**Topic Subscription Patterns** |  Array of String |  List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.  
**Assignments** |  Array of Topic Partition |  List of topic-partition pairs to assign. Consumers are not automatically rebalanced.  
**Default fetch minimum size** |  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.  
**Fetch Minimum Size Unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch minimum size** field. You can override this parameter at the source level.  
**Default fetch maximum size** |  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.  
**Default maximum fetch size unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch maximum size** field. You can override this parameter at the source level.  
**Default maximum partition fetch size** |  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.  
**Default maximum partition fetch unit** |  Enumeration, one of: |  Unit of measure for the **Default maximum partition fetch size** field. You can override this parameter at the source level.  
**Fetch Maximum Wait Timeout** |  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`.  
**Fetch Maximum Wait Timeout Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  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.  
### Consumer Kerberos Connection Type
Use Kerberos configuration files.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
Default group ID for all the Kafka consumers that use this configuration.  
**Consumer Amount** |  Determines the number of consumers the connection initially creates.  
**Maximum polling interval** |  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`.  
**Maximum Polling Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum polling interval** field. You can override this parameter at the source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED
|  Time unit for the **Maximum polling interval** field. You can override this parameter at the source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED

|  Controls how to read messages that are written transactionally.
  * `READ_COMMITTED` `consumer.poll()` returns only committed transactional messages are returned.
  * `READ_UNCOMMITTED` `consumer.poll()` returns all messages, even transactional messages that were aborted.

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`  
**Exclude internal topics** |  Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.  
**Auto offset reset** |  Enumeration, one of:
  * EARLIEST
  * LATEST
  * ERROR

|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
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.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Request Timeout** |  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.  
**Request Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Request Timeout** field. You can override this parameter at the source level.  
**Default record limit** |  Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Heartbeat interval** |  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.  
**Heartbeat Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Heartbeat interval** field.  
**Session Timeout** |  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`.  
**Session timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.
|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`.  
**Endpoint identification algorithm** |  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.  
**Topic Subscription Patterns** |  Array of String |  List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.  
**Assignments** |  Array of Topic Partition |  List of topic-partition pairs to assign. Consumers are not automatically rebalanced.  
**Default fetch minimum size** |  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.  
**Fetch Minimum Size Unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch minimum size** field. You can override this parameter at the source level.  
**Default fetch maximum size** |  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.  
**Default maximum fetch size unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch maximum size** field. You can override this parameter at the source level.  
**Default maximum partition fetch size** |  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.  
**Default maximum partition fetch unit** |  Enumeration, one of: |  Unit of measure for the **Default maximum partition fetch size** field. You can override this parameter at the source level.  
**Fetch Maximum Wait Timeout** |  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`.  
**Fetch Maximum Wait Timeout Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Fetch Maximum Wait Timeout** field. |  `MILLISECONDS`  
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.  
**Service name** |  Kerberos principal name that Kafka runs as.  
**Kerberos configuration file (krb5.conf)** |  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.  
**Use ticket cache** |  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:
  * On Solaris and Linux, the connector looks in `/tmp/krb5cc_uid`, in which the `uid` is the numeric user identifier.
  * If the ticket cache is not available in `/tmp/krb5cc_uid` or the app is on a Windows platform, the connector looks in `{user.home}{file.separator}krb5cc_{user.name}`. You can override the ticket cache location by setting a value for the **Ticket cache** field. In a Windows environment, if a ticket cannot be retrieved from the file ticket cache, Windows uses the Local Security Authority (LSA) API to get the ticket-granting ticket (TGT).

  
**Ticket cache** |  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.  
**Use keytab** |  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.  
Set this option to the file name of the keytab to obtain the principal’s secret key.  
Set this option to `true` to store the principal’s subject private credentials.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Consumer SASL/OAUTHBEARER - Client Credentials Connection Type
OAuth Bearer authentication is a mechanism for authenticating requests using bearer tokens, which are typically used to authenticate users and applications to access resources. Apache Kafka® supports OAuth Bearer authentication for establishing connections securely. Additionally, the connector supports only RFC6749 standard.
For more information, refer to the Confluent documentation.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Endpoint identification algorithm** |  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.  
Default Group ID for all the Kafka Consumers that use this configuration.  
**Consumer Amount** |  Determines the number of consumers the connection initially creates.  
**Maximum polling interval** |  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`.  
**Maximum Polling Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  Determines the time unit for request timeout scalar. This parameter can be overridden at source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED

|  Controls how to read messages that are written transactionally.
  * `READ_COMMITTED` `consumer.poll()` returns only committed transactional messages are returned.
  * `READ_UNCOMMITTED` `consumer.poll()` returns all messages, even transactional messages that were aborted.

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  
**Exclude internal topics** |  Whether internal topics matching a subscribed pattern is excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.  
**Auto offset reset** |  Enumeration, one of:
  * EARLIEST
  * LATEST
  * ERROR

|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the reconnect backoff timeout scalar. |  MILLISECONDS  
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.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  The unit of measure for the receive buffer size scalar. You can override this field at the source level.  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  The unit of measure for the send buffer size scalar. This parameter can be overridden at source level.  
**Request Timeout** |  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.  
**Request Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for request timeout scalar. This parameter can be overridden at source level.  
**Default record limit** |  The maximum number of records returned on a poll call to the Kafka cluster. This parameter can be overridden at source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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  
**Heartbeat interval** |  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.  
**Heartbeat Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for fetch heartbeat interval time scalar.  
**Session timeout** |  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.  
**Session timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for session timeout scalar.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this config.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for connections maximum idle time scalar.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.
|  Determines the time unit for session timeout scalar.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this config.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for connections maximum idle time scalar.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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'.  
**Topic Subscription Patterns** |  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.  
**Assignments** |  Array of Topic Partition |  The list of topic-partition pairs to assign. Note that there will be no automatic rebalance of the consumers  
**Default fetch minimum size** |  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.  
**Fetch Minimum Size Unit** |  Enumeration, one of:  
**Default fetch maximum size** |  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.  
**Default maximum fetch size unit** |  Enumeration, one of: |  The unit of measure for the maximum partition fetch size scalar. This parameter can be overridden at source level.  
**Default maximum partition fetch size** |  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.  
**Default maximum partition fetch unit** |  Enumeration, one of: |  The unit of measure for the maximum partition fetch size scalar. This parameter can be overridden at source level.  
**Fetch Maximum Wait Timeout** |  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.  
**Fetch Maximum Wait Timeout Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for fetch maximum wait timeout scalar. |  MILLISECONDS  
**Client ID Key** |  Specifies the configuration key used to define the OAuth client ID within the JAAS configuration.  
The value for the Client ID which refers to the unique identifier assigned to the OAuth client, facilitating secure authentication and authorization processes.  
**Client Secret Key** |  Specifies the configuration key used to define the OAuth client secret within the JAAS configuration. |  clientSecret  
**Client Secret** |  The value for the client secret which refers to the confidential authentication credential assigned to the OAuth client.  
**OAuth Token Endpoint Key** |  Specifies the configuration key used to define the OAuth token endpoint within the JAAS configuration. |  sasl.oauthbearer.token.endpoint.url  
**OAuth Token Endpoint** |  Describes the URL where clients can request access tokens from the authentication server.  
Specifies the configuration key used to define the OAuth scope within the JAAS configuration.  
Defines the specific permissions and access rights granted to the client application by the authorization server.  
**Audience Key** |  Specifies the configuration key used to define the OAuth audience within the JAAS configuration.  
Comma separated list of one or more target audiences for the access token.  
**Max Token Expiration Seconds** |  Sets the maximum duration, in seconds, for which an access token remains valid. This helps prevent unauthorized access by limiting the access token’s lifespan, ensuring that even if the access token is intercepted, it becomes invalid after a set period, thus enhancing security and enforcing re-authentication. The default value is null (not set). When this field is not configured, the connector uses the actual token expiration time provided by the OAuth server in the token response (`expires_in` field). The token is valid for its full natural lifespan as determined by the authorization server. Leaving this field unset uses the OAuth server’s intended token lifespan, balances security and performance, and follows best practices. If this value is set longer than the actual token TTL, the token expires at its natural expiration time as provided by the OAuth server, not at the configured maximum. If this value is set to the same value as the actual token TTL, the token expires at the configured time, which happens to match the server’s intended expiration time.  
**Principal Name** |  Unique identifier associated with an authenticated user, often used for authorization decisions and resource access.  
**Credentials Placement** |  Enumeration, one of:
  * BASIC_AUTHENTICATION_HEADER
|  Defines whether to include client credentials in the basic authentication header or in the body of the authentication request. |  BASIC_AUTHENTICATION_HEADER  
**OAuth Module Required** |  Indicates whether this login module must be successfully authenticated for Kafka clients to establish a connection using OAuth 2.0 bearer tokens.  
**Include Accept Header** |  Whether to include 'Accept-application/json' header in the authentication request.  
**OAuth Extensions** |  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. You don’t need to include the `extension_` part, such as in `extension_logicalCluster` or `extension_identityPoolId`.  
**Login Refresh Window Factor** |  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.  
**Login Refresh Window Jitter** |  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.  
**Login Refresh Minimum Period** |  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).  
**Login Refresh Buffer** |  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.  
**Use Persistent Connections** |  Indicates whether to use persistent connections: Mule uses persistent connections. Mule closes the connection after the first request completes.  
**Max Connections** |  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.  
**Connection Idle Timeout** |  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.  
**Stream Response** |  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.  
**Response Buffer Size** |  Size of the buffer that stores the HTTP response, in bytes. By default, the space is not limited.  
**Client Socket Properties** |  Tcp Client Socket Params |  Encapsulates the configurable properties for TCP client socket connections.  
**Proxy Config Params** |  Proxy Config Params |  Configures a proxy for outbound connections.  
**Follow Redirects** |  Specifies whether to follow redirects or not.  
**Response Timeout** |  Maximum time that the request element will block the execution of the flow waiting for the HTTP response.  
**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  
### Consumer SASL/SCRAM Connection Type
Use Salted Challenge Response Authentication Mechanism (SCRAM) or SASL/SCRAM, a family of SASL mechanisms that addresses the security concerns with traditional mechanisms that perform username and password authentication like PLAIN. Apache Kafka supports SCRAM-SHA-256 and SCRAM-SHA-512.
For more information, refer to the Confluent documentation.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
Default group ID for all the Kafka consumers that use this configuration.  
**Consumer Amount** |  Determines the number of consumers the connection initially creates.  
**Maximum polling interval** |  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`.  
**Maximum Polling Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum polling interval** field. You can override this parameter at the source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED

|  Controls how to read messages that are written transactionally.
  * `READ_COMMITTED` `consumer.poll()` returns only committed transactional messages are returned.
  * `READ_UNCOMMITTED` `consumer.poll()` returns all messages, even transactional messages that were aborted.

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`  
**Exclude internal topics** |  Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.  
**Auto offset reset** |  Enumeration, one of:
  * EARLIEST
  * LATEST
  * ERROR

|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
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.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Request Timeout** |  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.  
**Request Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  Time unit for the **Request Timeout** field. You can override this parameter at the source level.  
**Default record limit** |  Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Heartbeat interval** |  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.  
**Heartbeat Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Heartbeat interval** field.  
**Session Timeout** |  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`.  
**Session timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`.  
**Endpoint identification algorithm** |  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.  
**Topic Subscription Patterns** |  Array of String |  List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.  
**Assignments** |  Array of Topic Partition |  List of topic-partition pairs to assign. Consumers are not automatically rebalanced.  
**Default fetch minimum size** |  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.  
**Fetch Minimum Size Unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch minimum size** field. You can override this parameter at the source level.  
**Default fetch maximum size** |  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.  
**Default maximum fetch size unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch maximum size** field. You can override this parameter at the source level.  
**Default maximum partition fetch size** |  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.  
**Default maximum partition fetch unit** |  Enumeration, one of: |  Unit of measure for the **Default maximum partition fetch size** field. You can override this parameter at the source level.  
**Fetch Maximum Wait Timeout** |  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`.  
**Fetch Maximum Wait Timeout Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Fetch Maximum Wait Timeout** field. |  `MILLISECONDS`  
Username with which to login.  
Password with which to login.  
**Encryption type** |  Enumeration, one of:
  * SHA256
  * SHA512

|  Encryption algorithm used by SCRAM.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Consumer SASL/PLAIN Connection Type
Use SASL authenticated with a username and password.
For more information, refer to the Confluent documentation.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
Default group ID for all the Kafka consumers that use this configuration.  
**Consumer Amount** |  Determines the number of consumers the connection initially creates.  
**Maximum polling interval** |  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`.  
**Maximum Polling Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  Time unit for the **Maximum polling interval** field. You can override this parameter at the source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED

|  Controls how to read messages that are written transactionally.
  * `READ_COMMITTED` `consumer.poll()` returns only committed transactional messages are returned.
  * `READ_UNCOMMITTED` `consumer.poll()` returns all messages, even transactional messages that were aborted.

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`  
**Exclude internal topics** |  Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.  
**Auto offset reset** |  Enumeration, one of:
  * EARLIEST
  * LATEST
  * ERROR

|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
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.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Request Timeout** |  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.  
**Request Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Request Timeout** field. You can override this parameter at the source level.  
**Default record limit** |  Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Heartbeat interval** |  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.  
**Heartbeat Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Heartbeat interval** field.  
**Session Timeout** |  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`.  
**Session timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.
|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`.  
**Endpoint identification algorithm** |  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.  
**Topic Subscription Patterns** |  Array of String |  List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.  
**Assignments** |  Array of Topic Partition |  List of topic-partition pairs to assign. Consumers are not automatically rebalanced.  
**Default fetch minimum size** |  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.  
**Fetch Minimum Size Unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch minimum size** field. You can override this parameter at the source level.  
**Default fetch maximum size** |  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.  
**Default maximum fetch size unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch maximum size** field. You can override this parameter at the source level.  
**Default maximum partition fetch size** |  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.  
**Default maximum partition fetch unit** |  Enumeration, one of: |  Unit of measure for the **Default maximum partition fetch size** field. You can override this parameter at the source level.  
**Fetch Maximum Wait Timeout** |  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`.  
**Fetch Maximum Wait Timeout Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Fetch Maximum Wait Timeout** field. |  `MILLISECONDS`  
User used by the client to connect to the Kafka broker.  
Password used by the client to connect to the Kafka broker.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Consumer SASL/TOKEN Connection Type
Use delegation tokens to authenticate to the Kafka cluster.
Due to security reasons, a delegation token cannot be renewed if the initial authentication uses a delegation token. For more information, refer to Delegation Token Support for Kafka.   
---  
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
Default group ID for all the Kafka consumers that use this configuration.  
**Consumer Amount** |  Determines the number of consumers the connection initially creates.  
**Maximum polling interval** |  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`.  
**Maximum Polling Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum polling interval** field. You can override this parameter at the source level.  
**Isolation Level** |  Enumeration, one of:
  * READ_UNCOMMITTED
  * READ_COMMITTED

|  Controls how to read messages that are written transactionally.
  * `READ_COMMITTED` `consumer.poll()` returns only committed transactional messages are returned.
  * `READ_UNCOMMITTED` `consumer.poll()` returns all messages, even transactional messages that were aborted.

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`  
**Exclude internal topics** |  Determines whether internal topics matching a subscribed pattern are excluded from the subscription. It is always possible to explicitly subscribe to an internal topic.  
**Auto offset reset** |  Enumeration, one of:
  * EARLIEST
  * LATEST
  * ERROR

|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  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.
  * EARLIEST Automatically reset the offset to the earliest offset.
  * LATEST Automatically reset the offset to the latest offset.
  * ERROR Throw an error if no previous offset is found for the consumer’s group.

  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
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.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Request Timeout** |  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.  
**Request Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Request Timeout** field. You can override this parameter at the source level.  
**Default record limit** |  Maximum number of records returned on a poll call to the Kafka cluster. You can override this parameter at the source level.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Heartbeat interval** |  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.  
**Heartbeat Interval Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Heartbeat interval** field.  
**Session Timeout** |  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`.  
**Session timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.
|  Time unit for the **Session Timeout** field.  
**Connection maximum idle time** |  Close idle connections after the number of milliseconds specified by this configuration.  
**Connection maximum idle time time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connection maximum idle time** field.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`.  
**Endpoint identification algorithm** |  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.  
**Topic Subscription Patterns** |  Array of String |  List of subscription regular expressions to which to subscribe. Topics are automatically rebalanced between the topic consumers.  
**Assignments** |  Array of Topic Partition |  List of topic-partition pairs to assign. Consumers are not automatically rebalanced.  
**Default fetch minimum size** |  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.  
**Fetch Minimum Size Unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch minimum size** field. You can override this parameter at the source level.  
**Default fetch maximum size** |  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.  
**Default maximum fetch size unit** |  Enumeration, one of: |  Unit of measure for the **Default fetch maximum size** field. You can override this parameter at the source level.  
**Default maximum partition fetch size** |  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.  
**Default maximum partition fetch unit** |  Enumeration, one of: |  Unit of measure for the **Default maximum partition fetch size** field. You can override this parameter at the source level.  
**Fetch Maximum Wait Timeout** |  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`.  
**Fetch Maximum Wait Timeout Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Fetch Maximum Wait Timeout** field. |  `MILLISECONDS`  
ID of the token.  
**Token HMAC** |  Token HMAC.  
**Encryption type** |  Enumeration, one of:
  * SCRAM_SHA_256
  * SCRAM_SHA_512

|  Encryption algorithm used by SCRAM.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
## Producer Configuration
Configuration for producers for Apache Kafka Connector.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
Name for this configuration. Connectors reference the configuration with this name.  
**Connection** | 
  * Producer Plaintext Connection
  * Producer Kerberos Connection
  * Producer SASL/OAUTHBEARER - Client Credentials Connection
  * Producer SASL/SCRAM Connection
  * Producer SASL/PLAIN Connection
  * Producer SASL/TOKEN Connection

|  Connection types for this configuration.  
**Default topic** |  Default topic name to use by the producer operations, overridable at the operation’s configuration level. |  `defaultTopicName`  
Converts the provided timestamps into `ZonedLocalDateTimes` in the results. The default value is provided by the system.  
**Expiration Policy** |  Expiration Policy |  Configures the minimum amount of time that a dynamic configuration instance can remain idle before Mule considers it eligible for expiration.  
### Producer Plaintext Connection Type
Use an unauthenticated and non-encrypted connection type.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Batch size** |  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.  
**The batch size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Batch size** field.  
**Buffer size** |  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).  
**The buffer memory size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Buffer size** field.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY
|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connections maximum idle time** field.  
**Delivery Timeout** |  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`.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Delivery Timeout** field.  
**Enable idempotence** |  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.  
**Linger time** |  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.  
**Linger Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Linger time** field.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum block time** field.  
**Maximum in flight requests** |  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).  
**Maximum request size** |  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.  
**Maximum request size unit** |  Enumeration, one of: |  Unit of measure for the **Maximum request size** field.  
**Producer Acknowledge Mode** |  Enumeration, one of:
  * LEADER_ONLY

|  Number of acknowledgments the producer requires the leader to receive before considering a request complete. This controls the durability of records that are sent.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Retries amount** |  Setting a value greater than `0` causes 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` potentially changes 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 might 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.  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Default request timeout** |  Maximum amount of time the client waits for the response to a request. If the response is not received before the timeout elapses, the client resends the request if necessary or fails the request if retries are exhausted. This value must be larger than `replica.lag.time.max.ms` (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.  
**Default request timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.
|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`. If the broker configures `SSL` as the protocol, configure the keystore in the `tls:context` child element of the configuration and the connector will automatically use `SSL` (or `SASL_SSL` for SASL authentication) as the `security.protocol`.  
**Endpoint identification algorithm** |  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.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Producer Kerberos Connection Type
Use Kerberos configuration files.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Batch size** |  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.  
**The batch size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Batch size** field.  
**Buffer size** |  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).  
**The buffer memory size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Buffer size** field.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connections maximum idle time** field.  
**Delivery Timeout** |  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`.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Delivery Timeout** field.  
**Enable idempotence** |  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.  
**Linger time** |  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.  
**Linger Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Linger time** field.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  Time unit for the **Linger time** field.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum block time** field.  
**Maximum in flight requests** |  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).  
**Maximum request size** |  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.  
**Maximum request size unit** |  Enumeration, one of: |  Unit of measure for the **Maximum request size** field.  
**Producer Acknowledge Mode** |  Enumeration, one of:
  * LEADER_ONLY

|  Number of acknowledgments the producer requires the leader to receive before considering a request complete. This controls the durability of records that are sent.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Retries amount** |  Setting a value greater than `0` causes 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` potentially changes 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 might 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.  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Default request timeout** |  Maximum amount of time the client waits for the response to a request. If the response is not received before the timeout elapses, the client resends the request if necessary or fails the request if retries are exhausted. This value must be larger than `replica.lag.time.max.ms` (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.  
**Default request timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`. If the broker configures `SSL` as the protocol, configure the keystore in the `tls:context` child element of the configuration and the connector will automatically use `SSL` (or `SASL_SSL` for SASL authentication) as the `security.protocol`.  
**Endpoint identification algorithm** |  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.  
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.  
**Service name** |  Kerberos principal name that Kafka runs as.  
**Kerberos configuration file (krb5.conf)** |  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.  
**Use ticket cache** |  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:
  * On Solaris and Linux, the connector looks in `/tmp/krb5cc_uid`, in which the `uid` is the numeric user identifier.
  * If the ticket cache is not available in `/tmp/krb5cc_uid` or the app is on a Windows platform, the connector looks in `{user.home}{file.separator}krb5cc_{user.name}`. You can override the ticket cache location by setting a value for the **Ticket cache** field. In a Windows environment, if a ticket cannot be retrieved from the file ticket cache, Windows uses the Local Security Authority (LSA) API to get the ticket-granting ticket (TGT).
**Ticket cache** |  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.  
**Use keytab** |  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.  
Set this option to the file name of the keytab to obtain the principal’s secret key.  
Set this option to `true` to store the principal’s subject private credentials.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Producer SASL/OAUTHBEARER - Client Credentials Connection Type
OAuth Bearer authentication is a mechanism for authenticating requests using bearer tokens, which are typically used to authenticate users and applications to access resources. Apache Kafka® supports OAuth Bearer authentication for establishing connections securely.
For more information, refer to the Confluent documentation.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Endpoint identification algorithm** |  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.  
**Batch size** |  The producer attempts to batch records together into fewer requests whenever multiple records are being 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 will be made to batch records larger than this size. Requests sent to brokers will contain multiple batches, one for each partition with data available to be sent. A small batch size will make batching less common and may reduce throughput (a batch size of zero will disable batching entirely). A very large batch size may use memory a bit more wastefully as we will always allocate a buffer of the specified batch size in anticipation of additional records.  
**The batch size unit of measure.** |  Enumeration, one of: |  The unit of measure for the batch size scalar.  
**Buffer size** |  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 will throw an exception. This setting should correspond roughly to the total memory the producer will use, but is not a hard bound since not all memory the producer uses is used for buffering. Some additional memory will be used for compression (if compression is enabled) as well as for maintaining in-flight requests. The default value in the Kafka docs is of 33554432 (32MB), but this should be capped to align with expected values for mule instances in cloudhub (v0.1 core)  
**The buffer memory size unit of measure.** |  Enumeration, one of: |  The unit of measure for the max request size scalar.  
**DNS lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  The compression type for all data generated by the producer. The default is none (i.e. no compression). Valid values are none, gzip, snappy, lz4, or zstd. Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression).  
**Connections maximum idle time** |  Close idle connections after the value specified by this config.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the connections maximum idle scalar.  
**Delivery timeout** |  An upper bound 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 retriable send failures. The producer may report failure to send a record earlier than this config if either an unrecoverable error is encountered, the retries have been exhausted, or the record is added to a batch which reached an earlier delivery expiration deadline. The value of this config should be greater than or equal to the sum of request.timeout.ms and linger.ms.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the delivery timeout scalar.  
**Enable idempotence** |  When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer retries due to broker failures, etc, may write duplicates of the retried message in the stream. Note that 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 ConnectionException will be thrown  
**Linger time** |  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?that is, 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 can be thought of as analogous to Nagle’s algorithm in TCP. This setting gives the upper bound on the delay for batching: once we get batch.size worth of records for a partition it is sent immediately regardless of this setting, however if we have fewer than this many bytes accumulated for this partition we will 'linger' for the specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting linger.ms=5, for example, would have 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.  
**Linger Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the linger time scalar.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the maximum block time scalar.  
**Maximum in flight requests** |  The maximum number of unacknowledged requests the client will send on a single connection before blocking. Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (i.e., if retries are enabled).  
**Maximum request size** |  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.  
**Maximum request size unit.** |  Enumeration, one of: |  The unit of measure for the max request size scalar.  
**Producer acknowledge mode** |  Enumeration, one of:
  * LEADER_ONLY
|  Determines the time unit for the maximum block time scalar.  
**Maximum in flight requests** |  The maximum number of unacknowledged requests the client will send on a single connection before blocking. Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of message re-ordering due to retries (i.e., if retries are enabled).  
**Maximum request size** |  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.  
**Maximum request size unit.** |  Enumeration, one of: |  The unit of measure for the max request size scalar.  
**Producer acknowledge mode** |  Enumeration, one of:
  * LEADER_ONLY

|  The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent  
**Default receive buffer size** |  The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default will be used. This parameter can be overridden at source level.  
**Default receive buffer size unit** |  Enumeration, one of: |  The unit of measure for the receive buffer size scalar. This parameter can be overridden at source level.  
**Retries amount** |  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. Note additionally that produce requests will be failed before the number of retries has been exhausted if the timeout configured by delivery.timeout.ms expires first before successful acknowledgment. Users should generally prefer to leave this config unset and instead use delivery.timeout.ms to control retry behavior.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the retry backoff timeout time scalar. |  MILLISECONDS  
**Retry backoff timeout** |  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.  
**Default send buffer size** |  The size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default will be used. This parameter can be overridden at source level.  
**Default send buffer size unit** |  Enumeration, one of: |  The unit of measure for the send buffer size scalar. This parameter can be overridden at source level.  
**Default request timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Determines the time unit for the request timeout time scalar.  
**Default request timeout** |  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 should be larger than replica.lag.time.max.ms (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  The configuration controls which partitioning strategy is used when sending messages without providing a key or a partition.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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'.  
**Client ID Key** |  Specifies the configuration key used to define the OAuth client ID within the JAAS configuration.  
The value for the Client ID which refers to the unique identifier assigned to the OAuth client, facilitating secure authentication and authorization processes.  
**Client Secret Key** |  Specifies the configuration key used to define the OAuth client secret within the JAAS configuration. |  clientSecret  
**Client Secret** |  The value for the client secret which refers to the confidential authentication credential assigned to the OAuth client.  
**OAuth Token Endpoint Key** |  Specifies the configuration key used to define the OAuth token endpoint within the JAAS configuration. |  sasl.oauthbearer.token.endpoint.url  
**OAuth Token Endpoint** |  Describes the URL where clients can request access tokens from the authentication server.  
Specifies the configuration key used to define the OAuth scope within the JAAS configuration.  
Defines the specific permissions and access rights granted to the client application by the authorization server.  
**Audience Key** |  Specifies the configuration key used to define the OAuth audience within the JAAS configuration.  
Comma separated list of one or more target audiences for the access token.  
**Max Token Expiration Seconds** |  Sets the maximum duration, in seconds, for which an access token remains valid. This helps prevent unauthorized access by limiting the access token’s lifespan, ensuring that even if the access token is intercepted, it becomes invalid after a set period, thus enhancing security and enforcing re-authentication. The default value is null (not set). When this field is not configured, the connector uses the actual token expiration time provided by the OAuth server in the token response (`expires_in` field). The token is valid for its full natural lifespan as determined by the authorization server. Leaving this field unset uses the OAuth server’s intended token lifespan, balances security and performance, and follows best practices. If this value is set longer than the actual token TTL, the token expires at its natural expiration time as provided by the OAuth server, not at the configured maximum. If this value is set to the same value as the actual token TTL, the token expires at the configured time, which happens to match the server’s intended expiration time.  
**Principal Name** |  Unique identifier associated with an authenticated user, often used for authorization decisions and resource access.  
**Credentials Placement** |  Enumeration, one of:
  * BASIC_AUTHENTICATION_HEADER
|  Defines whether to include client credentials in the basic authentication header or in the body of the authentication request. |  BASIC_AUTHENTICATION_HEADER  
**OAuth Module Required** |  Indicates whether this login module must be successfully authenticated for Kafka clients to establish a connection using OAuth 2.0 bearer tokens.  
**Include Accept Header** |  Whether to include 'Accept-application/json' header in the authentication request.  
**OAuth Extensions** |  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. You don’t need to include the `extension_` part, such as in `extension_logicalCluster` or `extension_identityPoolId`.  
**Login Refresh Window Factor** |  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.  
**Login Refresh Window Jitter** |  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.  
**Login Refresh Minimum Period** |  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).  
**Login Refresh Buffer** |  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.  
**Use Persistent Connections** |  Indicates whether to use persistent connections: Mule uses persistent connections. Mule closes the connection after the first request completes.  
**Max Connections** |  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.  
**Connection Idle Timeout** |  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.  
**Stream Response** |  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.  
**Response Buffer Size** |  Size of the buffer that stores the HTTP response, in bytes. By default, the space is not limited.  
**Client Socket Properties** |  Tcp Client Socket Params |  Encapsulates the configurable properties for TCP client socket connections.  
**Proxy Config Params** |  Proxy Config Params |  Configures a proxy for outbound connections.  
**Follow Redirects** |  Specifies whether to follow redirects or not.  
**Response Timeout** |  Maximum time that the request element will block the execution of the flow waiting for the HTTP response.  
**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  
### Producer SASL/SCRAM Connection Type
Use Salted Challenge Response Authentication Mechanism (SCRAM) or SASL/SCRAM, a family of SASL mechanisms that addresses the security concerns with traditional mechanisms that perform username and password authentication like PLAIN. Apache Kafka supports SCRAM-SHA-256 and SCRAM-SHA-512.
For more information, refer to the Confluent documentation.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Batch size** |  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.  
**The batch size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Batch size** field.  
**Buffer size** |  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).  
**The buffer memory size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Buffer size** field.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connections maximum idle time** field.  
**Delivery Timeout** |  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`.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connections maximum idle time** field.  
**Delivery Timeout** |  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`.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Delivery Timeout** field.  
**Enable idempotence** |  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.  
**Linger time** |  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.  
**Linger Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Linger time** field.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum block time** field.  
**Maximum in flight requests** |  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).  
**Maximum request size** |  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.  
**Maximum request size unit** |  Enumeration, one of: |  Unit of measure for the **Maximum request size** field.  
**Producer Acknowledge Mode** |  Enumeration, one of:
  * LEADER_ONLY

|  Number of acknowledgments the producer requires the leader to receive before considering a request complete. This controls the durability of records that are sent.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Retries amount** |  Setting a value greater than `0` causes 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` potentially changes 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 might 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.  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Default request timeout** |  Maximum amount of time the client waits for the response to a request. If the response is not received before the timeout elapses, the client resends the request if necessary or fails the request if retries are exhausted. This value must be larger than `replica.lag.time.max.ms` (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.  
**Default request timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`. If the broker configures `SSL` as the protocol, configure the keystore in the `tls:context` child element of the configuration and the connector will automatically use `SSL` (or `SASL_SSL` for SASL authentication) as the `security.protocol`.  
**Endpoint identification algorithm** |  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.  
Username with which to login.  
Password with which to login.  
**EncryptionType** |  Enumeration, one of:
  * SHA256
  * SHA512
**TLS Configuration** |  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`. If the broker configures `SSL` as the protocol, configure the keystore in the `tls:context` child element of the configuration and the connector will automatically use `SSL` (or `SASL_SSL` for SASL authentication) as the `security.protocol`.  
**Endpoint identification algorithm** |  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.  
Username with which to login.  
Password with which to login.  
**EncryptionType** |  Enumeration, one of:
  * SHA256
  * SHA512

|  Encryption algorithm used by SCRAM.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Producer SASL/PLAIN Connection Type
Use SASL authenticated with a username and password.
For more information, refer to the Confluent documentation.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Batch size** |  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 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.  
**The batch size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Batch size** field.  
**Buffer size** |  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).  
**The buffer memory size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Buffer size** field.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connections maximum idle time** field.  
**Delivery Timeout** |  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`.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Delivery Timeout** field.  
**Enable idempotence** |  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.  
**Linger time** |  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.  
**Linger Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Linger time** field.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum block time** field.  
**Maximum in flight requests** |  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).  
**Maximum request size** |  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.  
**Maximum request size unit** |  Enumeration, one of: |  Unit of measure for the **Maximum request size** field.  
**Producer Acknowledge Mode** |  Enumeration, one of:
  * LEADER_ONLY
|  Time unit for the **Maximum block time** field.  
**Maximum in flight requests** |  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).  
**Maximum request size** |  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.  
**Maximum request size unit** |  Enumeration, one of: |  Unit of measure for the **Maximum request size** field.  
**Producer Acknowledge Mode** |  Enumeration, one of:
  * LEADER_ONLY

|  Number of acknowledgments the producer requires the leader to receive before considering a request complete. This controls the durability of records that are sent.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Retries amount** |  Setting a value greater than `0` causes 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` potentially changes 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 might 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.  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Default request timeout** |  Maximum amount of time the client waits for the response to a request. If the response is not received before the timeout elapses, the client resends the request if necessary or fails the request if retries are exhausted. This value must be larger than `replica.lag.time.max.ms` (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.  
**Default request timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`. If the broker configures `SSL` as the protocol, configure the keystore in the `tls:context` child element of the configuration and the connector will automatically use `SSL` (or `SASL_SSL` for SASL authentication) as the `security.protocol`.  
**Endpoint identification algorithm** |  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.  
Username with which to login.  
Password with which to login.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
### Producer SASL/TOKEN Connection Type
Use delegation tokens to authenticate to the Kafka cluster.
Due to security reasons, a delegation token cannot be renewed if the initial authentication uses a delegation token. For more information, refer to Delegation Token Support for Kafka.   
---  
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Bootstrap Server URLs** |  Array of String |  List of servers to bootstrap the connection with the Kafka cluster. This can be a partial list of the available servers. For example, `localhost:9071`.  
**Batch size** |  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 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.  
**The batch size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Batch size** field.  
**Buffer size** |  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).  
**The buffer memory size unit of measure.** |  Enumeration, one of: |  Unit of measure for the **Buffer size** field.  
**DNS Lookups** |  Enumeration, one of:
  * DEFAULT
  * USE_ALL_DNS_IPS
  * RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY

|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS
|  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` 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.
  * `RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY` Each entry is resolved and expanded into a list of canonical names.

|  `USE_ALL_DNS_IPS`  
**Compression type** |  Enumeration, one of:
  * SNAPPY

|  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.  
**Connections maximum idle time** |  Close idle connections after the specified time is reached.  
**Connections maximum idle time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Connections maximum idle time** field.  
**Delivery Timeout** |  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`.  
**Delivery Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Delivery Timeout** field.  
**Enable idempotence** |  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.  
**Linger time** |  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.  
**Linger Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Linger time** field.  
**Maximum block time** |  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.  
**Maximum block time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Maximum block time** field.  
**Maximum in flight requests** |  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).  
**Maximum request size** |  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.  
**Maximum request size unit** |  Enumeration, one of: |  Unit of measure for the **Maximum request size** field.  
**Producer Acknowledge Mode** |  Enumeration, one of:
  * LEADER_ONLY

|  Number of acknowledgments the producer requires the leader to receive before considering a request complete. This controls the durability of records that are sent.  
**Default receive buffer size** |  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.  
**Default receive buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default receive buffer size** field. You can override this parameter at the source level.  
**Retries amount** |  Setting a value greater than `0` causes 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` potentially changes 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 might 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.  
**Retry Backoff Timeout** |  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.  
**Retry Backoff Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Retry Backoff Timeout** field. |  `MILLISECONDS`  
**Default send buffer size** |  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.  
**Default send buffer size unit** |  Enumeration, one of: |  Unit of measure for the **Default send buffer size** field. You can override this parameter at the source level.  
**Default request timeout** |  Maximum amount of time the client waits for the response to a request. If the response is not received before the timeout elapses, the client resends the request if necessary or fails the request if retries are exhausted. This value must be larger than `replica.lag.time.max.ms` (a broker configuration) to reduce the possibility of message duplication due to unnecessary producer retries.  
**Default request timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.
|  Time unit for the **Default request timeout** field.  
**Partitioner** |  Enumeration, one of:
  * DEFAULT
  * ROUND_ROBIN
  * UNIFORM_STICKY

|  Controls the partitioning strategy.  
**Additional Properties** |  Additional properties used to customize the Kafka connection. For example:
  * For serialization, set **Key** to `key.serializer` or `value.serializer` and **Value** to `com.mulesoft.connectors.kafka.internal.model.serializer.MuleKafkaAvroSerializer`.
  * For deserialization, set **Key** to `key.deserializer` or `value.deserializer` and **Value** to `io.confluent.kafka.serializers.KafkaAvroDeserializer`.

  
**TLS Configuration** |  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`. If the broker configures `SSL` as the protocol, configure the keystore in the `tls:context` child element of the configuration and the connector will automatically use `SSL` (or `SASL_SSL` for SASL authentication) as the `security.protocol`.  
**Endpoint identification algorithm** |  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.  
ID of the token.  
**Token HMAC** |  Token HMAC.  
**Encryption type** |  Enumeration, one of:
  * SCRAM_SHA_256
  * SCRAM_SHA_512

|  Encryption algorithm used by SCRAM.  
**Reconnection** |  |  Configures a reconnection strategy to use when a connector operation fails to connect to an external server.  
## Consumer Sources
  * Batch Message Listener
  * Message Listener


### Batch Message Listener
`<kafka:batch-message-listener>`
This source supports the consumption of messages from a Kafka cluster, producing a list of messages to the flow.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
**Poll timeout** |  Amount of time to block. Defines the total timeout for polling.  
**Poll timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Poll timeout** field.  
**Acknowledgement mode** |  Enumeration, one of:
  * MANUAL
  * IMMEDIATE
  * DUPS_OK

|  Defines the way that the Kafka Broker instance is notified of the consumption of messages.
  * AUTO Commits messages automatically if the flow finishes successfully.
  * MANUAL The user must commit messages manually upon receipt by using the **Commit** operation.
  * IMMEDIATE Mule automatically commits the messages upon receipt before triggering the flow.
  * DUPS_OK Works in the same way as MANUAL, but the commits are asynchronous, which can lead to duplicate records.

  
**Amount of parallel consumers.** |  Number of parallel consumers.  
**Primary Node Only** |  Determines whether to execute this source on only the primary node when running Mule instances in a cluster.  
**Redelivery Policy** |  Redelivery Policy |  Defines a policy for processing the redelivery of the same message.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Output
Array of Record  
---  
Attributes Type |  Consumer Context  
#### Associated Configurations
  * Consumer Configuration


The **Batch Message Listener** source does not support configurable streaming strategies because the source takes a batch of records as an input stream. The streaming strategy configuration is non-repeatable-stream by default for the source.   
---  
### Message Listener
`<kafka:message-listener>`
This source supports the consumption of messages from a Kafka cluster, producing a single message to the flow.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
**Poll timeout** |  Amount of time to block. Defines the total timeout for polling.  
**Poll timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Poll timeout** field.  
**Acknowledgement mode** |  Enumeration, one of:
  * MANUAL
  * IMMEDIATE
  * DUPS_OK

|  Defines the way that the Kafka Broker instance is notified of the consumption of messages.
  * AUTO Commits messages automatically if the flow finishes successfully.
  * MANUAL The user must commit messages manually upon receipt by using the **Commit** operation.
  * IMMEDIATE Mule automatically commits the messages upon receipt before triggering the flow.
  * DUPS_OK Works in the same way as MANUAL, but the commits are asynchronous, which can lead to duplicate records.

  
**Amount of parallel consumers.** |  Number of parallel consumers.  
**Primary Node Only** |  Determines whether to execute this source on only the primary node when running Mule instances in a cluster.  
**Streaming Strategy** | 
  * Repeatable In Memory Stream
  * Repeatable File Store Stream
  * non-repeatable-stream

|  Configures how Mule processes streams. Repeatable streams are the default behavior.  
**Redelivery Policy** |  Redelivery Policy |  Defines a policy for processing the redelivery of the same message.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Output
Attributes Type |  Kafka Record Attributes  
---|---  
#### Associated Configurations
  * Consumer Configuration


## Consumer Operations


### Commit
`<kafka:commit>`
Commits the offsets associated to a message or batch of messages consumed in a **Message Listener**. This is a list or a single message consumed in the **Batch Message Listener** source.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
**Consumer commit key** |  Commit key of the last poll. This operation is valid only when used inside a flow that uses one of the sources (**Batch Message Listener** or **Message Listener**) which inserts this value as an attribute in the Mule event.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Associated Configurations
  * Consumer Configuration


#### Throws
  * KAFKA:INVALID_ACK_MODE
  * KAFKA:RETRY_EXHAUSTED
  * KAFKA:ALREADY_COMMITTED
  * KAFKA:TIMEOUT
  * KAFKA:SESSION_NOT_FOUND
  * KAFKA:NOT_FOUND
  * KAFKA:CONNECTIVITY


### Consume
`<kafka:consume>`
Receives messages from one or more Kafka topics. This operation works similarly to the **Message Listener** source, so all the operations that apply to that, apply to this operation as well.
The **Consume** operation does not return the `consumerCommitKey`.   
---  
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
**Consumption timeout** |  Number of time units that this operation waits for receiving messages.  
**Timeout time unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Consumption timeout** field.  
**Operation Timeout** |  Timeout for the operation to start executing.  
**Operation Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Operation Timeout** field.  
**Acknowledgement mode** |  Enumeration, one of:
  * MANUAL
  * IMMEDIATE
  * DUPS_OK

|  Defines the way that the Kafka Broker instance is notified of the consumption of messages.
  * AUTO Commits messages automatically if the flow finishes successfully.
  * MANUAL The user must commit messages manually upon receipt by using the **Commit** operation.
  * IMMEDIATE Mule automatically commits the messages upon receipt before triggering the flow.
  * DUPS_OK Works in the same way as MANUAL, but the commits are asynchronous, which can lead to duplicate records.

  
**Streaming Strategy** | 
  * Repeatable In Memory Stream
  * Repeatable File Store Stream
  * non-repeatable-stream
|  Time unit for the **Consumption timeout** field.  
**Operation Timeout** |  Timeout for the operation to start executing.  
**Operation Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Operation Timeout** field.  
**Acknowledgement mode** |  Enumeration, one of:
  * MANUAL
  * IMMEDIATE
  * DUPS_OK

|  Defines the way that the Kafka Broker instance is notified of the consumption of messages.
  * AUTO Commits messages automatically if the flow finishes successfully.
  * MANUAL The user must commit messages manually upon receipt by using the **Commit** operation.
  * IMMEDIATE Mule automatically commits the messages upon receipt before triggering the flow.
  * DUPS_OK Works in the same way as MANUAL, but the commits are asynchronous, which can lead to duplicate records.

  
**Streaming Strategy** | 
  * Repeatable In Memory Stream
  * Repeatable File Store Stream
  * non-repeatable-stream

|  Configures how Mule processes streams. Repeatable streams are the default behavior.  
**Target Variable** |  Name of the variable that stores the operation’s output.  
**Target Value** |  Expression that evaluates the operation’s output. The outcome of the expression is stored in the **Target Variable** field.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Output
Attributes Type |  Kafka Record Attributes  
---|---  
#### Associated Configurations
  * Consumer Configuration


#### Throws
  * KAFKA:RETRY_EXHAUSTED
  * KAFKA:ILLEGAL_STATE
  * KAFKA:TIMEOUT
  * KAFKA:SERIALIZATION_ERROR
  * KAFKA:INVALID_OFFSET
  * KAFKA:INVALID_INPUT
  * KAFKA:NOT_FOUND
  * KAFKA:CONNECTIVITY


### Seek
`<kafka:seek>`
Sets the current offset of the consumer for the given topic and partition to the provided offset value.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
Name of the topic on which the **Seek** operation is performed.  
Partition number that has its offset modified.  
Offset value to commit for the configured partition.  
**Operation Timeout** |  Timeout for the operation to start executing.  
**Operation Timeout Time Unit** |  Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS

|  Time unit for the **Operation Timeout** field.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Associated Configurations
  * Consumer Configuration


#### Throws
  * KAFKA:INVALID_TOPIC
  * KAFKA:RETRY_EXHAUSTED
  * KAFKA:TIMEOUT
  * KAFKA:INVALID_OFFSET
  * KAFKA:INVALID_INPUT
  * KAFKA:NOT_FOUND
  * KAFKA:CONNECTIVITY


## Producer Operation
  * 

### Bulk Publish
`<kafka:bulk-publish>`
Publishes a bulk of messages to the specified Kafka topic, optionally specifying the partition, key, and message content for it. This operation supports transactions.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
Array of Kafka Message |  List of messages to publish.  
**Transactional Action** |  Enumeration, one of:
  * ALWAYS_JOIN
  * JOIN_IF_POSSIBLE
  * NOT_SUPPORTED

|  Type of joining action that operations can take for transactions. |  `JOIN_IF_POSSIBLE`  
**Target Variable** |  Name of the variable that stores the operation’s output.  
**Target Value** |  Expression that evaluates the operation’s output. The outcome of the expression is stored in the **Target Variable** field.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Output
Array of Kafka Message Metadata  
---  
#### Associated Configurations
  * Producer Configuration


#### Throws
  * KAFKA:INVALID_TOPIC_PARTITION
  * KAFKA:RETRY_EXHAUSTED
  * KAFKA:TIMEOUT
  * KAFKA:AUTHENTICATION_ERROR
  * KAFKA:INPUT_TOO_LARGE
  * KAFKA:CONNECTIVITY


### Publish
`<kafka:publish>`
Publishes a message to the specified Kafka topic, optionally specifying the partition, key, and message content for it. This operation supports transactions.
Name | Type | Description | Default Value | Required  
---|---|---|---|---  
**Configuration** |  Name of the configuration to use.  
Topic to publish to.  
(Optional) Topic partition.  
(Optional) Key for the published message.  
(Optional) Message content of the message.  
(Optional) Headers for the message.  
**Transactional Action** |  Enumeration, one of:
  * ALWAYS_JOIN
  * JOIN_IF_POSSIBLE
  * NOT_SUPPORTED

|  Type of joining action that operations can take for transactions. |  `JOIN_IF_POSSIBLE`  
**Target Variable** |  Name of the variable that stores the operation’s output.  
**Target Value** |  Expression that evaluates the operation’s output. The outcome of the expression is stored in the **Target Variable** field.  
**Reconnection Strategy** | 
  *   * Reconnect Forever

|  Retry strategy in case of connectivity errors.  
#### Output
Kafka Message Metadata  
---  
#### Associated Configurations
  * Producer Configuration


#### Throws
  * KAFKA:INVALID_TOPIC_PARTITION
  * KAFKA:RETRY_EXHAUSTED
  * KAFKA:TIMEOUT
  * KAFKA:INPUT_TOO_LARGE
  * KAFKA:CONNECTIVITY


## Object Types
  * Consumer Context
  * Custom OCSP Responder
  * Expiration Policy
  *   * Kafka Message Attributes
  * Kafka Message Metadata
  * Kafka Record Attributes
  * Proxy Config Params
  *   * Reconnect Forever
  *   * Redelivery Policy
  * Repeatable File Store Stream
  * Repeatable In Memory Stream
  * Standard Revocation Check
  * Tcp Client Socket Params
  * Topic Partition
  * 

### Consumer Context
Type for the consumer context.
Field | Type | Description | Default Value | Required  
---|---|---|---|---  
**Consumer Commit Key** |  Consumer commit key.  
### CRL File
Specifies the location of the certification revocation list (CRL) file.
Field | Type | Description | Default Value | Required  
---|---|---|---|---  
Path to the CRL file.  
### Custom OCSP Responder
Configures a custom OCSP responder for certification revocation checks.
Field | Type | Description | Default Value | Required  
---|---|---|---|---  
URL of the OCSP responder.  
**Cert Alias** |  Alias of the signing certificate for the OCSP response. If specified, the alias must be in the truststore.  
### Expiration Policy
Configures an expiration policy strategy.
Field | Type | Description | Default Value | Required  
---|---|---|---|---  
**Max Idle Time** |  Configures the maximum amount of time that a dynamic configuration instance can remain idle before Mule considers it eligible for expiration.  
Enumeration, one of:
  * NANOSECONDS
  * MICROSECONDS
  * MILLISECONDS
  * SECONDS
  * MINUTES
  * HOURS