Configuring Consumer Settings

1. Setting Fetch Size

Property	Default	Description
fetch.min.bytes	`1`	Minimum amount of data the server should return for a fetch request. Helps reduce network overhead by batching small requests.
fetch.max.bytes	`52428800`	Maximum amount of data the server should return for a fetch request. Limits memory usage on consumer side.
fetch.max.wait.ms	`500`	Maximum time the consumer will wait for new messages before returning an empty response. Affects latency and throughput trade-offs.

Example: Configuring Fetch Size

{
  "fetch.min.bytes": 1024,
  "fetch.max.bytes": 1048576,
  "fetch.max.wait.ms": 1000
}

2. Configuring Fetch Wait Time

Property	Default Value	Description
fetch.wait.max.ms	`500`	Maximum time in milliseconds the consumer will wait for new messages before returning.
fetch.min.bytes	`1`	Minimum number of bytes the consumer should fetch in a single request.
fetch.max.bytes	`52428800`	Maximum number of bytes the consumer should fetch in a single request.

Example: Setting Fetch Wait Time

{
  "fetch.wait.max.ms": 1000,
  "fetch.min.bytes": 1024,
  "fetch.max.bytes": 1048576
}

3. Setting Max Poll Records

Property	Default	Description
max.poll.records	`500`	Maximum number of records returned in a single call to `poll()`.

Example: Configuring max.poll.records

{
  maxPollRecords: 1000
}

Note: Setting this value too high can lead to longer poll intervals and potential timeouts if the consumer is not keeping up with the data rate.

4. Setting Max Partition Fetch Bytes

Property	Default	Description
max.partition.fetch.bytes	`1048576`	Maximum number of bytes per partition the consumer will fetch in a single request.

Example: Configuring max.partition.fetch.bytes

{
  "max.partition.fetch.bytes": 5242880
}

Note: Increasing this value allows consumers to fetch larger messages but may increase memory usage and network overhead.

5. Configuring Partition Assignment Strategy

Property	Values	Description
partition.assignment.strategy	`org.apache.kafka.clients.consumer.RangeAssignor` `org.apache.kafka.clients.consumer.RoundRobinAssignor` `org.apache.kafka.clients.consumer.StickyAssignor`	Strategy used to assign partitions to consumers in a consumer group. Default is `RangeAssignor`.

Example: Setting Partition Assignment Strategy

{
  "partition.assignment.strategy": "org.apache.kafka.clients.consumer.StickyAssignor"
}

Assignment Strategy Comparison

RangeAssignor: Assigns contiguous ranges of partitions to consumers. Good for balanced loads but can cause uneven distribution with varying partition counts.
RoundRobinAssignor: Distributes partitions in a round-robin fashion across consumers. Provides better distribution than RangeAssignor but can lead to suboptimal performance in some cases.
StickyAssignor: Ensures minimal partition movement when consumers join or leave. Maintains stable assignments and is the default in newer Kafka versions.

6. Setting Auto Offset Reset

Property	Values	Description
auto.offset.reset	`earliest`, `latest`, `none`	Determines the offset to use when no initial offset is stored for a consumer group.

Example: Configuring auto.offset.reset

{
  "auto.offset.reset": "earliest"
}

Note: earliest will reset to the beginning of the topic, while latest will start from the most recent message. none throws an exception if no offset is found.

7. Configuring Isolation Level

Property	Values	Description
isolation.level	`read_committed`, `read_uncommitted`	Determines the transaction isolation level for reading messages. `read_committed` ensures only committed transactions are read, while `read_uncommitted` allows reading uncommitted data.

Example: Setting Isolation Level

{
  "isolation.level": "read_committed"
}

Note: read_committed is the default and recommended setting for most use cases to ensure data consistency.

8. Setting Session Timeout

Property	Default Value	Description
session.timeout.ms	`45000`	Timeout for consumer session. If the consumer doesn't send a heartbeat within this time, it is considered dead and a rebalance occurs.
heartbeat.interval.ms	`3000`	Interval between heartbeats to the consumer group coordinator. Should be less than `session.timeout.ms`.

Example: Configuring Session Timeout

{
  "session.timeout.ms": 60000,
  "heartbeat.interval.ms": 5000
}

Note: Setting session.timeout.ms too low can cause unnecessary rebalances during temporary network issues or high load. Setting it too high increases the time to detect dead consumers.

9. Setting Heartbeat Interval

Property	Default Value	Description
heartbeat.interval.ms	`3000`	Frequency in milliseconds at which the consumer sends heartbeat messages to the coordinator.

Example: Configuring Heartbeat Interval

{
  "heartbeat.interval.ms": 5000
}

Note: The heartbeat interval should be less than the session.timeout.ms to avoid session expiration.

10. Configuring Max Poll Interval

Property	Default Value	Description
max.poll.interval.ms	`300000`	The maximum delay between invocations of poll() when using `Consumer` in a loop. If the consumer takes longer than this interval to process messages, it will be considered as failed and a rebalance will occur.

Example: Setting max.poll.interval.ms

{
  "max.poll.interval.ms": 600000
}

Note: Increasing this value gives consumers more time to process messages but increases the risk of rebalances due to timeouts.

11. Configuring Check CRCs

Property	Default Value	Description
checksum	`false`	Whether to validate message checksums when consuming messages.
checksum.algorithm	`CRC32C`	Algorithm used to compute the checksum. Supported values: `CRC32C`, `CRC32`.

Example: Enabling CRC Checksum Validation

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("checksum", "true");
props.put("checksum.algorithm", "CRC32C");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

Note: Enabling checksum validation adds overhead to message consumption but improves data integrity.

12. Setting Client ID

Property	Default	Description
client.id	`""`	Client identifier used in metrics and logs. Helps distinguish between different consumer instances.

Example: Setting Client ID

{
  "client.id": "my-consumer-1"
}