Managing Consumer Groups

1. Understanding Group Coordinator

Property	Description	Default Value
group.coordinator	Identifier for the group coordinator in the Kafka cluster	auto
group.coordinator.heartbeat.interval.ms	Interval between heartbeats sent by the consumer to the group coordinator	3000
group.coordinator.session.timeout.ms	Timeout for consumer session with the group coordinator	45000

Example: Consumer Configuration with Group Coordinator

{"bootstrap.servers": "localhost:9092", "group.id": "my-consumer-group", "enable.auto.commit": true, "auto.offset.reset": "earliest", "group.coordinator.session.timeout.ms": 60000}

Note: The group coordinator is responsible for managing consumer group membership and rebalances. It ensures that consumers within the same group are assigned partitions in a balanced way.

2. Understanding Rebalancing

Note: Rebalancing is a critical process in Kafka consumer groups that ensures load distribution and fault tolerance.

Aspect	Description	Details
Trigger Conditions	When rebalancing occurs	New consumer joins the group Consumer leaves the group Topic subscription changes Partition count increases
Rebalance Phases	Stages of the rebalancing process	JOIN_PHASE: Consumers join the group SYNC_PHASE: Assignment is synchronized COMPLETE_PHASE: Assignment is finalized
Coordinator Role	Who manages rebalancing	The group coordinator (usually a broker) assigns partitions to consumers

Example: Consumer Joining a Group

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

Rebalancing Impact

Temporarily halts message consumption during rebalancing
Can cause performance degradation if frequent
Ensures even distribution of partitions across consumers

Warning: Frequent rebalancing can lead to consumer lag and reduced throughput. Monitor consumer group metrics to detect issues.

3. Listing Consumer Groups

Command	Description	Options	Example
kafka-consumer-groups.sh	List all consumer groups	`--bootstrap-server` `--list`	`kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list`
kafka-consumer-groups.sh	Describe consumer group details	`--bootstrap-server` `--describe` `--group`	`kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group`

4. Describing Consumer Groups

Property	Description	Default Value
group.id	Unique identifier for the consumer group	None
session.timeout.ms	Timeout for consumer session	45000
heartbeat.interval.ms	Interval between heartbeats	3000
auto.offset.reset	What to do when no offset is found	latest

Example: Describe Consumer Group

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

Note: The --describe option provides detailed information about the consumer group's state, including member details and partition assignments.

5. Viewing Consumer Lag

Property	Description	Default Value
kafka.consumer.lag	Indicates the number of messages a consumer is behind the latest offset in a partition	0
kafka.consumer.lag.ms	Time in milliseconds the consumer is lagging behind	0

Example: Using Kafka Consumer Lag Metrics

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

Note: Consumer lag can be monitored using tools like Kafka Manager or Confluent Control Center.

Key Metrics for Consumer Lag

lag: Number of messages behind
time lag: Time in milliseconds the consumer is behind

6. Resetting Offsets

Example: Reset Consumer Group Offsets to Beginning

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --reset-offsets --to-earliest --topic my-topic

Option	Values	Description
--group	`string`	Consumer group ID to reset offsets for
--topic	`string`	Topic to reset offsets for (optional, defaults to all topics)
--to-earliest	`flag`	Reset to earliest offset available
--to-latest	`flag`	Reset to latest offset available
--to-datetime	`datetime`	Reset to offset at specific datetime (format: 2024-01-01T00:00:00Z)
--to-offset	`number`	Reset to specific offset number
--dry-run	`flag`	Preview changes without applying them

Note: Use --dry-run to preview changes before applying them.

Warning: Resetting offsets can cause data loss or reprocessing of messages.

Example: Reset Offsets to Specific Datetime

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --reset-offsets --to-datetime 2024-01-01T00:00:00Z --topic my-topic

7. Deleting Consumer Groups

Property	Description	Default Value
group.id	Unique identifier for the consumer group	None
bootstrap.servers	Initial list of brokers to connect to	None

Example: Delete Consumer Group via CLI

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --delete

Note: The consumer group must be inactive (no active consumers) to be deleted.

Warning: Deleting a consumer group will remove all committed offsets for that group.

8. Configuring Session Timeout

Property	Default Value	Description
session.timeout.ms	`45000`	Timeout for consumer session. If the consumer does not send a heartbeat within this time, it is considered dead and a rebalance occurs.
heartbeat.interval.ms	`3000`	Interval between heartbeats sent to the broker. Must be less than session.timeout.ms.

Example: Consumer Configuration with Session Timeout

{
  "bootstrap.servers": "localhost:9092",
  "group.id": "my-consumer-group",
  "session.timeout.ms": 60000,
  "heartbeat.interval.ms": 5000
}

Note: The session.timeout.ms should be set higher than the maximum time a consumer might take to process a batch of messages. If it's too low, consumers may be marked as dead unnecessarily.

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 due to delayed heartbeats.

10. Configuring Max Poll Interval

Property	Default	Description
max.poll.interval.ms	`300000`	Maximum delay between invocations of poll() before the consumer is considered dead.

Example: Setting max.poll.interval.ms

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

Note: If your consumer takes longer than this interval to process messages, it will be considered dead and a rebalance will occur.