// This module is included in the following files:
//
// assembly-tuning-config.adoc

[id='con-consumer-config-properties-{context}']
= Kafka consumer configuration tuning

[role="_abstract"]
Use configuration properties to optimize the performance of Kafka consumers.
When tuning your consumers your primary concern will be ensuring that they cope efficiently with the amount of data ingested.
As with the producer tuning, be prepared to make incremental changes until the consumers operate as expected.

When tuning a consumer, consider the following aspects carefully, as they significantly impact its performance and behavior:

Scaling:: Consumer groups enable parallel processing of messages by distributing the load across multiple consumers, enhancing scalability and throughput. 
The number of topic partitions determines the maximum level of parallelism that you can achieve, as one partition can only be assigned to one consumer in a consumer group. 
Message ordering:: 
If absolute ordering within a topic is important, use a single-partition topic. 
A consumer observes messages in a single partition in the same order that they were committed to the broker, which means that Kafka only provides ordering guarantees for messages in a single partition. 
It is also possible to maintain message ordering for events specific to individual entities, such as users.
If a new entity is created, you can create a new topic dedicated to that entity.
You can use a unique ID, like a user ID, as the message key and route all messages with the same key to a single partition within the topic. 
Offset reset policy:: Setting the appropriate offset policy ensures that the consumer consumes messages from the desired starting point and handles message processing accordingly. 
The default Kafka reset value is `latest`, which starts at the end of the partition, and consequently means some messages might be missed, depending on the consumer's behavior and the state of the partition. 
Setting `auto.offset.reset` to `earliest` ensures that when connecting with a new `group.id`, all messages are retrieved from the beginning of the log.
Securing access:: Implement security measures for authentication, encryption, and authorization by setting up user accounts to xref:assembly-securing-access-{context}[manage secure access to Kafka].

== Basic consumer configuration

Connection and deserializer properties are required for every consumer.
Generally, it is good practice to add a client id for tracking.

In a consumer configuration, irrespective of any subsequent configuration:

* The consumer fetches from a given offset and consumes the messages in order, unless the offset is changed to skip or re-read messages.
* The broker does not know if the consumer processed the responses, even when committing offsets to Kafka, because the offsets might be sent to a different broker in the cluster.

.Basic consumer configuration properties
[source,env]
----
# ...
bootstrap.servers=localhost:9092 <1>
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer  <2>
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer  <3>
client.id=my-client <4>
group.id=my-group-id <5>
# ...
----
<1> (Required) Tells the consumer to connect to a Kafka cluster using a _host:port_ bootstrap server address for a Kafka broker.
The consumer uses the address to discover and connect to all brokers in the cluster.
Use a comma-separated list to specify two or three addresses in case a server is down, but it is not necessary to provide a list of all the brokers in the cluster.
If you are using a loadbalancer service to expose the Kafka cluster, you only need the address for the service because the availability is handled by the loadbalancer.
<2> (Required) Deserializer to transform the bytes fetched from the Kafka broker into message keys.
<3> (Required) Deserializer to transform the bytes fetched from the Kafka broker into message values.
<4> (Optional) The logical name for the client, which is used in logs and metrics to identify the source of a request. The id can also be used to throttle consumers based on processing time quotas.
<5> (Conditional) A group id is _required_ for a consumer to be able to join a consumer group.

== Scaling data consumption using consumer groups

Consumer groups share a typically large data stream generated by one or multiple producers from a given topic.
Consumers are grouped using a `group.id` property, allowing messages to be spread across the members.
One of the consumers in the group is elected leader and decides how the partitions are assigned to the consumers in the group.
Each partition can only be assigned to a single consumer.

If you do not already have as many consumers as partitions,
you can scale data consumption by adding more consumer instances with the same `group.id`.
Adding more consumers to a group than there are partitions will not help throughput,
but it does mean that there are consumers on standby should one stop functioning.
If you can meet throughput goals with fewer consumers, you save on resources.

Consumers within the same consumer group send offset commits and heartbeats to the same broker.
The consumer sends heartbeats to the Kafka broker to indicate its activity within the consumer group.
So the greater the number of consumers in the group, the higher the request load on the broker.

[source,env]
----
# ...
group.id=my-group-id <1>
# ...
----
<1> Add a consumer to a consumer group using a group id.

== Choosing the right partition assignment strategy

Select an appropriate partition assignment strategy, which determines how Kafka topic partitions are distributed among consumer instances in a group.

Partition strategies are supported by the following classes:

* `org.apache.kafka.clients.consumer.RangeAssignor`
* `org.apache.kafka.clients.consumer.RoundRobinAssignor`
* `org.apache.kafka.clients.consumer.StickyAssignor`
* `org.apache.kafka.clients.consumer.CooperativeStickyAssignor`

Specify a class using the `partition.assignment.strategy` consumer configuration property. 
The *range* assignment strategy assigns a range of partitions to each consumer, and is useful when you want to process related data together.

Alternatively, opt for a *round robin* assignment strategy for equal partition distribution among consumers, which is ideal for high-throughput scenarios requiring parallel processing.

For more stable partition assignments, consider the *sticky* and *cooperative sticky* strategies. 
Sticky strategies aim to maintain assigned partitions during rebalances, when possible. 
If a consumer was previously assigned certain partitions, the sticky strategies prioritize retaining those same partitions with the same consumer after a rebalance, while only revoking and reassigning the partitions that are actually moved to another consumer.
Leaving partition assignments in place reduces the overhead on partition movements.
The cooperative sticky strategy also supports cooperative rebalances, enabling uninterrupted consumption from partitions that are not reassigned.

If none of the available strategies suit your data, you can create a custom strategy tailored to your specific requirements.

== Message ordering guarantees

Kafka brokers receive fetch requests from consumers that ask the broker to send messages from a list of topics, partitions and offset positions.

A consumer observes messages in a single partition in the same order that they were committed to the broker,
which means that Kafka *only* provides ordering guarantees for messages in a single partition.
Conversely, if a consumer is consuming messages from multiple partitions, the order of messages in different partitions as observed by the consumer does not necessarily reflect the order in which they were sent.

If you want a strict ordering of messages from one topic, use one partition per consumer.

[id='con-consumer-config-properties-throughput-{context}']
== Optimizing consumers for throughput and latency

Control the number of messages returned when your client application calls `KafkaConsumer.poll()`.

Use the `fetch.max.wait.ms` and `fetch.min.bytes` properties to increase the minimum amount of data fetched by the consumer from the Kafka broker.
Time-based batching is configured using `fetch.max.wait.ms`, and size-based batching is configured using `fetch.min.bytes`.

If CPU utilization in the consumer or broker is high, it might be because there are too many requests from the consumer.
You can adjust `fetch.max.wait.ms` and `fetch.min.bytes` properties higher so that there are fewer requests and messages are delivered in bigger batches.
By adjusting higher, throughput is improved with some cost to latency.
You can also adjust higher if the amount of data being produced is low.

For example, if you set `fetch.max.wait.ms` to 500ms and `fetch.min.bytes` to 16384 bytes,
when Kafka receives a fetch request from the consumer it will respond when the first of either threshold is reached.

Conversely, you can adjust the `fetch.max.wait.ms` and `fetch.min.bytes` properties lower to improve end-to-end latency.

[source,env]
----
# ...
fetch.max.wait.ms=500 <1>
fetch.min.bytes=16384 <2>
# ...
----
<1> The maximum time in milliseconds the broker will wait before completing fetch requests.
The default is `500` milliseconds.
<2> If a minimum batch size in bytes is used, a request is sent when the minimum is reached, or messages have been queued for longer than `fetch.max.wait.ms` (whichever comes sooner).
Adding the delay allows batches to accumulate messages up to the batch size.

.Lowering latency by increasing the fetch request size

Use the `fetch.max.bytes` and `max.partition.fetch.bytes` properties to increase the maximum amount of data fetched by the consumer from the Kafka broker.

The `fetch.max.bytes` property sets a maximum limit in bytes on the amount of data fetched from the broker at one time.

The `max.partition.fetch.bytes` sets a maximum limit in bytes on how much data is returned for each partition,
which must always be larger than the number of bytes set in the broker or topic configuration for `max.message.bytes`.

The maximum amount of memory a client can consume is calculated approximately as:

[source,shell,subs="+quotes,attributes"]
----
_NUMBER-OF-BROKERS_ * fetch.max.bytes and _NUMBER-OF-PARTITIONS_ * max.partition.fetch.bytes
----

If memory usage can accommodate it, you can increase the values of these two properties.
By allowing more data in each request, latency is improved as there are fewer fetch requests.

[source,env]
----
# ...
fetch.max.bytes=52428800 <1>
max.partition.fetch.bytes=1048576 <2>
# ...
----
<1> The maximum amount of data in bytes returned for a fetch request.
<2> The maximum amount of data in bytes returned for each partition.

== Avoiding data loss or duplication when committing offsets

The Kafka _auto-commit mechanism_ allows a consumer to commit the offsets of messages automatically.
If enabled, the consumer will commit offsets received from polling the broker at 5000ms intervals.

The auto-commit mechanism is convenient, but it introduces a risk of data loss and duplication.
If a consumer has fetched and transformed a number of messages, but the system crashes with processed messages in the consumer buffer when performing an auto-commit, that data is lost.
If the system crashes after processing the messages, but before performing the auto-commit, the data is duplicated on another consumer instance after rebalancing.

Auto-committing can avoid data loss only when all messages are processed before the next poll to the broker,
or the consumer closes.

To minimize the likelihood of data loss or duplication, you can set `enable.auto.commit` to `false` and develop your client application to have more control over committing offsets.
Or you can use `auto.commit.interval.ms` to decrease the intervals between commits.

[source,env]
----
# ...
enable.auto.commit=false <1>
# ...
----
<1> Auto commit is set to false to provide more control over committing offsets.

By setting to `enable.auto.commit` to `false`, you can commit offsets after *all* processing has been performed and the message has been consumed.
For example, you can set up your application to call the Kafka `commitSync` and `commitAsync` commit APIs.

The `commitSync` API commits the offsets in a message batch returned from polling.
You call the API when you are finished processing all the messages in the batch.
If you use the `commitSync` API, the application will not poll for new messages until the last offset in the batch is committed.
If this negatively affects throughput, you can commit less frequently,
or you can use the `commitAsync` API.
The `commitAsync` API does not wait for the broker to respond to a commit request,
but risks creating more duplicates when rebalancing.
A common approach is to combine both commit APIs in an application, with the `commitSync` API used just before shutting the consumer down or rebalancing to make sure the final commit is successful.

=== Controlling transactional messages

Consider using transactional ids and enabling idempotence (`enable.idempotence=true`) on the producer side to guarantee exactly-once delivery.
On the consumer side, you can then use the `isolation.level` property to control how transactional messages are read by the consumer.

The `isolation.level` property has two valid values:

* `read_committed`
* `read_uncommitted` (default)

Use `read_committed` to ensure that only transactional messages that have been committed are read by the consumer.
However, this will cause an increase in end-to-end latency, because the consumer will not be able to return a message until the brokers have written the transaction markers that record the result of the transaction (_committed_ or _aborted_).

[source,env]
----
# ...
enable.auto.commit=false
isolation.level=read_committed <1>
# ...
----
<1> Set to `read_committed` so that only committed messages are read by the consumer.

== Recovering from failure to avoid data loss

In the event of failures within a consumer group, Kafka provides a rebalance protocol designed for effective detection and recovery.
To minimize the potential impact of these failures, one key strategy is to adjust the `max.poll.records` property to balance efficient processing with system stability. 
This property determines the maximum number of records a consumer can fetch in a single poll.
Fine-tuning `max.poll.records` helps to maintain a controlled consumption rate, preventing the consumer from overwhelming itself or the Kafka broker.

Additionally, Kafka offers advanced configuration properties like `session.timeout.ms` and `heartbeat.interval.ms`. 
These settings are typically reserved for more specialized use cases and may not require adjustment in standard scenarios.

The `session.timeout.ms` property specifies the maximum amount of time a consumer can go without sending a heartbeat to the Kafka broker to indicate it is active within the consumer group. 
If a consumer fails to send a heartbeat within the session timeout, it is considered inactive.
A consumer marked as inactive triggers a rebalancing of the partitions for the topic. 
Setting the `session.timeout.ms` property value too low can result in false-positive outcomes, while setting it too high can lead to delayed recovery from failures.

The `heartbeat.interval.ms` property determines how frequently a consumer sends heartbeats to the Kafka broker.    
A shorter interval between consecutive heartbeats allows for quicker detection of consumer failures. 
The heartbeat interval must be lower, usually by a third, than the session timeout. 
Decreasing the heartbeat interval reduces the chance of accidental rebalancing, but more frequent heartbeats increases the overhead on broker resources.

[source,env]
----
# ...
max.poll.records=100 # <1>
session.timeout.ms=30000 # <2>
heartbeat.interval.ms=5000 # <3>
# ...
----
<1> Set the number records returned to the consumer when calling the `poll()` method.
<2> Set the timeout for detecting client failure. 
If the broker configuration has a `group.min.session.timeout.ms` and `group.max.session.timeout.ms`, the session timeout value must be within that range.
<3> Adjust the heartbeat interval according to anticipated rebalances.

== Managing offset policy

Use the `auto.offset.reset` property to control how a consumer behaves when no offsets have been committed,
or a committed offset is no longer valid or deleted.

Suppose you deploy a consumer application for the first time, and it reads messages from an existing topic.
Because this is the first time the `group.id` is used, the `__consumer_offsets` topic does not contain any offset information for this application.
The new application can start processing all existing messages from the start of the log or only new messages.
The default reset value is `latest`, which starts at the end of the partition, and consequently means some messages are missed.
To avoid data loss, but increase the amount of processing, set `auto.offset.reset` to `earliest` to start at the beginning of the partition.

Also consider using the `earliest` option to avoid messages being lost when the offsets retention period (`offsets.retention.minutes`) configured for a broker has ended.
If a consumer group or standalone consumer is inactive and commits no offsets during the retention period, previously committed offsets are deleted from `__consumer_offsets`.

[source,env]
----
# ...
auto.offset.reset=earliest # <1>
# ...
----
<1> Set to `earliest` to return to the start of a partition and avoid data loss if offsets were not committed.

If the amount of data returned in a single fetch request is large,
a timeout might occur before the consumer has processed it.
In this case, you can lower `max.partition.fetch.bytes` or increase `session.timeout.ms`.

== Minimizing the impact of rebalances 

Rebalances in Kafka consumer groups can introduce latency and reduce throughput, impacting overall service performance. 
The rebalancing of a partition between active consumers in a group is the time it takes for the following to take place:

* Consumers to commit their offsets
* The new consumer group to be formed
* The group leader to assign partitions to group members
* The consumers in the group to receive their assignments and start fetching

Rebalances are triggered by changes in consumer health, network issues, configuration updates, and scaling events.
This process can increase service downtime, especially if it occurs frequently, such as during rolling restarts of consumers in a group.

To minimize the impact of rebalances, consider the following strategies and configurations:

Assess throughput and parallelism:: Assess the expected throughput (bytes and records per second) and parallelism (number of partitions) of the input topics against the number of consumers.
+
If adjustments are needed, start by setting up static membership, adopting a partition assignment strategy, and setting a limit on the number of records returned using the `max.poll.records` property. 
Add further configurations for timeouts and intervals, if required and with care, as these can introduce issues related to the handling of failures.

Use static membership:: Assign a unique identifier (`group.instance.id`) to each consumer instance. 
Static membership introduces persistence so static consumers retain partition assignments across restarts, reducing unnecessary rebalances. 

Adopt partition assignment strategies:: 
* Use appropriate partition assignment strategies to reduce the number of partitions that need to be reassigned during a rebalance, minimizing the impact on active consumers.
* The `org.apache.kafka.clients.consumer.CooperativeStickyAssignor` strategy is particularly effective, as it ensures minimal partition movement and better stability during rebalances.

Adjust record limits and poll intervals::
* Use the `max.poll.records` property to limit the number of records returned during each poll. 
Processing fewer messages more efficiently can prevent delays. 
* Use the `max.poll.interval.ms` property to prevent rebalances caused by prolonged processing tasks by setting the maximum interval between calls to the `poll()` method.
* Alternatively, consider pausing partitions to retrieve fewer records at a time.

Adjust session timeout and heartbeat intervals::
* Use the `session.timeout.ms` property to set a longer timeout to reduce rebalances caused by temporary network glitches or minor processing delays.
* Adjust the `heartbeat.interval.ms` property to balance failure detection checks with minimizing unnecessary rebalances.

Monitor consumer health:: Instability in consumer applications, such as frequent crashes, can trigger rebalances. 
Use Kafka consumer metrics to monitor such things as rebalance rates, session timouts, and failed fetch requests.

.Example configuration to minimize the impact of rebalances
[source,shell]
----
# ...
group.instance.id=<unique_id>
max.poll.interval.ms=300000
max.poll.records=500
session.timeout.ms=30000
heartbeat.interval.ms=5000
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
# ...
----

.Scaling strategies
To minimize the impact of rebalances during scaling of consumer groups, consider the following approaches:

Set a rebalance delay:: Use the `group.initial.rebalance.delay.ms` property in the Kafka configuration to delay the time it takes for consumers to join a new consumer group before performing a rebalance.
Introducing a delay helps avoid triggering several rebalances when starting multiple consumers near the same time. 
The appropriate delay depends on the orchestration used and might not be suitable in some circumstances. 

Avoid frequent scaling:: 
* Keep the number of consumers stable, scaling only when necessary and in controlled increments.
* Monitor system performance and adjust your scaling strategy as needed.
** Lag per partition should be constant and low.
** Records processed per second by consumers should match the records per second in the input topics.
* Use the Kafka Exporter to check for consumer lag and determine if scaling is required.

Implement dynamic scaling policies::
* If using dynamic or event-driven tools for scaling of consumer applications, set lag thresholds based on the backlog of messages.
* Define maximum and minimum replica counts for consumer groups.
* Set periods between scaling events to prevent rapid scaling.

NOTE: In cases where lengthy message processing is unavoidable, consider pausing and resuming partitions as needed.
If you pause all partitions, `poll()` returns no records, allowing you to keep calling it without overwhelming the consumers. 
Alternatively, you can offload the processing tasks to a pool of worker threads. 
This helps prevents delays and potential rebalances.