Ordering & Offsets

Solution and explanation

Supported codecs

Set with compression.type on the producer (and optionally on the topic or broker):

How it works

The producer compresses a **whole batch**. The broker stores it compressed (no recompression if the topic's compression.type=producer, the default), and the consumer decompresses. Compression therefore saves network bandwidth, broker disk and page cache, and replication traffic.

JSON event payloads with repeated field names often compress 3–10×.

compression.type=zstd       # producer
linger.ms=10                # bigger batches compress better
batch.size=131072

Topic-level compression

If a topic sets compression.type=gzip and producers send lz4, the broker must **recompress**, costing CPU. Keep topics at producer unless you need to enforce a codec.

Choosing

• General purpose, high throughput: **lz4** or **zstd**.
• Storage cost matters most (long retention): **zstd**.
• Very CPU-constrained producers: **snappy** or **lz4**.
• Already compressed payloads (images, encrypted data): **none**.

Summary

Kafka supports gzip, snappy, lz4 and zstd, applied per batch by the producer; lz4 and zstd are the usual choices, and larger batches improve the ratio.

Solution and explanation

The problem

A producer sends a record, the broker writes it, but the acknowledgement is lost (network glitch, timeout). The producer retries and the record is written **twice**.

The solution

With enable.idempotence=true (default since Kafka 3.0), the broker **deduplicates retries**:

• Each producer gets a **Producer ID (PID)** and an epoch.
• Each batch sent to a partition carries a **sequence number**.
• The broker remembers the last sequence numbers per PID and partition. A retried batch with a sequence it already has is acknowledged but not written again; a gap in sequence numbers is rejected.

Result: **exactly-once writes per partition within one producer session**, and ordering is preserved even with up to 5 in-flight requests.

Required settings (set automatically)

enable.idempotence=true
acks=all
retries=2147483647
max.in.flight.requests.per.connection=5   # must be ≤ 5

If you explicitly set conflicting values (e.g. acks=1), the producer throws a configuration error in Kafka 3.x.

Limits

• Covers **retries of the same send** within a producer session. If the application crashes and resends the same business event with a new producer, it gets a new PID, so the duplicate is not detected.
• Does not cover duplicates created by your own code calling send() twice.
• Per partition, not across partitions or topics. For atomic writes across partitions, use **transactions** (transactional.id).

Consumers still need idempotency

Because duplicates can still arise from application-level resends or consumer reprocessing, consumers should deduplicate with a business key (e.g. payment id with a unique constraint).

Summary

An idempotent producer uses a producer ID and per-partition sequence numbers so brokers discard duplicate retries, giving exactly-once, in-order writes per partition; it is on by default in modern Kafka.

Solution and explanation

The scenario

1. Producer sends record R (orderId=42) to the leader
2. Leader writes R at offset 100 and replicates it
3. The acknowledgement is lost or the request times out (request.timeout.ms)
4. Producer assumes failure and retries
5. Leader writes R again at offset 101
→ Two identical records; consumers process order 42 twice

The producer cannot distinguish "the request never arrived" from "it arrived but the answer was lost".

Other sources of duplicates

• **Leader failover:** the old leader acknowledged nothing, the producer retries on the new leader, but the old leader had in fact replicated it.
• **Application retries:** the service catches an exception and calls send() again with a new producer instance.
• **Consumer side:** a consumer processes records, crashes before committing offsets, and reprocesses them after restart (at-least-once delivery).
• **Upstream retries:** an HTTP client retries a "place order" call, and your service publishes the event twice.

Ordering problems too

Without idempotence, with max.in.flight.requests.per.connection > 1, batch 1 can fail while batch 2 succeeds; batch 1's retry is then written after batch 2, reversing order.

Prevention

1. **Idempotent producer** (enable.idempotence=true, default in 3.x) removes broker-level duplicates from retries and keeps order.
2. **Transactions** for read-process-write pipelines.
3. **Idempotent consumers:** store processed event ids or use unique constraints:

INSERT INTO processed_events (event_id) VALUES (:id)   -- PRIMARY KEY; duplicate → skip

1. **Idempotency keys at the API layer** so a retried HTTP call does not create a second event.

Summary

Retries duplicate records when a write succeeded but its acknowledgement was lost; idempotent producers fix this at the broker, and idempotent consumers handle the remaining cases.

Solution and explanation

What the key does

Each record has an optional key. The default partitioner hashes it (murmur2(key) % partitions), so **all records with the same key go to the same partition**.

Reasons to use a key

1. **Ordering per entity:** events for account ACC-42 (debit, credit, freeze) are consumed in the order they were produced, because one partition is ordered and read by one consumer.
2. **Locality of processing and state:** a consumer handling a customer's partition sees all of that customer's events, which is essential for stateful stream processing (running balance, session windows, joins).
3. **Log compaction:** compacted topics keep the latest record per key; without a key compaction is impossible.
4. **Even distribution:** a high-cardinality key (user id, order id) spreads load well.

producer.send(new ProducerRecord<>("account-events", accountId, eventJson));

Choosing a key

Pitfalls

• **Hot keys:** one huge customer (or a default value like "unknown") sends most traffic to one partition, overloading one consumer.
• **Changing the partition count** remaps keys, breaking ordering during the transition.
• Keys should be stable identifiers, not timestamps or random values (random keys spread load but give no ordering benefit).
• Keys are visible in plain form; avoid sensitive data such as PAN or Aadhaar numbers.

Summary

A key routes all related records to one partition, giving per-key ordering, consistent stateful processing and enabling compaction; choose a stable, high-cardinality business id.

Solution and explanation

The recipe

1. **Use the customer id as the record key.** Same key, same partition, and a partition is strictly ordered.
2. **Keep producer ordering safe on retries:** enable.idempotence=true (default) with max.in.flight.requests.per.connection ≤ 5. Without idempotence, set in-flight to 1.
3. **Process each partition sequentially** in the consumer (the default for one consumer thread per partition). If you parallelise inside a consumer, parallelise by key, not randomly.
4. **Do not change the partition count** of the topic without a migration plan.
5. **Produce from a single place in order:** if two services emit events for the same customer, their relative order is not guaranteed.

producer.send(new ProducerRecord<>("customer-events", customerId, event));

// Parallel processing that still keeps per-key order (Confluent Parallel Consumer)
ParallelConsumerOptions<String, String> options = ParallelConsumerOptions.<String, String>builder()
    .ordering(ProcessingOrder.KEY)
    .maxConcurrency(50)
    .consumer(kafkaConsumer)
    .build();

Retries and dead letters

If processing a customer's event fails and you send it to a retry topic while continuing with later events, the order breaks. Options:

• Retry in place (blocking) for transient errors.
• When parking an event, also park subsequent events of that customer until it is resolved.
• Use version numbers so consumers can ignore stale updates (if event.version <= stored.version skip).

Across topics

Ordering is per partition, so events for one customer in orders and payments topics have no guaranteed relative order. Put related events in one topic with a type field when strict cross-type order matters.

Summary

Key by customer id, keep the idempotent producer, process each partition (or key) sequentially, avoid repartitioning, and handle retries without letting later events overtake earlier ones.

Solution and explanation

Answer: no

Kafka guarantees order **only within a partition**. A topic with several partitions has no total order: partitions are written and read in parallel, possibly by different consumers on different machines.

Produced:  A(p0)  B(p1)  C(p0)  D(p1)
Consumer may process: A, B, D, C  or  B, A, C, D ...
Guaranteed only: A before C, and B before D

Getting global order (and its cost)

• **Single-partition topic:** total order, but throughput is limited to one partition and only **one consumer** in a group can work. Acceptable for low-volume streams (configuration changes, a small audit trail).
• **Sequence numbers:** add a global sequence or timestamp and reorder downstream in a buffer, which adds latency and complexity.

What most systems actually need

Usually you need order **per entity** (per account, per order, per user), not globally. Use the entity id as the key: you get correct order where it matters plus parallelism across entities.

Timestamps are not an ordering guarantee

Record timestamps come from producers' clocks (CreateTime) or brokers (LogAppendTime); clocks differ across machines, so sorting by timestamp across partitions is approximate.

Summary

Kafka orders records only within each partition; global order requires one partition (no parallelism), so design for per-key ordering instead.

Solution and explanation

What it is

An offset commit records, for a consumer group, **the position of the next record to read** in each partition. Kafka stores commits in the internal compacted topic __consumer_offsets. After a restart or rebalance, the consumer that gets the partition resumes from the committed offset.

Partition 0: ... 97 98 99 [100] 101 ...
Processed up to 99 → commit offset 100 (the NEXT record)

How commits are made

• **Automatic:** enable.auto.commit=true commits the latest polled offsets every auto.commit.interval.ms (5 s), during poll().
• **Manual synchronous:** commitSync() blocks until the broker confirms, and retries.
• **Manual asynchronous:** commitAsync(callback) does not block; no automatic retry (a later commit supersedes it).
• **Specific offsets:** commit per partition with Map<TopicPartition, OffsetAndMetadata>.
• **In a transaction:** producer.sendOffsetsToTransaction(...) commits offsets atomically with produced records (exactly-once).

Commit timing defines the delivery guarantee

Practical pattern

while (running) {
    ConsumerRecords<String, Payment> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, Payment> r : records) handle(r.value());
    consumer.commitAsync();                 // fast path
}
consumer.commitSync();                      // final, reliable commit on shutdown

Retention of commits

Committed offsets of an inactive group expire after offsets.retention.minutes (7 days); then auto.offset.reset decides where it starts.

Summary

An offset commit saves a group's progress (next offset to read) per partition in __consumer_offsets; when you commit relative to processing determines at-least-once or at-most-once behaviour.

Solution and explanation

What it is

With enable.auto.commit=true (the default for the plain Java consumer), the consumer **commits offsets periodically** in the background of poll(): every auto.commit.interval.ms (default 5000 ms) it commits the offsets of the records returned by the **previous** poll.

enable.auto.commit=true
auto.commit.interval.ms=5000

Behaviour

Because commits happen inside poll() for records already returned, a typical loop that fully processes each batch before polling again gives **at-least-once** delivery:

while (true) {
    ConsumerRecords<String, String> recs = consumer.poll(Duration.ofMillis(500));   // commits previous batch
    recs.forEach(this::process);
}

Where it goes wrong

1. **Asynchronous processing:** if you hand records to another thread and poll again, the offsets are committed even though processing has not finished. A crash then **loses** those records.
2. **Crash between commits:** up to 5 seconds of processed records are reprocessed after a restart (duplicates).
3. **Rebalances** in the middle of an interval also cause reprocessing.
4. Little control: you cannot commit only after a database transaction succeeds.

When it is fine

• Simple, idempotent processing in the polling thread (writing to a store with upserts).
• Analytics or logging where occasional duplicates are acceptable.

Spring Kafka note

Spring Kafka sets enable.auto.commit=false by default and commits itself according to AckMode (BATCH by default, after the listener returns), which is safer.

Summary

Auto commit periodically commits polled offsets during poll(); it is convenient and at-least-once for synchronous processing, but can lose or duplicate records with async processing or crashes.

Solution and explanation

What it is

With enable.auto.commit=false, **your code decides when offsets are committed**, typically after records are fully processed (written to the database, API called). This gives precise control over delivery guarantees.

Ways to commit

consumer.commitSync();                       // all offsets returned by the last poll, blocking, retries
consumer.commitAsync((offsets, ex) -> {      // non-blocking
    if (ex != null) log.warn("commit failed", ex);
});

// Commit exactly one partition/offset after processing that record
consumer.commitSync(Map.of(
    new TopicPartition(r.topic(), r.partition()),
    new OffsetAndMetadata(r.offset() + 1)));  // +1: next record to read

Typical at-least-once pattern

try {
    while (running) {
        ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, Order> r : records) {
            orderService.save(r.value());        // idempotent upsert
        }
        consumer.commitAsync();
    }
} finally {
    try { consumer.commitSync(); } finally { consumer.close(); }
}

Spring Kafka

spring.kafka.listener.ack-mode: manual_immediate

@KafkaListener(topics = "orders", groupId = "fulfilment")
public void listen(Order order, Acknowledgment ack) {
    fulfilment.process(order);
    ack.acknowledge();     // commit after success
}

Other modes: RECORD (after each record), BATCH (after each poll, default), MANUAL.

Trade-offs

• Committing after every record is safest but slower (one request per record); committing per batch is faster but reprocesses up to a batch after a crash.
• Commit in a ConsumerRebalanceListener.onPartitionsRevoked to avoid reprocessing on rebalances.

Summary

Manual commit means committing offsets yourself after successful processing, typically per batch with commitAsync plus a final commitSync, giving reliable at-least-once delivery.

Solution and explanation

Effect

It **turns off periodic automatic offset commits**. The consumer then commits only when your code calls commitSync()/commitAsync() (or the framework commits for you, as Spring Kafka does), so offsets move forward only after you decide processing succeeded.

Why use it

• Commit only after the database write or downstream call completes, preventing lost records.
• Commit specific offsets per partition, e.g. after each record in critical flows.
• Combine with transactions: offsets are committed through the producer (sendOffsetsToTransaction) for exactly-once read-process-write.

What happens if you never commit

• The group's offset stays where it was. After a restart or rebalance, the consumer restarts from the last committed position and **reprocesses** everything since.
• With no commit at all, a new group starts from auto.offset.reset (latest by default) every time, so it may skip records entirely.

Example

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

while (running) {
    var records = consumer.poll(Duration.ofSeconds(1));
    try {
        db.inTransaction(() -> records.forEach(this::apply));
        consumer.commitSync();
    } catch (Exception e) {
        // do not commit: seek back so the batch is retried
        records.partitions().forEach(tp -> consumer.seek(tp, records.records(tp).get(0).offset()));
    }
}

Storing offsets yourself

Some systems store the offset in the same database transaction as the results (e.g. a consumer_offsets table) and seek() to it on startup, which gives effectively exactly-once processing into that database.

Summary

enable.auto.commit=false hands offset commits to your code so progress is recorded only after successful processing; you must then commit explicitly or records will be reprocessed.