Offset Management & Commit Strategies

The Requirement

Two problems have been sitting quietly under everything you have built so far.

Problem 1. Your consumer restarts. How does it know exactly where to continue? It must not redo everything, and it must not silently skip data it never handled.

Problem 2. Real event streams sometimes contain a bad message: broken JSON, a missing field, something a bug produced somewhere. What does your consumer do then? Does it crash? Does it get stuck on the same message forever?

Both problems come from the same mechanism: offset commits. Today we fix both.

---

What Committing an Offset Actually Means

Is a committed offset "the last message I read"?

No. A committed offset is the next offset you should start from.

If you have fully handled everything up to offset 40, you commit offset 41. After a restart, the consumer resumes from 41. When you commit by hand for one specific record, you write it yourself: new OffsetAndMetadata(record.offset() + 1). When you call commitSync() with no arguments, or use auto-commit, the client uses the consumer's position, which already is the next offset.

Where does Kafka keep it?

In the internal topic __consumer_offsets, the same topic that picks your group coordinator. Each group has its own committed position per partition.

An offset goes through three stages. Keep these three words in mind, because the whole lecture depends on them:

Only a committed offset survives a restart. And notice the middle stage: the position moves when a record is returned to you, not when you have finished it. That gap is the source of the danger we look at soon.

---

auto.offset.reset — Clearing Up the Most Common Confusion

In the first consumer lecture, you set auto.offset.reset to "earliest" with a short explanation. Now that you know what a committed offset is, the full explanation will finally make sense.

Does auto.offset.reset decide where a consumer starts every time it runs?

No. This is the biggest misconception about this setting. It matters in only one situation.

When does auto.offset.reset actually apply?

Every time a consumer starts, it asks one question for each partition: "Is there already a valid committed offset for this group.id?"

• If yes, the consumer resumes from it. auto.offset.reset is never even looked at.
• If no, only then does auto.offset.reset decide the starting point.

When is there "no valid committed offset"?

Three common cases:

1. A brand-new group. This group.id has never run before.
2. The committed offset expired. Kafka deletes a group's committed offsets after the group has had no members for a while. The default is 7 days (the broker setting offsets.retention.minutes). So a consumer that was switched off for a week and then restarted behaves like a brand-new group.
3. The offset is out of range. The data it points to no longer exists. For example, old messages were deleted by the topic's retention, or the topic was recreated with fewer messages.

That is why it is called a reset, not a default. It answers the question: "What do we do when we have no reliable position?"

The Options, and When to Choose Each

(The duration is written in the standard ISO 8601 form. PT24H means 24 hours.)

Which one is the default if I set nothing?

latest. That is why we set earliest on purpose from the start.

What does none look like in practice?

The consumer reads nothing. Instead, poll() throws a NoOffsetForPartitionException, with a message like Undefined offset with no reset policy for partitions: [clickstream-events-0, ...].

Output / Note

Optional experiment: Take RecommendationConsumer. Change GROUP_ID to a brand-new name and set auto.offset.reset to "none". Run it. You will see the exception, and no events. Then try "latest" (no events until you produce new ones) and "earliest" (all the history). Put everything back afterwards, because the next steps use the original group.

Let's think about it with our own pipeline. Suppose you start a new group to backfill six months of clickstream-events into an analytics table. You want earliest, because missing history defeats the purpose. But if you launch a live "what is trending right now" dashboard, replaying six months of old clicks at startup would be wrong. You want latest.

---

Now let's go back to committing. Even with a good starting point, the default commit behavior is more fragile than it looks.

---

The Hidden Danger of the Defaults

The Java consumer has two settings that work together:

• enable.auto.commit (default true). The consumer commits its position by itself. It does this inside your calls to poll(): once auto.commit.interval.ms (default 5 seconds) has passed, the next poll() commits the current position. It also commits when a rebalance starts, and when you call close().
• The position itself. Remember the middle stage of the table: the position moves forward the moment poll() returns records. Not when you have processed them.

So auto-commit does not ask "did you finish?" It only knows "the last poll() returned everything up to here". It trusts that you did the work.

Is that a problem?

Not always. If your loop is simple, and every record is fully processed before the next poll(), the defaults give you at-least-once delivery. The periodic commit happens at the start of a poll(), and by then the previous batch is done.

But real code is not always simple. Let's look at the picture.

Default settings vs. the safe pattern

Let's walk through the left side.

1. poll() returns a batch of 10 records, with offsets 0 to 9. The position moves to 10 right away.
2. Your loop starts processing. It handles records 0, 1, 2 and 3. Then record 4 throws an exception that nobody catches.
3. The exception leaves the loop. Your finally block runs consumer.close(). With auto-commit on, close() commits the position: offset 10.
4. You restart. The consumer resumes at 10.

What happened to records 4 to 9?

They are gone. They were not retried. They were not logged as missing. They were silently skipped, because Kafka was told they were done when they never were.

There is a second way to get the same result: handing records to other threads. Your loop takes a record, gives it to a worker, and returns to poll(). To the consumer, that record is "consumed". If the worker fails later, no commit knows about it.

You do not have to take this on trust. In a moment, you will run a small class that shows the numbers.

---

Let's Build It — The Correct Pattern

The fix is small. We stop letting the client commit behind our back, and commit ourselves, after the work.

Step 1: Turn Off Auto-Commit

javaCopy
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

Now nothing is committed unless we say so. That includes close(): it commits nothing.

Step 2: Commit After the Batch Is Done

javaCopy
while (true) {
    ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofSeconds(1));

    for (ConsumerRecord<byte[], byte[]> record : records) {
        // ... process every record ...
    }

    // Only now, after EVERY record of this batch is done or dead-lettered,
    // do we commit. A crash before this line means the batch is read again.
    if (!records.isEmpty()) {
        consumer.commitSync();
    }
}

What does commitSync() commit?

The consumer's current position for every partition it owns. That is "everything the last poll() returned". This is safe now, for one reason: we call it only after the for loop finished. Every record of the batch is done.

Why "sync"?

commitSync() waits until Kafka confirms the commit, and retries temporary problems by itself. If it cannot succeed, it throws an exception, so you know. There is also commitAsync(): it sends the commit and moves on, which is faster, but you learn about failures later, in a callback. A common mix is commitAsync() inside the loop and one commitSync() before closing. We keep it simple and use commitSync().

What if a rebalance takes the partitions away while we work?

Then the commit fails with a CommitFailedException, because we no longer own those partitions. The new owner will re-read the records. That is fine for at-least-once. It is one more reason to keep batches short.

Is this now perfect?

Almost, and it is good to be honest about the last gap. If the consumer crashes after processing a record but before commitSync() runs, that batch is processed again after the restart. This pattern gives you at-least-once processing: nothing is lost, but a record can be seen twice. So write your processing to be safe when repeated, for example an "update" that gives the same result when done twice.

Could I commit after every single record?

Yes: commitSync(Map.of(partition, new OffsetAndMetadata(record.offset() + 1))). It gives you exact control, but it is slower, because each commit is a round trip to Kafka. Committing once per batch is a good balance for most consumers.

Step 3: Handle the Poison Pill

What is a poison pill?

A message that can never be processed successfully. Retrying does not help, because the message itself is the problem. For example, invalid JSON, or a missing field.

In Java, there are two kinds, and it is important to tell them apart:

Let's see what each one does to a consumer that has no protection.

The first kind. Our RecommendationConsumer uses a JsonDeserializer. When poll() reaches a record that is not valid JSON, the deserializer fails, and poll() throws a RecordDeserializationException. Its message says: "Error deserializing VALUE for partition ... at offset ... If needed, please seek past the record to continue consumption." Our loop has no protection, so the program ends. Worse, the consumer's position has not moved past the bad record. So when you restart, it reads the same record again, and fails again. Forever. This is a very common production incident.

The second kind. The record is valid JSON, so the deserializer is happy. But the field is missing. Depending on your code, it crashes with a NullPointerException, or, even more dangerous, it quietly continues with null and writes bad data downstream.

The Dead-Letter Path

The fix for both kinds is the same idea. We send the bad message to a dead-letter topic. That is a separate topic that only holds messages which could not be processed, kept for someone to inspect later. And then we move on. The offset is committed as part of the normal batch commit, because we have really handled the message, just not in the way we first planned.

Poison pill: stuck forever vs. dead-lettered and moving on

Suppose we only logged the error and gave up on the message without committing past it. We would avoid a crash right now. But on any future restart, the consumer would resume from before this message, and hit it again. And again. Forever. One message that can never succeed becomes a permanent blockage, triggered by every restart.

How do we handle both kinds in one place?

With one design decision: the consumer reads the raw bytes, and we parse them ourselves, inside our try block.

javaCopy
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());

ByteArrayDeserializer does nothing to the bytes. So poll() can never fail because of a bad value. The parsing moves into our loop:

javaCopy
for (ConsumerRecord<byte[], byte[]> record : records) {
    try {
        processEvent(record);

    } catch (Exception e) {
        String reason = e.getClass().getSimpleName() + ": "
                + String.valueOf(e.getMessage()).lines().findFirst().orElse("");
        System.out.printf("[safe-consumer] POISON PILL at partition %d offset %d: %s%n",
                record.partition(), record.offset(), reason);
        // We have "handled" this message by dead-lettering it,
        // so the consumer moves on instead of retrying it forever.
        sendToDlq(dlqProducer, record, reason);
    }
}

And processEvent is where both kinds of failure appear:

javaCopy
private static void processEvent(ConsumerRecord<byte[], byte[]> record) throws IOException {
    ClickEvent event = MAPPER.readValue(record.value(), ClickEvent.class);   // fails if not valid JSON
    if (event.userId() == null) {
        throw new IllegalArgumentException("missing field: user_id");
    }
    System.out.printf("[safe-consumer] partition %d offset %d user=%s event=%s%n",
            record.partition(), record.offset(), event.userId(), event.eventType());
}

Notice: processEvent does not catch anything. It throws. The caller decides what to do with a failure. And the caller always has the original bytes, which is exactly what the dead-letter topic needs.

Output / Note

What if I want to keep a deserializer? You can. Then the bad record surfaces as a RecordDeserializationException from poll(). That exception carries everything you need: the partition, the offset, and the raw bytes. The remedy is to save the record, and then seek past it:

javaCopy
// Not part of the lab: what you would do if you kept a JsonDeserializer
try {
    records = consumer.poll(Duration.ofSeconds(1));
} catch (RecordDeserializationException e) {
    // e.topicPartition(), e.offset(), e.valueBuffer() hold what you need for the dead-letter topic
    consumer.seek(e.topicPartition(), e.offset() + 1);   // move the position past the bad record
    continue;
}

Both ways work. We use raw bytes, because one try block then covers both kinds of poison pill.

Sending to the Dead-Letter Topic

javaCopy
private static void sendToDlq(KafkaProducer<byte[], byte[]> dlqProducer,
                              ConsumerRecord<byte[], byte[]> record, String errorReason) {
    ProducerRecord<byte[], byte[]> dlqRecord = new ProducerRecord<>(DLQ_TOPIC, record.key(), record.value());
    dlqRecord.headers().add("error_reason", errorReason.getBytes(StandardCharsets.UTF_8));
    dlqRecord.headers().add("source_topic", record.topic().getBytes(StandardCharsets.UTF_8));
    dlqRecord.headers().add("source_partition", String.valueOf(record.partition()).getBytes(StandardCharsets.UTF_8));
    dlqRecord.headers().add("source_offset", String.valueOf(record.offset()).getBytes(StandardCharsets.UTF_8));

    try {
        dlqProducer.send(dlqRecord).get(10, TimeUnit.SECONDS);   // here, waiting is exactly what we want
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new IllegalStateException("Interrupted while writing to the dead-letter topic", e);
    } catch (ExecutionException | TimeoutException e) {
        throw new IllegalStateException("Could not write to the dead-letter topic", e);
    }
}

This reuses what you already know: a producer sending a message, just aimed at a topic for failures instead of successes. Notice three details:

• Headers. We attach the error reason, and where the message came from (topic, partition, offset). Later, whoever inspects the dead-letter topic can see exactly what went wrong and where. Headers are small name and value pairs that travel with a message.
• We wait for confirmation. Remember the pitfall from the producer chapter: calling get() after every send() is slow. Here it is deliberate. Bad messages are rare, and we must be sure this one is saved before we go on.
• If the dead-letter write fails, we throw. The batch is not committed, and the consumer stops. That is intentional. If we cannot save the bad message, it is better to stop and try again after a restart than to lose the message.

Not Every Error Is a Poison Pill

One warning about catch (Exception e). It catches everything: bad data, but also bugs in your own code, and temporary problems, like a database that is down for a minute.

• A bad message will fail every time. Dead-lettering it is right.
• A temporary problem would succeed if you tried again later. Sending those messages to the dead-letter topic would fill it with perfectly good events.

In a real system, you separate the two: retry (with a pause) for temporary failures, and dead-letter only the permanent ones. Our class keeps things simple and treats every processing error as permanent, so you can see the pattern clearly.

The Complete Class

Create SafeConsumerWithDlq in the ch05 package:

javaCopy
package com.kafkacourse.ch05;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kafkacourse.common.ClickEvent;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Recommendation Consumer with safe offset handling and poison-pill survival:
 * 1. Offsets are committed only AFTER a batch has really been handled.
 * 2. A message that cannot be processed (a "poison pill") goes to a dead-letter topic
 *    instead of crashing the consumer or blocking it forever.
 */
public class SafeConsumerWithDlq {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9094,localhost:9095";
    private static final String TOPIC = "clickstream-events";
    private static final String DLQ_TOPIC = "clickstream-events-dlq";
    private static final String GROUP_ID = "recommendation-service";

    // Fields we do not know (for example "tier") are ignored. Everything else that is wrong is a poison pill.
    private static final ObjectMapper MAPPER = new ObjectMapper()
            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

    public static void main(String[] args) {
        Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        config.put(ConsumerConfig.CLIENT_ID_CONFIG, "safe-consumer");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Step 1: We commit ourselves, after the work is done. Nothing is committed behind our back.
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // We read the raw bytes and parse them ourselves, inside our try block.
        // A bad message can then never break poll() itself.
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());

        // A producer, used only to send bad messages to the dead-letter topic.
        Properties dlqConfig = new Properties();
        dlqConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        dlqConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "dlq-producer");
        dlqConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        dlqConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());

        KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(config);
        KafkaProducer<byte[], byte[]> dlqProducer = new KafkaProducer<>(dlqConfig);

        Thread mainThread = Thread.currentThread();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("\n[safe-consumer] stopping...");
            consumer.wakeup();
            try {
                mainThread.join();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }));

        try {
            consumer.subscribe(List.of(TOPIC));
            System.out.println("[safe-consumer] listening on '" + TOPIC + "'... press Stop to end.");

            while (true) {
                ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofSeconds(1));

                for (ConsumerRecord<byte[], byte[]> record : records) {
                    try {
                        processEvent(record);

                    } catch (Exception e) {
                        String reason = e.getClass().getSimpleName() + ": "
                                + String.valueOf(e.getMessage()).lines().findFirst().orElse("");
                        System.out.printf("[safe-consumer] POISON PILL at partition %d offset %d: %s%n",
                                record.partition(), record.offset(), reason);
                        // We have "handled" this message by dead-lettering it,
                        // so the consumer moves on instead of retrying it forever.
                        sendToDlq(dlqProducer, record, reason);
                    }
                }

                // Step 2: Only now, after EVERY record of this batch is done or dead-lettered,
                // do we commit. A crash before this line means the batch is read again.
                if (!records.isEmpty()) {
                    consumer.commitSync();
                }
            }

        } catch (WakeupException e) {
            // Expected: this is how wakeup() ends a poll() that is waiting.
        } finally {
            consumer.close();
            dlqProducer.close();
            System.out.println("[safe-consumer] stopped.");
        }
    }

    /**
     * Throws an exception if the message can't be handled. It is deliberately not caught here.
     * The caller decides what to do with a failure.
     */
    private static void processEvent(ConsumerRecord<byte[], byte[]> record) throws IOException {
        ClickEvent event = MAPPER.readValue(record.value(), ClickEvent.class);   // fails if not valid JSON
        if (event.userId() == null) {
            throw new IllegalArgumentException("missing field: user_id");
        }
        System.out.printf("[safe-consumer] partition %d offset %d user=%s event=%s%n",
                record.partition(), record.offset(), event.userId(), event.eventType());
    }

    /**
     * Sends the bad message to the dead-letter topic and WAITS until Kafka confirms it.
     * If that fails, we throw, so the caller does not commit. A message must never be dropped without being saved.
     */
    private static void sendToDlq(KafkaProducer<byte[], byte[]> dlqProducer,
                                  ConsumerRecord<byte[], byte[]> record, String errorReason) {
        ProducerRecord<byte[], byte[]> dlqRecord = new ProducerRecord<>(DLQ_TOPIC, record.key(), record.value());
        dlqRecord.headers().add("error_reason", errorReason.getBytes(StandardCharsets.UTF_8));
        dlqRecord.headers().add("source_topic", record.topic().getBytes(StandardCharsets.UTF_8));
        dlqRecord.headers().add("source_partition", String.valueOf(record.partition()).getBytes(StandardCharsets.UTF_8));
        dlqRecord.headers().add("source_offset", String.valueOf(record.offset()).getBytes(StandardCharsets.UTF_8));

        try {
            dlqProducer.send(dlqRecord).get(10, TimeUnit.SECONDS);   // here, waiting is exactly what we want
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Interrupted while writing to the dead-letter topic", e);
        } catch (ExecutionException | TimeoutException e) {
            throw new IllegalStateException("Could not write to the dead-letter topic", e);
        }
    }
}

---

Running It

Step 1: Start Your Cluster

powershellCopy
docker compose start
docker ps

Step 2: See the Danger With Real Numbers

Before we build the fix, let's watch the problem. The class OffsetDangerDemo does the whole experiment by itself. Each time it runs, it:

1. creates a small topic with 10 messages, message-0 to message-9,
2. runs a consumer that fails on message-4,
3. asks Kafka what was committed,
4. deletes the topic.

Create it in the ch05 package:

javaCopy
package com.kafkacourse.ch05;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

/**
 * Shows, with real numbers, what a commit strategy does when processing fails.
 * Program argument: unsafe (auto-commit, the default) or safe (commit after the work is done).
 *
 * Each run creates its own small topic with 10 messages, "message-0" to "message-9",
 * lets a consumer fail on "message-4", and then asks Kafka: what was committed?
 * The topic is deleted at the end.
 */
public class OffsetDangerDemo {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092,localhost:9094,localhost:9095";
    private static final int MESSAGES = 10;
    private static final int FAILS_ON = 4;

    public static void main(String[] args) throws Exception {
        if (args.length != 1 || !(args[0].equals("unsafe") || args[0].equals("safe"))) {
            System.out.println("Usage: OffsetDangerDemo [unsafe|safe]");
            System.out.println("In IntelliJ: Run > Edit Configurations > Program arguments.");
            return;
        }
        boolean safe = args[0].equals("safe");

        String suffix = String.valueOf(System.currentTimeMillis());
        String topic = "offset-demo-" + suffix;
        String groupId = "offset-demo-" + args[0] + "-" + suffix;
        TopicPartition partition = new TopicPartition(topic, 0);

        createTopic(topic);
        produceMessages(topic);

        System.out.println("Mode: " + args[0] + " (auto-commit is " + (safe ? "OFF, we commit after the work" : "ON, the default") + ")");
        try {
            runConsumer(topic, groupId, safe);
        } catch (IllegalStateException e) {
            System.out.println("The consumer crashed: " + e.getMessage());
        }

        OffsetAndMetadata committed = readCommittedOffset(groupId, partition);
        System.out.println();
        if (committed == null) {
            System.out.println("Committed offset: none.");
            System.out.println("A restart begins at message-0. Some messages will be processed again, but nothing is lost.");
        } else if (committed.offset() > FAILS_ON) {
            System.out.println("Committed offset: " + committed.offset());
            System.out.println("A restart continues at offset " + committed.offset() + ". So message-" + FAILS_ON
                    + " to message-" + (committed.offset() - 1) + " will NEVER be processed. They are lost.");
        } else {
            System.out.println("Committed offset: " + committed.offset());
            System.out.println("A restart continues at offset " + committed.offset()
                    + ", so message-" + FAILS_ON + " is tried again. Nothing is lost.");
        }

        deleteTopic(topic);
    }

    private static void runConsumer(String topic, String groupId, boolean safe) {
        Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, !safe);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config);
        try {
            consumer.subscribe(List.of(topic));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));

                for (ConsumerRecord<String, String> record : records) {
                    if (record.value().equals("message-" + FAILS_ON)) {
                        throw new IllegalStateException("cannot process " + record.value());
                    }
                    System.out.println("  processed " + record.value() + " (offset " + record.offset() + ")");
                }

                if (safe && !records.isEmpty()) {
                    consumer.commitSync();   // only after the whole batch is done
                }
            }
        } finally {
            consumer.close();   // with auto-commit ON, close() also commits the position
        }
    }

    private static OffsetAndMetadata readCommittedOffset(String groupId, TopicPartition partition) {
        Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config)) {
            Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(Set.of(partition));
            return committed.get(partition);
        }
    }

    private static void createTopic(String topic) throws Exception {
        Properties config = new Properties();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        try (Admin admin = Admin.create(config)) {
            admin.createTopics(List.of(new NewTopic(topic, 1, (short) 3))).all().get();
        }
    }

    private static void deleteTopic(String topic) throws Exception {
        Properties config = new Properties();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        try (Admin admin = Admin.create(config)) {
            admin.deleteTopics(List.of(topic)).all().get();
        }
    }

    private static void produceMessages(String topic) {
        Properties config = new Properties();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
            for (int i = 0; i < MESSAGES; i++) {
                producer.send(new ProducerRecord<>(topic, "message-" + i));
            }
            producer.flush();
        }
    }
}

You do not need to study the helper methods. Read only runConsumer. Notice ENABLE_AUTO_COMMIT_CONFIG is !safe, and the commitSync() line that runs only in safe mode.

Create two run configurations, like in earlier lectures: danger-unsafe with the program argument unsafe, and danger-safe with the argument safe.

Run danger-unsafe first. You should see processed message-0 to processed message-3, and then The consumer crashed: cannot process message-4. Then look at the end:

Committed offset: 10
A restart continues at offset 10. So message-4 to message-9 will NEVER be processed. They are lost.

That is the left side of the picture, with real numbers. The consumer failed on message-4, but Kafka was told everything up to 10 was done. (If your batch happened to be split differently, the number can be smaller than 10. The lesson is the same.)

Now run danger-safe. Same failure, same message. But this time:

Committed offset: none.
A restart begins at message-0. Some messages will be processed again, but nothing is lost.

Nothing was committed, because the batch was never finished. After a restart, the consumer reads it again. Some messages are processed twice. None is lost. That is at-least-once, and it is the right side of the picture.

(Each run creates and deletes its own topic. If you stop the class half way, a topic named offset-demo-... may stay behind. You can delete it with kafka-topics.sh --delete.)

Step 3: Create the Dead-Letter Topic

powershellCopy
docker exec -it kafka-1 bash

bashCopy
export PATH=$PATH:/opt/kafka/bin

bashCopy
kafka-topics.sh --create \
  --topic clickstream-events-dlq \
  --bootstrap-server kafka-1:29092 \
  --partitions 3 \
  --replication-factor 3

Step 4: Inject Two Poison Pills

In the same broker shell, start a console producer for clickstream-events:

bashCopy
kafka-console-producer.sh --topic clickstream-events --bootstrap-server kafka-1:29092

Type these two lines, pressing Enter after each. The first is not valid JSON. The second is valid JSON, but the user_id field is missing:

this is not valid json at all

{"event_type": "click"}

Then press Ctrl+C to exit the producer.

Step 5: Feel the Pain First

Run RecommendationConsumer, the class from the first consumer lecture. It reads the old events, and then it reaches the first poison pill. The program ends with an exception:

Exception in thread "main" org.apache.kafka.common.errors.RecordDeserializationException:
Error deserializing VALUE for partition clickstream-events-... at offset ... If needed, please seek past the record to continue consumption.

Run it again. It ends with the same exception, at the same offset. It is stuck. Every restart hits the same message. That is the left side of the poison-pill picture.

(Notice something else. The second poison pill, the valid JSON with the missing field, is behind the first, so our old consumer never gets there. If it did, it would not even crash. It would print user=null and go on. That is a quieter kind of bad data.)

Step 6: Start the Safe Consumer

Run SafeConsumerWithDlq.

You should see two lines starting with POISON PILL at partition ... offset .... One says JsonParseException, and the other says IllegalArgumentException: missing field: user_id.

And most importantly, the consumer keeps running. It did not crash and it did not get stuck.

Step 7: Look Into the Dead-Letter Topic

In the broker shell:

bashCopy
kafka-console-consumer.sh --topic clickstream-events-dlq --bootstrap-server kafka-1:29092 --from-beginning --timeout-ms 5000 --property print.headers=true

You should see both messages, with their headers (error_reason, source_topic, source_partition, source_offset). After 5 seconds without new messages, the command stops on its own.

Step 8: Prove the Consumer Moved On

bashCopy
kafka-consumer-groups.sh --describe --group recommendation-service --bootstrap-server kafka-1:29092

The LAG column should be 0 for every partition. The bad messages were counted as handled, because we committed right after the batch.

Now press Stop on the consumer, and start it again.

Do the poison pills come back?

No. This time there are no POISON PILL lines. The commit moved past them, so the restart begins after them. Compare that with the "stuck forever" picture: this is the difference our commit makes.

Step 9: Stop Everything

Press Stop on the consumer. Leave the broker shell with exit. Then:

powershellCopy
docker compose stop

---

Common Consumer Pitfalls

Trusting default auto-commit blindly. You now know exactly why it is fragile. It is not a vague warning. The position moves when poll() returns records, and close() commits it, whether or not you finished the work.

Letting one exception escape the poll loop. One bad message should not take down a consumer, and it must not leave a half-finished batch committed. Always wrap the processing of each record in try/catch, and decide on purpose what happens on failure.

Committing before the work is done. Any commit, whether automatic or by your own code, marks records as finished. Commit only after they are.

Handing records to other threads and going back to poll(). The consumer thinks they are done. Either wait for the workers before you commit, or use a design where the commit follows the workers' results.

Building a poison-pill trap without noticing. A bad record that is neither dead-lettered nor skipped comes back on every restart. Forever.

Dead-lettering without checking the send worked. If the dead-letter write fails silently and you still commit, the message is lost. Always confirm delivery.

Sending temporary failures to the dead-letter topic. A database outage would fill it with good messages. Retry temporary failures instead.

Never monitoring the dead-letter topic. A dead-letter topic that grows unnoticed is just a slower way to lose data. Someone has to watch it.