Consumers & Replication

Solution and explanation

What it is

A consumer group is a set of consumers sharing the same group.id that **cooperate to read a topic**. Kafka assigns each partition to exactly one consumer in the group, so the work is divided and each record is processed once by the group.

Topic payments: P0 P1 P2 P3 P4 P5
Group billing-service (3 consumers):
  consumer A ← P0, P1
  consumer B ← P2, P3
  consumer C ← P4, P5

Scaling rules

• Adding consumers (up to the number of partitions) increases parallelism.
• **More consumers than partitions:** extra consumers stay idle (useful only as hot standby).
• If a consumer dies or a new one joins, the group **rebalances** and partitions are reassigned.

Group coordination

One broker acts as the **group coordinator**. Consumers send heartbeats to it; if heartbeats stop (session.timeout.ms) or a consumer does not call poll in time (max.poll.interval.ms), it is removed and its partitions go to others. The group's committed offsets are stored in __consumer_offsets.

Groups as "subscribers"

Each group gets the full stream independently. A queue-style service uses one group with many members; pub/sub is achieved by having several groups.

Inspecting a group

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing-service
# TOPIC     PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# payments  0          15020           15030           10   consumer-A

Summary

A consumer group is a team of consumers that split a topic's partitions between them, giving parallel, load-balanced processing with progress tracked per group.

Solution and explanation

Answer: yes, and this is a core strength

Any number of consumer groups can read the same topic **independently**. Each group has its own committed offsets, so one group being slow, stopped or replaying history does not affect the others. Records are not removed when read.

Topic orders.placed
  ├─ group inventory-service     (reserves stock)          offset 98,200
  ├─ group payment-service       (charges the customer)    offset 98,215
  ├─ group notification-service  (sends SMS/email)         offset 97,900
  └─ group analytics-loader      (loads to warehouse)      offset 40,000 (catching up)

Why this matters

• **Fan-out without coupling:** the order service publishes once; adding a new consumer (e.g. fraud detection) needs no change in the producer.
• **Independent speed:** analytics can process in bulk every hour while payments react in milliseconds.
• **Replay per group:** a new team can start with auto.offset.reset=earliest and process the full retained history.
• **Different technologies:** a Java service, a Python ML job and a Kafka Connect sink can all read the same topic.

Cost

Each group reads the data separately, so network egress from brokers grows with the number of groups. Popular data is usually served from the OS page cache, so disk reads are not repeated.

Contrast with traditional queues

In RabbitMQ or SQS, a message is removed after one consumer acknowledges it. Delivering one message to several services needs a fan-out exchange or several queues. Kafka gets this naturally with groups.

Summary

Yes: every consumer group gets the entire topic and tracks its own offsets, enabling fan-out, independent pacing and replay without affecting other groups.

Solution and explanation

Answer: no (at the same time)

Within one consumer group, each partition is assigned to **exactly one** consumer at a time. Two consumers in the same group never read the same partition simultaneously.

Why Kafka enforces this

1. **Ordering:** a partition is ordered; if two consumers processed it in parallel, record 11 could finish before record 10.
2. **Simple progress tracking:** one committed offset per partition per group. Two readers would need complex coordination of which records each processed.
3. **No duplicate work** within the group.

What happens with more consumers than partitions

Topic with 3 partitions, group of 5 consumers:
  C1 ← P0   C2 ← P1   C3 ← P2   C4 idle   C5 idle

Idle consumers take over during a rebalance if an active one fails.

Getting more parallelism than partitions

• **Add partitions** (planned carefully because key mapping changes).
• **Process records concurrently inside one consumer** while keeping per-key order, e.g. Confluent's **Parallel Consumer** library, or hand records to a thread pool keyed by record key, and commit only offsets whose records are all finished.

// Spring Kafka: concurrency creates several consumers, still max one per partition
@KafkaListener(topics = "payments", groupId = "billing", concurrency = "6")

• Kafka 4.0 adds **share groups** (KIP-932, "queues for Kafka"), where several consumers can share a partition with per-record acknowledgement, giving queue semantics without ordering guarantees.

Different groups

Consumers in **different** groups can of course read the same partition, each with its own offset.

Summary

In a classic consumer group a partition has one owner at a time, preserving order and simple offset tracking; scale by adding partitions or by in-consumer parallelism.

Solution and explanation

Answer: yes

A consumer can be assigned many partitions, from one topic or from several topics. When a group has fewer consumers than partitions, partitions are divided among them.

Topic orders with 6 partitions, group of 2 consumers:
  consumer A ← P0, P1, P2
  consumer B ← P3, P4, P5

A single consumer in a group reads all 6 partitions.

How it works

poll() fetches from all assigned partitions and returns a batch that can contain records from several partitions. Order is preserved **within each partition** but records from different partitions are interleaved.

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (TopicPartition tp : records.partitions()) {
    List<ConsumerRecord<String, String>> partRecords = records.records(tp);
    for (ConsumerRecord<String, String> r : partRecords) process(r);
    long last = partRecords.get(partRecords.size() - 1).offset();
    consumer.commitSync(Map.of(tp, new OffsetAndMetadata(last + 1)));   // commit per partition
}

Assignment strategies

Configured with partition.assignment.strategy:

• **RangeAssignor:** contiguous ranges per topic (can be uneven across topics).
• **RoundRobinAssignor:** spreads partitions evenly.
• **StickyAssignor:** even and keeps previous assignments where possible.
• **CooperativeStickyAssignor:** sticky plus incremental (cooperative) rebalancing, so consumers keep most partitions during a rebalance. Recommended default for new apps. Kafka 4.0's new consumer protocol (KIP-848) moves assignment to the broker.

Manual assignment

consumer.assign(List.of(new TopicPartition("orders", 2))) reads specific partitions without group management (no rebalancing); useful for tools and replays.

Summary

Yes: a consumer can own multiple partitions; it receives interleaved batches while order is kept per partition, and assignors decide how partitions are spread.

Solution and explanation

What it is

Replication keeps **copies of each partition on several brokers** so data survives broker failures. The number of copies is the **replication factor** (commonly 3). One copy is the **leader**; the others are **followers** that continuously fetch new records from the leader.

Partition payments-0, replication factor 3
  Broker 1: leader     [0..1050]
  Broker 2: follower   [0..1050]   in sync
  Broker 3: follower   [0..1048]   catching up

How it works

1. The producer writes to the leader.
2. Followers send fetch requests to the leader and append the same records to their logs.
3. Followers that are up to date are in the **ISR** (in-sync replica set).
4. The **high watermark** advances when all ISR members have a record; consumers only read up to the high watermark, so they never see data that could disappear after a failover.
5. With acks=all, the producer gets success only after all ISR replicas have the record.

Durability settings that work together

replication.factor=3          # three copies
min.insync.replicas=2         # at least two copies must confirm a write
acks=all                      # producer waits for the ISR
unclean.leader.election.enable=false   # never elect a stale replica

This tolerates one broker failure with no data loss and continued writes.

Placement

Replicas are placed on different brokers, and with broker.rack set, on different racks or availability zones, so a whole rack or AZ failure is survivable.

Summary

Replication copies every partition to several brokers (leader plus followers); with acks=all and min.insync.replicas, committed data survives broker failures.

Solution and explanation

What it is

For every partition, one replica is the **leader**. The leader handles all **writes** from producers and, by default, all **reads** from consumers. Followers only copy data from it.

Why one leader

A single writer per partition gives a strict order of records and simple consistency: followers replicate exactly the leader's log.

Leader distribution

Leaders of different partitions are spread across brokers, so each broker is the leader for some partitions and a follower for others. This shares the load. The first replica in a partition's replica list is its **preferred leader**; Kafka periodically moves leadership back to preferred leaders (auto.leader.rebalance.enable=true) so load stays balanced after failures.

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments
# Topic: payments  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3
# Topic: payments  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1

Leader election

If the leader's broker fails, the **controller** picks a new leader from the ISR and notifies brokers; clients refresh metadata and switch. With unclean.leader.election.enable=false, only in-sync replicas can become leader, preventing data loss.

Reading from followers

Since Kafka 2.4, consumers can **fetch from the closest replica** (KIP-392) using client.rack and replica.selector.class=RackAwareReplicaSelector, which reduces cross-AZ network cost in the cloud.

Summary

The leader is the replica that serves writes (and normally reads) for a partition; leaders are spread over brokers and re-elected from the ISR by the controller on failure.

Solution and explanation

What it is

**ISR (In-Sync Replicas)** is the set of replicas of a partition, including the leader, that are fully caught up with the leader's log. Only ISR members count when a write needs acknowledgement (acks=all), and only they can become leader (with clean leader election).

Staying in the ISR

A follower stays in the ISR as long as it has fetched up to the leader's log end within replica.lag.time.max.ms (default 30 seconds). If it falls behind longer (slow disk, network trouble, GC pause, broker down), the leader removes it from the ISR. When it catches up again, it is added back.

Replicas: 1,2,3   ISR: 1,2,3   healthy
Replicas: 1,2,3   ISR: 1,3     broker 2 lagging → partition is under-replicated

Why it matters

• **Durability:** with acks=all, a write is committed once all ISR members have it. Combined with min.insync.replicas=2, at least two copies exist before the producer is told "success".
• **Availability trade-off:** if ISR shrinks below min.insync.replicas, producers using acks=all get NotEnoughReplicasException (writes are rejected rather than accepted with too few copies). Reads continue.
• **Safe failover:** a new leader is chosen from ISR, so no committed data is lost.

Monitoring

• UnderReplicatedPartitions > 0: some partitions have ISR smaller than the replica list.
• UnderMinIsrPartitionCount > 0: writes with acks=all are failing for those partitions.
• Frequent ISR shrink/expand events point to overloaded brokers or network problems.

Summary

The ISR is the set of replicas fully caught up with the leader; acknowledgements and leader elections rely on it, and its size versus min.insync.replicas decides whether writes are accepted.

Solution and explanation

What it is

The replication factor (RF) is the **number of copies** of each partition in the cluster, including the leader. RF=3 means one leader and two followers on three different brokers.

kafka-topics.sh --create --topic payments --partitions 12 --replication-factor 3 \
  --bootstrap-server localhost:9092

Default for auto-created topics: default.replication.factor (broker setting).

Choosing a value

Constraints and costs

• RF cannot exceed the number of brokers.
• Storage and replication network traffic multiply by RF: 1 TB of data at RF=3 uses 3 TB of disk.
• Write latency with acks=all depends on the slowest ISR member.

The usual production trio

replication.factor=3
min.insync.replicas=2
acks=all

Survives one broker outage with no data loss and writes still accepted; with two brokers down, writes stop (safety first) but data is preserved.

Changing RF later

Use partition reassignment with a JSON plan that lists the new replicas (kafka-reassign-partitions.sh --execute). Internal topics (__consumer_offsets, __transaction_state) should also use RF=3 (offsets.topic.replication.factor=3).

Summary

Replication factor is the number of copies of each partition; RF=3 with min.insync.replicas=2 and acks=all is the standard balance of durability, availability and cost.

Solution and explanation

Sequence of events

1. **Detection:** the controller notices the broker is gone (it stops sending heartbeats in KRaft, or its ZooKeeper session expires).
2. **Leader election:** for every partition that broker led, the controller picks a new leader from the remaining **ISR** members.
3. **Metadata update:** the new leadership and ISR are written to the metadata log and pushed to all brokers.
4. **Clients recover:** producers and consumers get NOT_LEADER_OR_FOLLOWER or connection errors, refresh metadata and reconnect to the new leaders. Producers retry automatically (retries, delivery.timeout.ms).
5. **Followers re-point:** remaining followers fetch from the new leader; the old broker's replicas are removed from the ISR.
6. **Return:** when the failed broker restarts, it truncates any uncommitted records beyond the high watermark, catches up as a follower, rejoins the ISR, and later may become leader again through preferred-leader rebalancing.

Data safety

• With acks=all, min.insync.replicas=2 and clean leader election, every acknowledged record exists on the new leader: **no data loss**.
• With acks=1, a record acknowledged by the old leader but not yet copied is **lost**.
• With unclean.leader.election.enable=true, an out-of-sync replica can become leader if no ISR member is alive: availability is restored but data can be lost.

Impact

Failover usually takes from under a second to a few seconds (faster with KRaft). During that time, writes to affected partitions are retried or delayed, and consumers pause for those partitions. Idempotent producers prevent duplicates caused by these retries.

Summary

The controller elects a new leader from the ISR, clients refresh metadata and retry, and with acks=all plus min.insync.replicas no acknowledged data is lost.

Solution and explanation

What it is

Retention defines **how long Kafka keeps records**, independent of whether anyone consumed them. Kafka deletes old data per **segment** (a file of records) once the whole segment is older than the retention limit or the partition exceeds its size limit.

Settings

kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics \
  --entity-name clickstream --add-config retention.ms=86400000     # 1 day

How deletion actually works

Only **closed** segments are deleted. If segment.ms is 7 days and retention.ms is 1 day, data may stay much longer than a day because the active segment has not rolled yet. Low-volume topics with short retention need a smaller segment.ms.

Choosing retention

• Clickstream or logs: hours to a few days.
• Business events used for replay or onboarding new consumers: 7–30 days or more.
• Event sourcing/audit: unlimited (retention.ms=-1), ideally with **tiered storage** (Kafka 3.6+), which moves old segments to S3-like object storage cheaply.
• Latest-state topics (customer profile): **compaction** instead of time retention.

Consumer impact

If a consumer is offline longer than retention, its committed offset points to deleted data; it then follows auto.offset.reset (and may skip records). Monitor lag against retention.

Summary

Retention removes old segments based on time or size, regardless of consumption; choose it from replay needs and storage cost, and use compaction or tiered storage for long-lived data.