Consumer Operations

Solution and explanation

What happens

1. The consumer stops sending heartbeats; after session.timeout.ms (45 s default) the group coordinator removes it (or immediately if it closed cleanly).
2. A **rebalance** assigns its partitions to other consumers.
3. The new owner starts from the **last committed offset**.
4. Records that were processed after that commit are **processed again**.

Committed offset: 500
Processed 500–549, crashed before committing 550
New consumer reads from 500 → 500–549 processed twice

No records are lost (they are still in the log), but there are **duplicates**: this is at-least-once delivery.

Consequences

• Duplicate emails or SMS, double inventory reservations, or double payment attempts if the consumer is not idempotent.
• The processing time of the reprocessed range adds lag.

Making reprocessing safe

1. **Idempotent processing:** upserts, unique constraints, "processed event id" tables:

INSERT INTO payments (payment_id, amount, status) VALUES (?, ?, 'CAPTURED')
ON DUPLICATE KEY UPDATE status = status;

1. **Commit more often** (per record or small batches) to shrink the reprocessing window, at a performance cost.
2. **Transactions** for Kafka-to-Kafka pipelines: output records and offset commit are atomic, so a crash rolls both back.
3. **Store offsets with results** in the same database transaction.
4. **Idempotency keys** when calling external APIs (payment gateways accept them).

Summary

After a crash, the partition's new owner resumes from the last committed offset, so records processed after that commit are redelivered; design consumers to be idempotent.

Solution and explanation

Definition

Every record is processed **one or more times**: none are lost, but duplicates are possible. It is Kafka's most common and practical guarantee.

How it is achieved

• **Producer:** acks=all with retries, so records are not lost (retries can duplicate, unless idempotence is on).
• **Consumer:** commit offsets **after** processing. A crash after processing but before commit causes reprocessing.

for (ConsumerRecord<String, Event> r : consumer.poll(timeout)) {
    process(r);            // 1. do the work
}
consumer.commitSync();     // 2. then record progress

Where duplicates come from

• Consumer crash or rebalance before commit.
• Producer retries without idempotence.
• Commit failures.

Handling duplicates

• **Idempotent writes:** INSERT ... ON CONFLICT DO NOTHING, upsert by business id.
• **Deduplication table** keyed by event id with a retention period.
• **Natural idempotency:** "set status to SHIPPED" is safe to repeat; "add ₹100 to balance" is not, so model it as "apply transaction T123 once".
• **Version checks:** ignore events with a version older than the stored one.

Why it is preferred

Losing a payment or order event is usually far worse than processing it twice and ignoring the duplicate. At-least-once plus idempotent consumers achieves "effectively exactly once" outcomes in databases and external systems, where Kafka transactions cannot reach.

Summary

At-least-once means no loss but possible duplicates, achieved by committing after processing; pair it with idempotent consumers for correct results.

Solution and explanation

Definition

Every record is processed **zero or one time**: never duplicated, but some may be lost.

How it happens

• **Consumer:** commit offsets **before** processing. If it crashes after committing but before finishing, those records are skipped.

ConsumerRecords<String, Metric> recs = consumer.poll(timeout);
consumer.commitSync();        // progress recorded first
recs.forEach(this::process);  // crash here → records never processed

• **Producer:** acks=0, or no retries, so failed sends are not repeated.
• **Auto commit with asynchronous processing:** offsets get committed while work is still pending.

When it is acceptable

• High-volume telemetry, metrics or debug logs where a gap is harmless and duplicates would distort counts.
• Real-time location or price ticks where only the latest value matters and stale data is useless.
• Notifications where sending twice is worse than not sending (rare; usually you want at-least-once with dedupe).

When it is wrong

Payments, orders, ledger entries, job applications, audit trails: any loss is a business or compliance problem.

Comparison

Summary

At-most-once commits before processing (or never retries), so records may be lost but never duplicated; use it only for disposable data.

Solution and explanation

Definition

Each record's effect is applied **exactly once**, even with failures and retries. In Kafka this is called **EOS (exactly-once semantics)** and applies to **read-process-write pipelines inside Kafka**: consume from topics, process, produce to topics.

Building blocks

1. **Idempotent producer:** no duplicates from retries (PID + sequence numbers).
2. **Transactions:** a producer with a transactional.id writes to several partitions and commits consumer offsets **atomically**:

producer.initTransactions();
while (running) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(500));
    producer.beginTransaction();
    try {
        for (var r : records) producer.send(new ProducerRecord<>("invoices", r.key(), toInvoice(r.value())));
        producer.sendOffsetsToTransaction(currentOffsets(records), consumer.groupMetadata());
        producer.commitTransaction();
    } catch (ProducerFencedException e) {
        producer.close(); throw e;               // another instance took over
    } catch (KafkaException e) {
        producer.abortTransaction();             // outputs and offsets both rolled back
    }
}

1. **isolation.level=read_committed** on downstream consumers so they never see aborted records.

Kafka Streams enables all of this with one setting: processing.guarantee=exactly_once_v2.

The boundary

EOS covers Kafka-to-Kafka. When the output is a database, email or payment API, Kafka transactions cannot roll those back. Then use:

• at-least-once delivery + **idempotent sinks** (upsert by key, unique event id),
• storing offsets in the same database transaction as the results,
• the **outbox pattern** for producing events from database changes,
• idempotency keys for external APIs.

Cost

Transactions add latency (commit markers, coordinator round trips) and some throughput overhead; consumers with read_committed wait until transactions complete.

Summary

Exactly-once in Kafka combines idempotent producers, transactions that commit outputs and offsets atomically, and read_committed consumers; outside Kafka you achieve it with idempotent sinks.

Solution and explanation

Definition

Consumer lag is **how far behind a consumer group is** in a partition:

lag = log end offset (latest record) − committed offset of the group

Total lag for a topic is the sum over its partitions. Lag can also be measured in time ("the oldest unprocessed record is 4 minutes old"), which is often more meaningful to the business.

Checking lag

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group payment-service
# TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# payments   0          1_204_550       1_204_600       50
# payments   1          1_198_020       1_260_020       62000   ← problem

Tools: Burrow, Kafka Exporter + Prometheus/Grafana, Confluent Control Center, Kafka UI, consumer metrics records-lag-max.

Why it matters

• Business latency: an order event lagging 10 minutes means a confirmation email 10 minutes late, or fraud checks happening after the money moved.
• Data loss risk: if lag exceeds retention, unconsumed records are deleted.
• Growing lag means consumers cannot keep up with producers.

Reading lag correctly

• Small, stable lag is normal (records arrive continuously).
• Lag concentrated in one partition indicates a hot key or a stuck consumer.
• Steadily increasing lag across all partitions means throughput is insufficient.

Alerting

Alert on **sustained growth** or on time lag above an SLO (e.g. > 60 s for payments), not on a fixed number of messages, because volumes vary.

Summary

Consumer lag is the gap between the newest offset and the group's committed offset; monitor it per partition and in time to catch slow or stuck consumers.

Solution and explanation

Common causes

1. **Slow processing per record:** synchronous calls to slow databases or external APIs (payment gateway, SMS provider), heavy JSON parsing, no batching of database writes.
2. **Not enough parallelism:** fewer consumers than partitions, or too few partitions for the load.
3. **Traffic spikes:** a sale, a batch job or a replay suddenly produces far more records.
4. **Hot partitions:** one key (or a bad partitioner) concentrates most records in one partition handled by one consumer.
5. **Frequent rebalances:** consumers exceeding max.poll.interval.ms are kicked out, partitions move, work restarts; repeated rebalances can stop progress.
6. **Errors and retries:** a poison message retried forever blocks the partition.
7. **Downstream outages:** the database is down or throttling, so the consumer waits.
8. **Resource limits:** CPU throttling in containers, long GC pauses, network bandwidth, too small fetch.max.bytes.
9. **Broker problems:** under-replicated partitions, slow disks, or leaders concentrated on one broker.

Diagnosing

• Is lag on **all** partitions (capacity issue) or **one** (hot key or stuck consumer)?
• Is the group stable? Check rebalance logs and max.poll.interval.ms errors.
• Measure processing time per record and per external call (tracing, metrics).
• Compare producer rate with consumer rate.
• Check the consumer's error logs for repeated failures on the same offset.

Summary

High lag comes from slow processing, too little parallelism, spikes, hot partitions, rebalances, poison messages, downstream outages or resource limits; find out whether it is global or per-partition first.

Solution and explanation

Scale out

• **Add consumers** to the group, up to the number of partitions.
• **Add partitions** if consumers are already equal to partitions (plan for key remapping).
• **Process in parallel inside a consumer** while keeping per-key order (Confluent Parallel Consumer, or a thread per key hash).

Make each record cheaper

• **Batch downstream writes:** insert 500 rows in one statement instead of 500 statements.

@KafkaListener(topics = "clicks", batch = "true")
public void onBatch(List<ClickEvent> events) { clickRepository.saveAllBatch(events); }

• Cache reference data instead of querying the database per record.
• Use async, non-blocking clients for external APIs, with bounded concurrency.
• Use efficient serialisation (Avro/Protobuf) instead of large JSON.

Tune the consumer

max.poll.records=500          # more records per poll if processing is fast
fetch.min.bytes=65536         # fewer, larger fetches
fetch.max.wait.ms=100
max.partition.fetch.bytes=4194304

Keep max.poll.interval.ms above the worst-case batch processing time to avoid rebalances, and use the cooperative sticky assignor so rebalances do not stop everyone.

Handle failures without blocking

Send poison messages to a **retry or dead-letter topic** after a few attempts instead of retrying forever.

Fix hot partitions

Better keys (add a sub-key or salt), or separate topics for very large tenants.

Prioritise

Separate urgent and bulk traffic into different topics with different consumer groups so a bulk backlog does not delay critical events.

Summary

Reduce lag by adding parallelism (consumers, partitions, in-consumer concurrency), making processing cheaper (batching, caching, async I/O), tuning fetch and poll settings, and moving failures out of the main path.

Solution and explanation

What it is

A rebalance is the process of **redistributing partitions among the consumers of a group**. It is coordinated by the group coordinator (a broker) and the group leader (one consumer, in the classic protocol).

Triggers

• A consumer joins the group (scale-out, deployment).
• A consumer leaves (shutdown, crash, or missing heartbeats for session.timeout.ms).
• A consumer exceeds max.poll.interval.ms without calling poll() and is considered failed.
• Partitions are added to a subscribed topic, or a new topic matches a subscription pattern.

Eager (classic) rebalance

1. All consumers stop and **revoke all** partitions ("stop the world").
2. They rejoin; the leader computes a new assignment.
3. Everyone gets new partitions and resumes from committed offsets.

During this time the whole group processes nothing; with many consumers this can take seconds to minutes.

Problems caused by rebalances

• Processing pauses and growing lag.
• Duplicates: work done after the last commit is redone by the new owner.
• **Rebalance storms** during rolling deployments, or when processing is slow and consumers keep timing out.

Reducing their impact

• **Cooperative sticky assignor** (incremental rebalancing): only moved partitions are revoked.
• **Static membership:** group.instance.id per consumer, so a restart within session.timeout.ms does not trigger a rebalance (great for Kubernetes pods).
• Commit offsets in onPartitionsRevoked.
• Keep processing per poll short (max.poll.records, max.poll.interval.ms).
• Kafka 4.0's new consumer group protocol (KIP-848) moves assignment to the broker and removes the global synchronisation barrier.

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    public void onPartitionsRevoked(Collection<TopicPartition> parts) { consumer.commitSync(currentOffsets); }
    public void onPartitionsAssigned(Collection<TopicPartition> parts) { /* load state if needed */ }
});

Summary

A rebalance reassigns partitions when group membership or subscriptions change; it pauses processing, so use cooperative assignment, static membership and quick polling to minimise it.

Solution and explanation

What it is

Cooperative (incremental) rebalancing, introduced in Kafka 2.4 (KIP-429), lets consumers **keep the partitions they already own** during a rebalance. Only partitions that actually need to move are revoked, and they are moved in a second round.

Eager versus cooperative

Example

Group with consumers A and B, 6 partitions; consumer C joins.

• **Eager:** A and B drop all 6; everyone pauses; partitions are reassigned 2/2/2.
• **Cooperative:** A gives up one partition, B gives up one; they keep processing their other partitions; C then receives the two freed partitions.

Enabling it

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

To migrate a running group from eager without downtime, do two rolling restarts: first list both CooperativeStickyAssignor and the old assignor, then remove the old one.

Spring Boot:

spring.kafka.consumer.properties.partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Kafka Streams and Kafka 4.0

Kafka Streams uses cooperative rebalancing by default. Kafka 4.0's new consumer rebalance protocol (KIP-848, group.protocol=consumer) goes further: the broker computes assignments incrementally and consumers do not need to synchronise at all.

Summary

Cooperative rebalancing revokes only the partitions that move, so most of the group keeps processing during scaling and deployments; enable it with the CooperativeStickyAssignor.

Solution and explanation

What it controls

max.poll.interval.ms (default **300000 ms = 5 minutes**) is the **maximum time allowed between two poll() calls**. If the consumer does not call poll() within this time, it is considered stuck: it leaves the group, a rebalance happens and its partitions go to other consumers.

This is separate from heartbeats: a background thread sends heartbeats (so the consumer looks alive), but max.poll.interval.ms detects a consumer whose processing loop is stuck or too slow.

Typical failure

poll() returns 500 records
Each record calls a slow API: 1 s each → 500 s > 300 s
→ consumer kicked out, rebalance, another consumer reprocesses the same records
→ it also takes 500 s → endless rebalance loop, lag grows

Log message: "consumer poll timeout has expired... the poll loop is spending too much time processing messages".

Fixes

• Lower max.poll.records so each batch finishes in time:

max.poll.records=50
max.poll.interval.ms=300000

• Increase max.poll.interval.ms if processing legitimately takes longer (e.g. 15 minutes for heavy batch jobs), knowing that a truly stuck consumer will be detected later.
• Speed up processing (batch writes, async calls with bounded concurrency).
• For very long tasks, pause() the partitions, process in another thread while continuing to call poll() (which returns nothing for paused partitions), then resume().

Related settings

• session.timeout.ms: liveness via heartbeats (process crash, network).
• heartbeat.interval.ms: how often heartbeats are sent.

Summary

max.poll.interval.ms is the maximum gap between polls before a consumer is treated as failed; keep batch processing time well below it by tuning max.poll.records or processing speed.