DE Mastery GitHub

Apache Kafka · KFK-03

Producers — batching, idempotence, ordering under retries

From your world. SOX: no duplicate and no reordered financial events, even on retry

← All exercises

Tune for throughput while keeping per-key order and no duplicates

A high-volume account-event feed must be fast and correct. For any single account, a consumer must read the events in the order that the producer produced them. There must be no duplicates, even if a produce request is retried.

Implement in solution.py:

def tune_for_throughput() -> dict:
    ...

def publish_ordered(bootstrap_servers, topic, records) -> None:
    ...

tune_for_throughput()

Return a producer config dict that is tuned for throughput. It must preserve ordering and prevent duplicates. It must satisfy all of these rules:

  • enable.idempotence is True (no duplicates, no reordering on retry).
  • acks is "all" (set it explicitly. Idempotence requires it anyway).
  • max.in.flight.requests.per.connection is an integer <= 5 (the ordering limit under retry).
  • linger.ms is a positive number (batch a little before sending).
  • batch.size is larger than the Kafka default 16384 (bigger batches).
  • compression.type is one of "lz4", "zstd", "snappy", "gzip".

Do not put bootstrap.servers in this dict. The caller adds it.

publish_ordered(bootstrap_servers, topic, records)

records is a list of (key, value) string tuples. They are already in the order that you must preserve for each key. Do these steps:

  1. Build a confluent_kafka.Producer from tune_for_throughput() plus {"bootstrap.servers": bootstrap_servers}.
  2. Produce every record with its key.
  3. Flush, so that all records are delivered.

Return nothing.

The tests check your config against each rule above (an independent rule oracle). Then they produce a stream where several keys interleave, and they consume it back. They assert two things. First, for every key, the sequence of values that they read equals the sequence that you produced for that key (per-key order preserved). Second, there are no duplicates.

Starter code

solution.py
"""Starter — not implemented yet (must FAIL the tests: gate G1). See README.md."""


def tune_for_throughput():
    # TODO: return a throughput-tuned, order-preserving, dedup-safe config.
    return {}


def publish_ordered(bootstrap_servers, topic, records):
    # TODO: build a Producer from tune_for_throughput() + bootstrap.servers and
    #       produce every record keyed by its key, then flush.
    return None

Hints

Hint 1

tune_for_throughput: return a dict with enable.idempotence=True (which pins acks='all'), a positive linger.ms, a batch.size larger than the 16384 default, a compression.type of lz4/zstd/snappy/gzip, and max.in.flight.requests.per.connection <= 5. Set acks='all' explicitly too.

Hint 2

publish_ordered: build a Producer from tune_for_throughput() PLUS the given bootstrap.servers, then produce every (key, value) in order with the record's key, and flush. Keying each record routes a key to one partition so its order is preserved.

Hint 3

The test consumes the topic back and, for each key, checks the sequence of values equals the order you produced for that key. Idempotence + in-flight<=5 keep that true even if a request is retried.

Run it locally
uv run de start KFK-03/01-ordering-under-retries
# edit workspace/KFK-03/01-ordering-under-retries/
uv run de test