Kafka Producers and Consumers

Data Engineering & Modern Data Platforms · Lesson 38 · Real-Time Data Engineering and Kafka

Lesson Overview

The cluster from Apache Kafka Architecture can describe leaders for learning.progress. Nothing in that describe call accepts a learner's progress or writes a raw object. The player still has a JSON object in memory. The landing job still has an empty prefix. A first integration attempt used a community client whose method names looked close enough, set a flag called exactly_once that the library documented as "best effort," and committed offsets inside a background thread the landing code never joined. The lab printed record counts. The raw zone missed a burst, then double-wrote another, and the flag stayed true the entire time.

This lesson uses the client Apache Kafka ships: org.apache.kafka.clients.producer.KafkaProducer and org.apache.kafka.clients.consumer.KafkaConsumer, artifact kafka-clients. The configuration keys are the producer's and consumer's own keys, the ones in the official documentation. There is no method on these classes that means "deliver this business fact once to the lakehouse." send appends bytes to a partition and, with the acknowledgements we set, waits until the leader and the in-sync replicas accept them. poll returns records at the consumer's position. What you do before the next commitSync is the landing job's durability story, and lessons 40 and 41 finish it. This lesson builds a correct loop so those lessons have something real to harden.

The specialty primer covers the same classes in Producer API, Consumer API, Batching, and Serialization. Use those pages for the full property list and for client tuning drills. Use this lesson to decide what the player sends, what the landing job is willing to commit, and how a bad JSON payload gets off the partition without a custom client. Keys, UTF-8 JSON bytes, acknowledgements, the poll loop, and a quarantine publish are the pipeline. Spring templates and listener annotations are KafkaTemplate and @KafkaListener. They are not the API this module standardizes on.

Learning Objectives

By the end of this lesson you will be able to:

1. Produce learning.progress with KafkaProducer, a string key, and UTF-8 JSON bytes, using documented configuration keys.
2. Explain what send has and has not done when it returns, and what acks=all waits for.
3. Choose learner_id as the key and describe what a null key does under the sticky partitioner.
4. Run one KafkaConsumer on one thread, poll in a loop, and parse bytes in application code.
5. Quarantine an invalid payload to learning.progress.dlq with the original bytes and a reason header, then continue the partition.
6. Call commitSync only after the raw write or the quarantine publish has succeeded, and name the crash window that remains.
7. State the thread-safety rules: one producer may be shared; one consumer may not.

Prerequisites

• Apache Kafka Architecture: leaders, bootstrap, and the rule that clients discover leaders rather than writing to a controller.
• Streaming Data Architecture: the JSON contract, the key, and the landing handoff. The Python classify_progress rules still define "valid."
• Data Formats and Serialization: the value is an encoded document. The broker stores bytes.
• Java 17 and org.apache.kafka:kafka-clients on the 3.9 line, or the patch your platform qualifies. The class names in this lesson are the Apache client, not a wrapper.
• /learn/kafka/producer-api and /learn/kafka/consumer-api if you want every callback variant. This lesson uses the ones the bus needs.

Detailed Core Concepts

The producer appends; it does not land the lake

KafkaProducer is a client process. It batches records, compresses batches when you ask, picks a partition, and sends produce requests to the current leader. It retries those requests according to its timeout settings. Since Apache Kafka 3.0 the defaults include acks=all and enable.idempotence=true, which means a retry of the same batch in the same producer session does not leave a second copy on the broker. That sentence is narrower than it sounds. Lesson 41 spends a whole lesson on the difference between a retried batch and a player that starts over and sends the same event_id in a new session. For this lesson, remember the split: the producer default stops duplicate appends caused by its own retries; it does not collapse two intentional send calls.

send is asynchronous. It hands the record to the producer's buffer and returns a Future. The record is not durable because send returned. It is durable, in the cluster sense, when the future completes without exception, or when your Callback is invoked with a null exception and a RecordMetadata that carries topic, partition, and offset. flush blocks until the buffered records for that producer complete. close flushes and then releases the sender thread. A player process that calls send and exits without flush or close will lose the tail. A long-running player calls send with a callback and treats callback failure as "the event was not acknowledged," which is the only honest signal to show the learner's client.

The producer object is thread-safe. The player can share one KafkaProducer across request threads. Creating a producer per event pays for a new network stack and defeats batching. Creating one producer and then wrapping every call in a process-wide lock also defeats the point. Share the producer. Do not share a consumer.

Keys, partitions, and the sticky partitioner

The record key is learner_id, a Java String, serialized with StringSerializer (UTF-8). The default partitioner hashes that key and maps it onto the current partition count. The same learner stays on the same partition while the count stays constant, which is the order the contract asked for. The value repeats learner_id so the landing validator can reject a key that does not match the payload.

A null key does not mean "you pick a stable partition for me." On current clients the default partitioner is sticky about null keys: it fills a batch against one partition, then switches. That is good for throughput of keyless logs and useless for per-learner order. Progress events therefore always have a key. A test that sends a null key to learning.progress fails in the producer wrapper before send, not on the broker. The broker will happily store a null key. The contract will not.

You usually do not set the partition number yourself. A hard-coded partition bypasses the key and piles every learner you forgot to hash onto one leader. The only time this bus sets a partition is a repair tool that must append to the partition it is repairing, and that tool is not the player.

Bytes in, bytes out

ByteArraySerializer and ByteArrayDeserializer copy bytes unchanged. The player encodes JSON as UTF-8. The landing job decodes UTF-8 and parses JSON. If the bytes are not valid UTF-8 or not the object the contract requires, application code quarantines them. A StringDeserializer would also give you a string and would still leave JSON parsing to you. Bytes are the better landing type because a quarantine publish can forward the original array without a second encoding, and a document that is not valid UTF-8 still has a body to store.

Serializers run on the caller of send, before the record enters the batch. A serializer exception fails that send immediately. It is not a broker error. Deserializers run inside poll on the consumer thread. A deserializer that throws will fail the poll and, if you let it, stall the assignment. That is why this consumer uses ByteArrayDeserializer, which does not throw on payload shape, and a parse method that returns a reason. Spring's error-handling deserializer is a different library. We do not need it to keep the partition moving.

Headers are optional key-value pairs on the record. The broker does not interpret them as the partitioning key. The landing job adds headers on the quarantine record: reason, origin_topic, origin_partition, origin_offset. Operators can list a bad hour without decoding every body. The original value bytes stay the body, so a repaired publisher can be tested against the real payload.

Acknowledgements, batches, and linger

acks=all asks the leader to wait until the in-sync replicas required by the topic have the record. It does not by itself choose how many replicas exist. Lesson 39 sets replication factor and min.insync.replicas. With those at 3 and 2, acks=all is the producer's half of "an acknowledged write survives one broker loss." acks=1 returns when the leader has the record and the followers might not. acks=0 returns when the buffer accepts the send. The player uses all. A metrics sidecar that can lose a sample might use 1. Progress is not a sample.

linger.ms is how long the producer waits to fill a batch. batch.size is how large that batch may get. The defaults favor latency: linger of 0 sends as soon as the sender thread can. A live lesson is allowed to add a small linger, on the order of 10 to 50 milliseconds, because the tile's bound is minutes and a fuller batch uses the network and the disk better. A linger of several seconds makes the tile and the landing lag look like a broker problem. Compression (compression.type, lz4 or zstd are the usual choices; both are available to the official client) applies to the batch. It saves retention bytes and costs CPU on the producer and on the broker that serves fetches. For JSON progress events the saving is real. It does not change the record the consumer sees. The consumer reads uncompressed records. You do not gunzip in application code.

delivery.timeout.ms bounds how long the producer will retry. The default is 120 seconds. When it expires, the callback gets a timeout and the player must decide to retry with the same event_id or surface an error. An infinite application retry on top of an idempotent producer is safe for broker duplicates from that process and unsafe if you also increment a local "sent" counter before the callback. Count the event when the callback succeeds.

The consumer is a loop on one thread

KafkaConsumer is not thread-safe. One thread calls poll, processes the records, and commits. A second thread calling poll on the same instance corrupts the client. If you need parallelism, you run more consumer instances in the same group, up to the number of partitions, which is lesson 40. You do not hand the same KafkaConsumer to a pool.

poll(Duration) returns the records the broker has for this member, up to max.poll.records (default 500), and it also sends the heartbeats and join logic the group protocol requires. The duration argument is how long poll may block waiting for records. It is not max.poll.interval.ms. That second setting is the longest you may go between poll calls before the group decides you are stuck. Lesson 40 makes the interval a landing-job design parameter. This lesson keeps processing inside the loop short enough that a default of five minutes is not interesting: validate, append a line, maybe publish a quarantine, commit.

subscribe with a topic list asks the group coordinator to assign partitions. You do not pick partitions in the landing job. assign is for a tool that must read a specific partition without joining a group, such as a one-off replay that you do not want to commit into learning-progress-raw-landing. A replay tool that accidentally uses the production group.id will move the landing cursor. The group id is part of the production configuration and is absent from ad-hoc tools.

enable.auto.commit defaults to true. Leave it false on the landing consumer. Auto-commit records positions on a timer inside poll, which is not the same moment your object store returns success. Lesson 40 explains the failure in detail. The loop in this lesson sets the flag to false and calls commitSync after a successful write. commitSync blocks until the coordinator accepts the offsets or throws. The offset you commit is the next one to read, which commitSync() with no arguments derives from the last poll. That detail is the whole of the next-but-one lesson's opening. Use the call correctly now; we will justify the arithmetic in lesson 40.

What the landing write is

The practical consumer appends JSON Lines to a file named by topic, partition, and the first offset in the batch, and it fsyncs that file before commitSync. A real lake landing swaps the local file for an object put that is durable before the method returns. The idempotency key is the same either way: if the process restarts and polls the same offsets, it writes the same object name again. A crash after the fsync and before commitSync repeats the write and does not create a second name. A crash before the fsync repeats the work and the previous partial file is replaced. This is at-least-once processing with an idempotent side effect. It is not a Kafka transaction. Lesson 41 says when a transaction would be the wrong tool for this hop and when it is the right tool for a consumer that also produces to another topic.

The Spark tile does not use this consumer. It uses the Spark Kafka source and a checkpoint. Do not "simplify" the platform by replacing the tile with this loop. Do not "simplify" landing by pointing Spark at the raw prefix and also committing a group offset in the same job without deciding which cursor wins. One job, one cursor.

Architecture and Design Diagrams

mermaidCopy
flowchart LR
    subgraph player [Player process]
        enc[UTF-8 JSON bytes]
        prod[KafkaProducer shared across threads]
    end

    subgraph cluster [Partition leader]
        log[learning.progress log]
        dlq[learning.progress.dlq]
    end

    subgraph landing [Landing process one thread]
        cons[KafkaConsumer poll]
        rules[classify payload]
        raw[Durable JSON Lines]
        bad[Quarantine producer]
    end

    enc --> prod
    prod -->|key learner_id acks all| log
    log --> cons
    cons --> rules
    rules -->|valid| raw
    rules -->|invalid bytes| bad
    bad --> dlq
    raw -->|then commitSync| cons
    bad -->|then commitSync| cons

The quarantine publish is a produce to a different topic, using a second KafkaProducer inside the landing process. It happens before the offset commit. A failure to publish the quarantine fails the loop. The offset stays put. The same bad record returns on the next poll. That is louder than skipping, and it is what you want until the quarantine topic is writable.

mermaidCopy
sequenceDiagram
    participant App as Player thread
    participant KP as KafkaProducer
    participant Leader as Partition leader
    participant Land as Landing thread
    participant Store as Raw file or object

    App->>KP: send key and bytes
    Note over KP: buffered, not yet acknowledged
    KP->>Leader: produce batch
    Leader-->>KP: RecordMetadata offset
    KP-->>App: callback exception is null
    Land->>Leader: poll from committed position
    Leader-->>Land: records including that offset
    Land->>Store: write and fsync object for those offsets
    Land->>Leader: commitSync next offset

If the process dies on the line between fsync and commitSync, the next poll reads the same records and the same object name is written again. If the process dies before the callback, the player retries with the same event_id, and a second offset may appear. Both behaviors are visible in the raw zone. Hidden behavior is what we are designing out.

Technical Explanation

Configuration that belongs to this bus

Producer keys we set explicitly, even when they match a default, so a platform upgrade that changes a default does not change the bus silently:

Consumer keys for landing:

auto.offset.reset=earliest applies when the group has no committed offset. It does not rewind a group that already has one. Resetting a live group is an administrative act, lesson 40, not a config reload. Choosing latest on the landing group would drop history the first time the group id was recreated, which is the same hole as a lost Spark checkpoint with startingOffsets=latest. Earliest is the loud, complete choice for a group whose job is to land what the log still holds. If retention has already deleted the morning you needed, earliest cannot invent it. That is a retention incident, not a reason to prefer latest.

max.poll.records can stay at 500 while the write is a local fsync. If the object store needs smaller batches so a single poll stays well under max.poll.interval.ms, lower max.poll.records. Do not raise the interval into hours to hide a stuck put. An hour-long poll gap is an hour the group may be reassigning, and the commit will fail with CommitFailedException because the member no longer owns the partitions. You will meet that exception properly in lesson 40. The practical defense in this lesson is a bounded batch.

Callbacks and futures

A callback runs on the producer I/O thread. It must be quick. It must not call back into a KafkaConsumer. It may record the failure in a counter the player thread reads. Blocking the callback blocks the sender and stalls every request sharing the producer.

Future.get on the application thread is the right shape for a batch republisher that reads a file and must not advance the file offset until Kafka accepts the record. It is the wrong shape for the web player, where a request thread should not hold a container thread for up to delivery.timeout.ms. The player uses a callback. The file republisher uses get. Both check the exception. Neither treats a returned future as an acknowledgement.

RecordMetadata.offset is the offset the leader assigned. Log it with event_id on the player only if you are allowed to retain that mapping. It is the best join key when a learner reports a lost progress update and you need to see whether the callback ever ran. Do not use it as event_id. The next produce of the same logical click, if the player retries after a lost callback response, receives a new offset.

Quarantine without stopping the partition

For each record, the landing loop runs the contract checks. On success it adds a JSON line that includes the payload plus kafka_topic, kafka_partition, kafka_offset, and kafka_timestamp from the ConsumerRecord. On failure it produces the original bytes to learning.progress.dlq with headers and waits for that produce to complete (get, because this is the republisher shape). Then it moves on. After the whole poll batch is either lined in the raw file or present on the DLQ, it fsyncs and commitSyncs.

If the DLQ produce fails, the loop does not commit. The next poll will see the bad record again and will retry the quarantine. You may also see the valid records in the same poll written again. The raw file name includes the first offset, so the rewrite replaces the batch. That is the idempotent side effect. A line-oriented append to a single growing file named only by date is not idempotent: a retry duplicates lines. The walkthrough therefore uses one file per poll batch, named by partition and starting offset. A production object store does the same with an object key. Lesson 41 generalizes the pattern. The file layout here is already the one that makes the crash window safe.

Ordering the landing job can rely on

Within one partition, poll returns records in offset order. Across partitions in the same poll, there is no order. The landing job must not sort the batch by event_time and then assume that order is how the log will replay. Write each record with its offset. Let the curator sort by event_time when it builds course_day. A landing job that reorders and drops offsets will build a raw zone the manifest cannot repair.

The consumer sees records up to the high watermark, as lesson 37 described. A record whose callback has not fired is not visible to landing, because it is not in the log yet. You cannot test the landing job against "sent but not acknowledged" and expect a file.

Metrics that tell you which process is sick

client.id shows up in client metrics: outgoing byte rate, request latency, record error rate on the producer; fetch latency, records consumed rate, and commit rate on the consumer. Broker metrics still matter for under-replication. A producer error rate with a healthy ISR points at the client: serialization, message size, or authentication. A consumer that polls and commits while the raw prefix is idle points at a write path that commits too early, which is the bug this lesson's loop is structured to avoid. If you only alert on consume rate, that bug looks like health.

Practical Implementation or Design Walkthrough

Problem

Implement the player publish and the landing consume for learning.progress with the official client. Valid records become a durable JSON Lines batch. Invalid records become DLQ records. The group offset moves only after those writes succeed. The Spark tile stays on its checkpoint and is not this program.

Dependencies

xmlCopy
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.9.0</version>
</dependency>

Java 17. The topic exists with the replication settings of lesson 39; until that lesson, a lab topic is acceptable if you label it as a lab. learning.progress.dlq exists and the landing principal can write it.

Producer

javaCopy
import org.apache.kafka.clients.producer.Callback;
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.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

public final class ProgressProducer implements AutoCloseable {
    private final KafkaProducer<String, byte[]> producer;

    public ProgressProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "web-player-progress");
        this.producer = new KafkaProducer<>(props);
    }

    /** Returns after the record is buffered. Durability is the callback, not the return. */
    public void publish(String learnerId, byte[] jsonUtf8, Callback callback) {
        if (learnerId == null || learnerId.isBlank()) {
            throw new IllegalArgumentException("progress key is required");
        }
        ProducerRecord<String, byte[]> record =
                new ProducerRecord<>("learning.progress", learnerId, jsonUtf8);
        producer.send(record, callback);
    }

    @Override
    public void close() {
        producer.close();
    }
}

A caller that must block, such as a file republisher, uses producer.send(record).get() inside a try and treats InterruptedException and ExecutionException as failure to advance the input file. The web tier uses the callback form:

javaCopy
publisher.publish(learnerId, json, (RecordMetadata metadata, Exception error) -> {
    if (error != null) {
        // The player may retry later with the same event_id. Do not mark the click sent.
        failureCounter.increment();
        return;
    }
    acknowledged.increment();
});

Consumer

The loop below is the landing job's core. writeBatch fsyncs a file whose name contains the partition and the first offset. quarantine uses send(...).get() so a DLQ failure throws before the commit. classify is the Java equivalent of lesson 36's rules; keep the two implementations on one shared test list so they cannot drift.

javaCopy
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.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public final class ProgressLandingLoop implements AutoCloseable {
    private final KafkaConsumer<String, byte[]> consumer;
    private final KafkaProducer<String, byte[]> dlqProducer;

    public ProgressLandingLoop(String bootstrapServers) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "learning-progress-raw-landing");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "progress-raw-landing");
        this.consumer = new KafkaConsumer<>(props);
        Properties dlqProps = new Properties();
        dlqProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        dlqProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        dlqProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        dlqProps.put(ProducerConfig.ACKS_CONFIG, "all");
        dlqProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        dlqProps.put(ProducerConfig.CLIENT_ID_CONFIG, "progress-raw-landing-dlq");
        this.dlqProducer = new KafkaProducer<>(dlqProps);
        this.consumer.subscribe(List.of("learning.progress"));
    }

    public void pollOnce() throws Exception {
        ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(500));
        if (records.isEmpty()) {
            return;
        }
        for (ConsumerRecord<String, byte[]> record : records) {
            String reason = classify(record.key(), record.value());
            if (reason == null) {
                writeRaw(record);
            } else {
                quarantine(record, reason);
            }
        }
        syncRaw();
        consumer.commitSync();
    }

    private void quarantine(ConsumerRecord<String, byte[]> record, String reason) throws Exception {
        ProducerRecord<String, byte[]> out = new ProducerRecord<>(
                "learning.progress.dlq", record.key(), record.value());
        out.headers()
                .add(new RecordHeader("reason", reason.getBytes(StandardCharsets.UTF_8)))
                .add(new RecordHeader("origin_topic", record.topic().getBytes(StandardCharsets.UTF_8)))
                .add(new RecordHeader("origin_partition",
                        Integer.toString(record.partition()).getBytes(StandardCharsets.UTF_8)))
                .add(new RecordHeader("origin_offset",
                        Long.toString(record.offset()).getBytes(StandardCharsets.UTF_8)));
        dlqProducer.send(out).get();
    }

    @Override
    public void close() {
        consumer.close();
        dlqProducer.close();
    }

    // classify, writeRaw, and syncRaw are the contract and the fsync. They do not call Kafka.
    private static String classify(String key, byte[] value) { return null; }
    private static void writeRaw(ConsumerRecord<String, byte[]> record) {}
    private static void syncRaw() {}
}

classify returning null for the snippet means "wire the real checks before this runs." A stub that always returns null will land invalid events and the exercise will catch it. writeRaw must include the offset in the file name or object key before you call the loop against a real broker.

Shutdown

Close the consumer from the poll thread, or call consumer.wakeup() from another thread and let the poll thread see WakeupException, commit if the last batch was synced, and then close. wakeup is the supported cross-thread signal. Closing from a foreign thread is not. The producer closes after the consumer, so a final quarantine send is not racing a closed producer. On the player, close the producer on process shutdown so the last linger batch flushes.

What you should observe

Publish the sample progress document from lesson 36 with key lrn_1842. The callback's metadata shows a partition and an offset. The landing file for that partition and offset contains one line, and a second pollOnce without new publishes does not create a second file once the commit has succeeded. Publish a payload with progress_pct equal to 140. The raw file does not contain it. The DLQ record's reason header is bad_progress_pct, and the body is the original bytes. Kill the landing process after syncRaw and before commitSync, restart, and confirm the same file name is written again and the line count inside it is still one. That kill is the test. A green log line that says "consumed" is not the test.

Enterprise Real-World Scenarios

Scenario 1: The player shares one producer and respects the callback

A live lesson sends tens of progress updates per second per busy course, across many learners. The web tier keeps a single ProgressProducer for the process. Request threads call publish and complete the HTTP request when the callback succeeds, or they return a retryable error when it fails. They do not mark the lesson complete in the browser solely because send returned.

acks=all and a 20 millisecond linger add well under a tenth of a second to a path whose product bound is the five-minute tile. The cost is producer CPU for lz4 and a bit of buffer memory, defaulting to 32 MB (buffer.memory) per producer. That is cheap next to a lost burst. Idempotence stays on so a leader change during the lesson does not duplicate the append from this process. The browser's own retry, if the HTTP response was lost, still carries the same event_id, and the raw zone may show two offsets. The morning model dedupes on event_id if finance's definition is "one observation." The landing job does not drop the second offset. Support can see both.

The player's principal can write learning.progress only. It cannot read the raw zone and it cannot write the DLQ. A compromise of the player edge does not become a rewrite of history. Security here is the ACL plus the fact that the producer never received lake credentials.

Scenario 2: Landing quarantines a bad release without stopping the tile

A player release sends progress_pct as a JSON float for twenty minutes. The landing loop publishes those bytes to the DLQ and commits past them only after the DLQ acknowledgement. Valid events in the same batches still land. The Spark tile is a different consumer with a different parser. If the tile's parser throws on a float, the micro-batch fails and the checkpoint does not advance, which is safe and loud. If the tile's parser coerces the float, the tile and the morning table diverge. The shared contract tests exist so both parsers reject the float. The incident response is a player rollback, a count of DLQ reasons, and a decision about whether any quarantined body can be repaired into a valid integer without inventing fields. The group offset is not rewound unless a valid event was skipped, which this loop does not do.

Cost of the DLQ is a second small topic with the same replication factor, retaining the bad hour. That is cheaper than a silent under-count and a backfill argument with no samples. The landing principal's write ACL on the DLQ is the security boundary. A wide "write all topics" ACL would let a bug in classify overwrite learning.enrollment once that topic exists.

Scenario 3: An ad-hoc consumer joins the landing group

An engineer debugging a missing learner runs a console consumer with --group learning-progress-raw-landing from a laptop. The group rebalances. The laptop is assigned a partition, the engineer hits Ctrl-C, and the committed position moves or the assignment flaps. Landing lag spikes. A partition the laptop read and did not write to the lake is now committed past, if the console tool committed, or is duplicated, if it did not.

Ad-hoc reads use a new group id, or assign with no commit, and they start at an offset you choose for the investigation. The production group id is treated as a credential. Lesson 40 shows how to inspect lag without joining the group as a member that commits. The console consumer in Apache Kafka's distribution is a real tool and a real foot-gun when you pass it the production group. The primer's CLI Basics shows the flags. This scenario's rule is the group id.

Best Practices

• Use KafkaProducer and KafkaConsumer from kafka-clients. Configure them with ProducerConfig and ConsumerConfig constants so a typo fails at compile time.
• Share one producer per process. Keep one consumer per thread, and do not hand that consumer to a pool.
• Key progress by learner_id. Reject a null key in the player before send.
• Store UTF-8 JSON as bytes. Parse in the application. Quarantine the original bytes.
• Set acks=all and leave idempotence on for the player. Treat the callback, not send's return, as the acknowledgement.
• Set enable.auto.commit=false on landing and call commitSync after the fsync or the object put, including after a successful DLQ produce.
• Name raw objects with topic, partition, and offset so a repeated poll replaces the same object.
• Set client.id to the service name. You will need it the first time two processes blame each other.
• Give ad-hoc tools their own group id. The landing group id is production state.
• Keep the Spark tile on the Spark source. This loop is the landing product, not a replacement for the checkpoint.

Common Mistakes and Anti-Patterns

Treating send as a completed write. The record is buffered. A process exit drops it. Wait for the callback or get, and close the producer on shutdown so linger does not eat the tail.

A new producer per request. No batching, a thread storm, and linger never helps. One producer, many threads.

Sharing a KafkaConsumer across threads. The client is not thread-safe. poll on two threads is a corrupted assignment. Scale with group members, one consumer each.

acks=0 on the player because a benchmark got faster. The benchmark measured the buffer. The log did not promise to keep the event. Progress uses all.

Null keys on progress. Sticky batching spreads one learner across partitions as soon as you retry, and per-learner order is gone. The contract requires the key.

A deserializer that throws on bad JSON. poll throws, the loop dies, and if you restart without a plan you either tight-loop the same record or skip it outside the DLQ. Bytes plus an application classify keep the failure in the contract.

Appending to one daily file and then committing. A retry appends again. The morning count doubles for that partition. Name the object by the offsets it contains.

Auto-commit left on because it is the default. The commit timer is not the object-store acknowledgement. You will skip or duplicate for reasons that do not appear in the landing log. Turn it off and commit in the loop where the write succeeded.

Joining the production group from a laptop. You become a member. Your commits and your rebalances are production. Use another group id.

Troubleshooting and Design Review

Symptom: The player logs "published" and the landing file never shows the event_id. Broker ISR is healthy.

Likely cause: "published" was printed after send returned, and the callback later received a timeout, or the process exited during linger.

Investigation: log the callback's exception class and, on success, the offset. Compare client.id request error rate with the HTTP success rate.

Corrective action: move the success log into the callback. On timeout, retry with the same event_id. Do not advance the browser's local "sent" state before the callback.

Symptom: Landing CPU is high and the raw prefix has duplicate lines for one offset range. Commits are succeeding.

Likely cause: the output file is opened in append mode under a date-only name. A commit failed once, the loop retried, and the second write appended.

Investigation: file names versus offsets inside the lines. Two lines with the same kafka_offset confirm it.

Corrective action: stop the job, rewrite the affected objects so each offset appears once, switch the writer to an offset-named object, and resume from the last good committed offset. Do not reset to earliest unless you intend to rebuild the prefix.

Symptom: poll throws on a specific payload and the process restarts in a loop.

Likely cause: a deserializer or an unbounded parse throws, or classify throws instead of returning a reason.

Investigation: the exception is on the poll thread. Capture the bytes before the parse, from a ByteArrayDeserializer path.

Corrective action: return a reason and quarantine. If the DLQ itself is down, the loop should stop without committing, which looks like lag, not like data loss. Fix the DLQ ACL or disk, then let the same offsets retry.

Symptom: After a deploy, lag is stable, commits happen, and a partition you expected on instance A is on instance B. A few minutes of events appear twice in a non-idempotent sink you added beside the file.

Likely cause: a rebalance, which is normal, plus a side sink that is not idempotent. The file path replaced its object. The extra sink appended.

Investigation: consumer group member list and the extra sink's write mode. Lesson 40 covers the rebalance itself.

Corrective action: remove the extra sink or give it the same offset identity. Do not disable rebalances. Deploys will still need them.

Symptom: A console consumer on the landing group id shows the partitions, and the landing job logs rebalances every time the engineer starts it.

Likely cause: two members, one of them a laptop, one group id.

Investigation: kafka-consumer-groups.sh --describe --group learning-progress-raw-landing --members. You will see a client that is not the landing client.id.

Corrective action: stop the console process. Confirm members. If it committed, compare the committed offset with the max offset in the raw prefix before you trust the lake. Rewind only the ranges the laptop committed without writing. The command for a controlled rewind is lesson 40. The action today is to get the laptop out of the group.

A design review of the client asks: Which class sends, and which callback defines success? What is the key, and what happens to a null? Where is JSON parsed? When does commitSync run relative to fsync and relative to the DLQ get? What is the object name? What is group.id, and which tools are forbidden to use it? What is client.id? Is idempotence on, and do we know what it does not cover? A review that shows a library method named exactly_once without the source of that method is not finished.

Interview Questions and Answers

Q1. What does KafkaProducer.send guarantee at the moment it returns?

It guarantees the record was accepted into the producer's buffer, or it throws before that, for example if the key check or the serializer fails. It does not guarantee the broker has the record. The Future or the Callback carries either an exception or RecordMetadata with the assigned offset. I wait for that completion before I tell the player the click was stored, and I close the producer on shutdown so a linger buffer is not discarded.

Q2. Why is the progress key learner_id, and why do you reject a null key?

The default partitioner hashes a non-null key, so one learner's events stay on one partition while the partition count is stable. That is the order the contract needs. A null key uses the sticky partitioner, which batches to one partition and then moves. Retries and later events for the same learner can land on different partitions, and the per-learner order is gone. The broker will store a null key. The player should refuse to send one.

Q3. Why bytes for the value instead of a string serializer or a JSON serializer in the client?

The contract is UTF-8 JSON, and the landing zone must keep the original bytes when the JSON is invalid. ByteArraySerializer does not interpret them. Parsing in application code can return a quarantine reason instead of throwing inside poll. A serializer that validates and throws on the producer is a reasonable extra guard in the player, but the landing job still has to accept that older or buggy producers may have written bad bytes. The broker is not the parser.

Q4. What does acks=all mean, and what does it not set?

The leader waits until the minimum in-sync replicas have the record before it answers success. It does not choose the replication factor or min.insync.replicas. Those are topic and broker settings in the next lesson. With a factor of three and a minimum of two, acks=all is the producer half of surviving one broker loss. acks=1 returns from the leader alone. I would not use it for progress. I would consider it only for a metric that is allowed to vanish.

Q5. Why is one producer shared, and why is one consumer not?

KafkaProducer is thread-safe and batches across callers. A producer per request loses that and multiplies connections. KafkaConsumer is not thread-safe. One thread polls, processes, and commits. Parallelism for landing is more processes in the same group.id, bounded by the number of partitions, which is a lesson 40 design. Sharing one consumer across a thread pool races the protocol state.

Q6. Where do you parse and quarantine, and why must a bad record not throw from poll?

poll returns bytes. Application code classifies them. A bad payload is produced to learning.progress.dlq with the original bytes and a reason header, and that produce completes before commitSync. If classification throws out of poll, the loop cannot commit a reasoned skip and cannot move the partition. If the DLQ produce fails, we do not commit, lag grows, and the same offset is retried. That is the failure we can operate. A skipped offset with no DLQ record is a hole in course_day.

Q7. Why is enable.auto.commit false on this consumer?

The default commits the latest polled position on a timer inside subsequent polls. That timer is not the fsync and not the DLQ acknowledgement. Committing early loses the offset if the process then dies before the write. The landing loop commits synchronously after syncRaw and after any quarantine get. The remaining crash window, fsync succeeded and commit did not, rewrites the same offset-named object. I want that window. I do not want the timer.

Q8. What is linger.ms doing on the player, and how do you know you set it too high?

It delays the batch so more records share a produce request and a compression context. Twenty milliseconds is invisible next to a five-minute tile and a fifteen-minute cohort. If landing lag and player callback latency move together by seconds after a config change, linger and batch.size are the first place I look, before I blame the brokers. I would not set linger to seconds to improve a benchmark if the callback is on the user path.

Q9. How do you keep a debug console consumer from damaging landing?

I do not pass group.id=learning-progress-raw-landing. I use a new group id that starts at the offset I am investigating, or I assign partitions without committing. The production group is the landing cursor. Joining it makes the laptop a member, triggers a rebalance, and may commit positions the lake never wrote. Inspecting lag with the consumer-groups tool, which lesson 40 uses, does not require joining as a committing member.

Q10. What will you hand lesson 40 as already decided?

The group id, auto-commit off, commitSync after a durable idempotent write, auto.offset.reset=earliest for a group with no offset, and a single poll thread. I have not yet justified commit arithmetic, rebalance timing, or lag. I have also not claimed that idempotence plus this loop is exactly-once end to end. The loop is at-least-once with an offset-named object. Lesson 41 gets to say that in the vocabulary of delivery semantics.

Hands-On Exercises

Exercise: Publish, land, quarantine, and kill the commit

Problem statement. Implement ProgressProducer and a landing loop against a lab topic. Prove acknowledgement, quarantine, and the crash between fsync and commit.

Tasks.

1. Pin kafka-clients and configure the producer with the keys in the walkthrough. Reject a null or blank learner_id.
2. Publish the valid sample from lesson 36. Log the offset from the callback only.
3. Land it with enable.auto.commit=false. The object or file name must contain topic, partition, and starting offset. Include those coordinates in the line.
4. Publish progress_pct 140. Show the DLQ header reason and show that the raw object does not contain that body.
5. Stop the process after a durable write and before commitSync. Restart and show that the object is replaced, not duplicated, and that a later successful commit advances the group.
6. State what acks=all does not configure, in one sentence that points at lesson 39.
7. Name the group id you used, and name a second group id you would use for a debug consumer.
8. Do not point this loop at the Spark checkpoint path. Say which cursor the tile still owns.

Expected deliverables. The two classes, the callback log, the raw line, the DLQ header dump, and the restart listing that shows one object.

Evaluation rubric.

You pass when the kill-before-commit restart leaves a single copy of the valid offset and the invalid payload is only on the DLQ.

Lesson Summary

KafkaProducer buffers, batches, and appends. The callback or Future is the acknowledgement, and acks=all waits for the ISR policy the topic must still define. Idempotence on that producer stops duplicates from its own retries and does not collapse a second send of the same event_id. The key is learner_id. A null key is a client bug, not a partition strategy. KafkaConsumer belongs to one thread, polls bytes, classifies them in application code, writes an offset-named object or a DLQ record, and only then commitSync. Auto-commit stays off. The Spark tile keeps its checkpoint. Ad-hoc tools keep their own group id.

Kafka Topics, Partitions and Replication sets the partition count, replication, and retention that make acks=all mean what the player thinks it means. Group assignment, lag, and commit arithmetic are Kafka Consumer Groups and Offset Management. The API tour in the primer is Producer API and Consumer API.

Further Reading

• Producer configurations — acks, enable.idempotence, linger.ms, compression.type, delivery.timeout.ms. The values in this lesson are a subset you set on purpose.
• Consumer configurations — group.id, enable.auto.commit, auto.offset.reset, max.poll.records, max.poll.interval.ms.
• KafkaProducer Javadoc and KafkaConsumer Javadoc — thread safety, wakeup, and the meaning of a committed offset as the next record to read.
• Serialization, Acknowledgements, and Batching — primer pages for the property-level detail behind the tables above.
• Streaming Data Architecture — the envelope classify implements.