7. Asynchronous Processing and Messaging

Asynchronous processing decouples the moment work is requested from the moment it is done, which is how systems absorb spikes, survive downstream failures, and keep user-facing latency low while heavy work happens elsewhere. This chapter covers the two families of messaging infrastructure — queues and logs — the delivery guarantees they can and cannot offer, and the patterns for running the background work they feed.

<a id="71-message-queues"></a>

7.1 Message Queues

A message queue is a durable buffer between a component that produces work and a component that performs it. Everything else in this subchapter follows from that one structural change: the two sides no longer have to be available, fast, or even alive at the same time.

Producer-Consumer Model

The producer-consumer model is the foundation of all asynchronous messaging. A producer creates a message describing work to be done and hands it to a broker. A consumer takes messages from the broker and does the work. Neither knows anything about the other — not its address, its availability, or how many of it there are.

The reason to introduce a broker between them is that a direct synchronous call couples the caller to the callee in four ways at once, and every one of those couplings is a failure mode:

Coupling	Synchronous call	With a queue
Temporal	Both must be up at the same instant	Consumer can be down; work waits
Capacity	Caller runs at the callee's speed	Queue absorbs the difference
Location	Caller needs the callee's address	Both only know the broker
Failure	Callee's failure becomes the caller's	Failure is retried, not propagated

Consider a user uploading a video. Synchronously, the HTTP request must wait for transcoding to finish — perhaps four minutes, holding a connection, a thread, and the user's attention, and failing entirely if the transcoder is restarting. Asynchronously, the API writes one message and returns 202 Accepted in 20 milliseconds; the transcoder picks the job up whenever it is ready.

The analogy is a restaurant's ticket rail. Waiters clip orders to the rail and walk away; cooks pull tickets when they finish the previous dish. Nobody waits for anybody. A rush of customers makes the rail longer, not the waiters slower, and a cook stepping away for two minutes costs throughput, not orders.

 SYNCHRONOUS                          ASYNCHRONOUS

  client                               client
    | POST /videos                       | POST /videos
    v                                    v
  [ API ] ---- blocks 4 min ----+      [ API ] --> 202 Accepted (20 ms)
    |                           |        |
    v                           |        v  enqueue
  [ transcoder ]                |      [ QUEUE ]  <- absorbs bursts, survives
    |                           |        |            consumer restarts
    v                           |        v  dequeue
  done -----------------------> |      [ transcoder x N ]
                                         |
  Transcoder down = request fails        v
  Traffic spike  = timeouts           done -> notify via callback/websocket

                                       Transcoder down = queue grows
                                       Traffic spike   = queue grows

# Producer: the API's only job is to record intent and return.
import json, uuid

def request_transcode(video_id, formats):
    message = {
        "job_id": str(uuid.uuid4()),   # a stable identity for dedup and tracing
        "video_id": video_id,
        "formats": formats,
        "requested_at": time.time(),
    }
    channel.basic_publish(
        exchange="",
        routing_key="transcode",
        body=json.dumps(message),
        properties=pika.BasicProperties(
            delivery_mode=2,           # persist to disk: survives a broker restart
            message_id=message["job_id"],
        ),
    )
    return {"status": "accepted", "job_id": message["job_id"]}   # 202, not 200

# Consumer: pulls work, does it, and only then acknowledges.
def on_message(ch, method, properties, body):
    job = json.loads(body)
    try:
        transcode(job["video_id"], job["formats"])
        ch.basic_ack(delivery_tag=method.delivery_tag)      # done: remove it
    except TransientError:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)  # retry
    except PermanentError:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False) # -> DLQ

channel.basic_qos(prefetch_count=10)   # never hold more than 10 unacked jobs:
                                       # this is backpressure at the consumer
channel.basic_consume(queue="transcode", on_message_callback=on_message)

The costs are real and should be stated plainly. You have added a component to operate, the result is no longer available when the request returns (so the client needs polling, a callback, or a push channel), failures happen out of band where no user sees them, and debugging now means correlating logs across a boundary. Use a queue when the work is genuinely deferrable; a synchronous call is simpler and simplicity has value.

Key Takeaways

• A broker decouples producer and consumer in time, capacity, location, and failure — that decoupling is the entire point.
• The producer's job is to record intent durably and return immediately, typically with 202 Accepted and a job ID.
• Consumers acknowledge only after the work succeeds, so a crash mid-work redelivers rather than loses.
• The costs are an extra component, deferred results, and out-of-band failures; do not queue work that the caller genuinely needs answered now.

🧪 Practice

1. List three operations in a typical e-commerce checkout that should stay synchronous and three that should be queued. Justify each.
2. The consumer above crashes after transcoding but before basic_ack. Trace exactly what happens and what the user observes.
3. Interview: A queue lets an API return in 20 ms instead of 4 minutes. What did the user actually gain, and what problem did you create for them? (Hint: the work still takes 4 minutes — who is now responsible for telling them it finished?)

Point-to-Point vs. Publish-Subscribe

Once messages flow through a broker, the next question is how many consumers should receive each one. There are exactly two answers, and they correspond to two different intentions.

Point-to-point (a queue): each message is delivered to exactly one consumer. The message represents a task that must be performed once — charge this card, resize this image, send this email. Adding consumers adds throughput, not duplication.

Publish-subscribe (a topic): each message is delivered to every subscriber. The message represents a fact that has occurred — an order was placed, a user signed up — and any number of independent parties may care about it. Adding a subscriber adds a new reaction to the same fact.

The distinction shows up in the language you use to name things. Queues carry commands (imperative: SendWelcomeEmail), topics carry events (past tense: UserRegistered). If you find yourself publishing SendEmail to a topic where three services subscribe, you have three services each sending an email.

 POINT-TO-POINT (queue)              PUBLISH-SUBSCRIBE (topic)

  producer                            publisher
     |                                    |
     v                                    v
  [  queue: m1 m2 m3  ]               [ topic: OrderPlaced ]
     |      |      |                    /       |        \
     v      v      v                   v        v         v
  worker1 worker2 worker3        [ email q ] [ ship q ] [ analytics q ]
                                      |          |           |
  m1 -> worker1 ONLY                  v          v           v
  m2 -> worker2 ONLY              consumer   consumer    consumer

  Each message handled once.       EVERY subscriber gets a COPY of m1.
  Scale = add workers.             Add a subscriber = add a reaction.

The architecturally important consequence is who knows about whom. With point-to-point, the producer chooses the queue, and therefore chooses the handler — the coupling is reduced but not removed. With publish-subscribe, the publisher names only the fact; subscribers appear and disappear without the publisher's code changing. That is what makes pub/sub the substrate for event-driven architecture: new features become new subscribers.

# Point-to-point: the producer names the task and its handler queue.
sqs.send_message(QueueUrl=CHARGE_QUEUE, MessageBody=json.dumps({
    "command": "ChargeCard", "order_id": "o-991", "amount_cents": 4999,
}))
# Exactly one worker charges the card. Two would be a duplicate charge.

# Publish-subscribe: the publisher names the FACT and knows no subscribers.
sns.publish(TopicArn=ORDER_EVENTS, Message=json.dumps({
    "event": "OrderPlaced",           # past tense: it already happened
    "order_id": "o-991",
    "user_id": "u-42",
    "total_cents": 4999,
    "occurred_at": "2026-08-25T10:14:02Z",
}))
# Email, shipping, analytics, and fraud each receive their own copy.
# Adding a loyalty-points service tomorrow requires NO change here.

The near-universal production pattern combines both: a topic fans out to one queue per subscriber. This matters more than it first appears. If subscribers read the topic directly, a slow or broken subscriber affects the others; with a queue per subscriber, each has its own buffer, its own backlog, its own retry behaviour, and its own dead-letter queue. One subscriber failing becomes one growing queue, not a shared incident.

Property	Point-to-point	Publish-subscribe
Deliveries per message	One consumer	Every subscriber
Message names a	Command (imperative)	Event (past tense)
Adding consumers	More throughput	More reactions
Producer knows the handler	Yes (chooses the queue)	No
Typical use	Task execution	State-change notification
Failure isolation	Natural (one queue)	Needs a queue per subscriber

Key Takeaways

• Queues deliver each message once and scale throughput; topics deliver to every subscriber and scale reactions.
• Name the difference in your payloads: commands are imperative, events are past tense.
• Pub/sub removes the publisher's knowledge of consumers, which is what makes event-driven extension possible.
• Fan a topic out into one queue per subscriber so a broken subscriber is isolated behind its own backlog.

🧪 Practice

1. Classify these as command or event and pick the pattern: PaymentCaptured, SendPasswordReset, InventoryReserved, GenerateMonthlyReport.
2. Draw the topology for an order system where email, warehouse, and analytics all react to an order, and the warehouse consumer is frequently slow.
3. Interview: A team publishes SendConfirmationEmail to a topic and users report duplicate emails. Diagnose it. (Hint: how many subscribers does a topic deliver to, and what does the message name imply about how many times the action should occur?)

Queue Durability and Acknowledgments

A queue's value depends entirely on whether it can be trusted not to lose work. Two mechanisms provide that trust, and they protect against different failures: durability protects against the broker dying, acknowledgments protect against the consumer dying.

Durability means the message is written to stable storage before the broker confirms receipt to the producer. Without it, a message lives only in the broker's memory and vanishes on restart. There are three layers, and all three must be configured — getting two right and one wrong loses messages.

1. Durable queue: the queue definition survives a restart. Without this the queue itself disappears, taking everything in it.
2. Persistent message: this specific message is written to disk, not only held in memory.
3. Publisher confirms: the broker tells the producer "I have it, durably." Without confirms the producer cannot know whether a message it sent survived, and send() returning is not evidence of anything.

 THE WINDOW WHERE A MESSAGE CAN VANISH

  producer            broker                        consumer
     |                   |                              |
     |--- publish ------>|                              |
     |                   | (in memory only)             |
     |<-- (no confirm) --|   X broker crashes HERE      |
     |                   |     -> message gone, and the producer
     |                   |        believes it was sent

  With persistence + publisher confirms:

     |--- publish ------>|
     |                   |-- fsync to disk --> [ log ]
     |<-- ack (confirm) -|
     |                   |   X crash -> message recovered on restart

  Cost: an fsync per message (or per batch). Throughput drops; durability
  is bought with latency, and the exchange rate is set by batching.

Acknowledgments cover the other half. When a consumer receives a message, the broker does not delete it — it marks it in flight and starts a timer. The consumer acknowledges only after the work is complete. If the consumer crashes, the timer expires and the message returns to the queue for someone else.

The critical rule follows directly: acknowledge after the work, never before. Auto-acknowledge mode — where receiving is treated as completing — turns every consumer crash into silent data loss.

# WRONG: auto-ack. The broker deletes the message on delivery.
channel.basic_consume(queue="jobs", on_message_callback=handle, auto_ack=True)
# A crash inside handle() loses the job permanently, with no error anywhere.

# RIGHT: manual ack after successful completion.
def handle(ch, method, properties, body):
    job = json.loads(body)
    try:
        process(job)                                        # do the work FIRST
        ch.basic_ack(delivery_tag=method.delivery_tag)      # then confirm
    except Exception:
        # requeue=False sends it to the dead-letter exchange instead of
        # spinning on the same failure forever.
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

channel.basic_consume(queue="jobs", on_message_callback=handle, auto_ack=False)

The in-flight timer — SQS calls it the visibility timeout, RabbitMQ the consumer timeout — is the parameter people get wrong most often. Set it shorter than the job's real duration and the message is redelivered while the first consumer is still working on it, producing duplicate processing. Set it far too long and a crashed consumer's job sits invisible for that entire period.

# For jobs with variable duration, extend the lease instead of guessing high.
def process_with_heartbeat(receipt_handle, job):
    for progress in do_work_incrementally(job):
        sqs.change_message_visibility(
            QueueUrl=QUEUE, ReceiptHandle=receipt_handle,
            VisibilityTimeout=60,      # push the deadline out another 60s while
        )                              # we are demonstrably still alive
    sqs.delete_message(QueueUrl=QUEUE, ReceiptHandle=receipt_handle)  # the ack

Note what this combination guarantees and what it does not. Persistence plus manual acknowledgment gives at-least-once delivery: nothing is lost, but a crash between finishing the work and sending the ack means the work runs again. That duplicate is not a bug you can configure away — it is the price of not losing messages, and Section 7.3 covers how consumers are made to tolerate it.

Key Takeaways

• Durability needs all three of: a durable queue, persistent messages, and publisher confirms — any one missing reopens a loss window.
• Acknowledge after the work completes; auto-ack converts consumer crashes into silent loss.
• The in-flight/visibility timeout must exceed the real job duration, or messages are redelivered while still being processed.
• Persistence plus manual ack yields at-least-once delivery, so duplicates are guaranteed to happen eventually.

🧪 Practice

1. A job takes 30-300 seconds and the visibility timeout is 60 seconds. Describe the failure and give two fixes.
2. Which of the three durability layers is missing if the producer's send() succeeds but the message is absent after a broker restart?
3. Interview: What throughput cost does persistence impose, and how do brokers reduce it without abandoning durability? (Hint: the expensive operation is the disk flush — must it happen once per message?)

Dead Letter Queues

Some messages can never succeed. The payload is malformed, it references a record that was deleted, it triggers a bug, or it asks for something the system no longer supports. Without a plan for these, a single such message becomes an infinite loop: receive, fail, requeue, receive, fail — consuming capacity forever and, in a strictly ordered queue, blocking everything behind it.

A dead letter queue (DLQ) is where such messages go after a bounded number of failed attempts. It converts an unbounded retry loop into a finite one plus a durable record of what could not be processed.

The name for the failure it prevents is a poison message: one message whose processing always fails, occupying a consumer indefinitely. In a system with ordering guarantees, one poison message can halt an entire partition.

 WITHOUT A DLQ                        WITH A DLQ (maxReceiveCount = 5)

  [ queue: BAD m2 m3 ]                 [ queue: BAD m2 m3 ]
       |                                    |
       v  receive                           v  receive (attempt 1..5)
   consumer -> FAILS                    consumer -> FAILS
       |                                    |
       +---- requeue ----+                  +---- requeue ----+  (x4)
       |                 |                  |                 |
       <-----------------+                  <-----------------+
                                            |
   Loops forever. m2 and m3 wait            v  after attempt 5
   behind it in an ordered queue.       [ DLQ: BAD ]  -> alert, inspect, fix
   One bad byte stops the system.
                                        m2 and m3 proceed normally.

# SQS: the redrive policy is the entire configuration.
JobQueue:
  Type: AWS::SQS::Queue
  Properties:
    VisibilityTimeout: 300
    RedrivePolicy:
      deadLetterTargetArn: !GetAtt JobDLQ.Arn
      maxReceiveCount: 5        # 5 receives, then it is moved to the DLQ

JobDLQ:
  Type: AWS::SQS::Queue
  Properties:
    MessageRetentionPeriod: 1209600   # 14 days: long enough to notice, triage,
                                      # fix the bug, and redrive

A DLQ is only useful if three things follow from a message arriving in it.

It must be alarmed. A DLQ nobody watches is a folder where work goes to be forgotten. DLQ depth > 0 is one of the few alert conditions that is almost always meaningful, because it means work was accepted and definitively not done.

It must carry context. The message alone rarely explains the failure. Attach the error, the attempt count, the consumer version, and the trace ID.

# Explicit dead-lettering with diagnostic context attached.
def consume(msg):
    attempts = int(msg.attributes.get("ApproximateReceiveCount", 1))
    try:
        process(msg.body)
        msg.delete()
    except PermanentError as e:
        # Do not waste four more attempts on something that cannot succeed.
        send_to_dlq(msg, reason=str(e), attempts=attempts, terminal=True)
        msg.delete()
    except TransientError as e:
        if attempts >= MAX_ATTEMPTS:
            send_to_dlq(msg, reason=str(e), attempts=attempts, terminal=False)
            msg.delete()
        else:
            raise            # let the visibility timeout redeliver it

def send_to_dlq(msg, reason, attempts, terminal):
    sqs.send_message(QueueUrl=DLQ_URL, MessageBody=msg.body,
        MessageAttributes={
            "error":        {"DataType": "String", "StringValue": reason[:256]},
            "attempts":     {"DataType": "Number", "StringValue": str(attempts)},
            "terminal":     {"DataType": "String", "StringValue": str(terminal)},
            "consumer_ver": {"DataType": "String", "StringValue": VERSION},
            "trace_id":     {"DataType": "String", "StringValue": current_trace()},
        })

It must be redrivable. Most DLQ messages are not poison at all — they are ordinary messages that failed during an outage of something downstream. Once the dependency recovers, they should be moved back to the main queue. Redrive slowly: dumping fifty thousand messages back at once re-creates the overload that caused them.

The distinction is worth internalizing, because it changes what you do next. Transient failures (a timeout, a 503) deserve retries and later redrive. Permanent failures (schema violation, deleted entity) deserve immediate dead-lettering — retrying them four more times only delays the inevitable while burning capacity.

Key Takeaways

• A DLQ bounds retries and captures messages that cannot succeed, preventing infinite loops and head-of-line blocking.
• Alert on DLQ depth: it represents accepted work that was definitively not done.
• Attach the error, attempt count, consumer version, and trace ID, or triage becomes guesswork.
• Dead-letter permanent errors immediately; retry transient ones and redrive them gradually after recovery.

🧪 Practice

1. With maxReceiveCount: 5 and a 300-second visibility timeout, how long before a permanently failing message reaches the DLQ?
2. Write the classification logic that decides between immediate dead-lettering and retry for: a 404, a 429, a JSON parse error, a 503.
3. Interview: Your DLQ has 50,000 messages from a three-hour database outage. How do you get them processed? (Hint: what happens to the database if all 50,000 arrive in the next thirty seconds?)

Message Ordering Guarantees

Ordering is the guarantee that message B, sent after message A, is processed after A. It sounds elementary and is surprisingly expensive, because ordering and parallelism are in direct opposition: the only way to guarantee two messages are processed in order is to refuse to process them at the same time.

That tension explains every design in this area. A single queue with a single consumer is perfectly ordered and cannot scale. Add a second consumer and both pull messages concurrently; message 2 may finish before message 1 even if it was delivered second. Ordering was not lost by the broker — it was lost by the concurrency you added on purpose.

 ORDER LOST BY CONCURRENCY, NOT BY THE BROKER

  queue: [ m1 m2 m3 m4 ]      delivered in order, processed in parallel

    worker A  <- m1   ......... 900 ms (slow: cold cache) ......> done
    worker B  <- m2   ... 40 ms ...> done          <- FINISHES FIRST
    worker A  <- m3   ... 50 ms ...> done

  Completion order: m2, m3, m1.  If these are "set name = Alice",
  "set name = Bob", "set name = Carol", the final value is Alice.

The resolution used by every real system is that global ordering is almost never required — per-entity ordering is. Nobody needs order 991's events ordered relative to order 992's. They need order 991's own events in order. That insight converts an unscalable requirement into a scalable one: partition by entity, guarantee order within a partition, and process partitions in parallel.

 PARTITIONED ORDERING: order per key, parallelism across keys

  hash(order_id) % 4

  partition 0: [ o-100:created  o-100:paid  o-100:shipped ]  -> consumer 1
  partition 1: [ o-101:created  o-101:paid ]                 -> consumer 2
  partition 2: [ o-102:created ]                             -> consumer 3
  partition 3: [ o-103:created  o-103:cancelled ]            -> consumer 4

  Strict order WITHIN each key. No order guarantee ACROSS keys -- and none
  is needed. Parallelism = number of partitions.

# Kafka: the partition key is the ordering unit.
producer.send(
    topic="order-events",
    key=order_id.encode(),      # SAME key -> SAME partition -> strict order
    value=json.dumps(event).encode(),
)
# Choosing this key is the single most consequential decision in the design.
# Too coarse (key="orders") -> one partition, no parallelism.
# Too fine  (key=event_id)  -> perfect parallelism, no ordering at all.

# SQS FIFO: the same idea under a different name.
sqs.send_message(
    QueueUrl=FIFO_QUEUE_URL,
    MessageBody=json.dumps(event),
    MessageGroupId=order_id,             # ordering scope; groups run in parallel
    MessageDeduplicationId=event_id,     # 5-minute dedup window
)
# One stuck message blocks its GROUP only -- which is exactly the trade:
# finer groups mean more parallelism and a smaller blast radius when stuck.

Guarantee level	Parallelism	Cost
No ordering	Unlimited	Consumers must be order-independent
Per-key / per-partition	One per partition	Hot keys become hot partitions
Global (total) order	One consumer	No horizontal scaling; a single point

Two practical warnings. First, a retry breaks ordering even inside a partition: if m1 fails and is retried while m2 succeeds, m2 landed first. Ordered processing therefore requires stopping the partition on failure, not skipping ahead — which is why one poison message can stall a Kafka partition entirely. Second, ordering guarantees are per partition, not per topic, so "Kafka guarantees ordering" is only true with the key that makes it true.

The alternative worth considering before paying for ordering at all: make consumers commutative or version-aware, so order stops mattering. Attaching a version or timestamp to each event and ignoring anything older than what you have already applied gives you correct final state under any delivery order — often far cheaper than serializing the pipeline.

Key Takeaways

• Ordering and parallelism are opposed; strict global order means exactly one consumer.
• Real systems need per-entity ordering, delivered by partitioning on an entity key.
• The partition key choice sets both the ordering scope and the maximum parallelism — too coarse kills throughput, too fine kills ordering.
• Retries reorder within a partition, so ordered consumers must halt on failure rather than skip; version-aware consumers avoid the problem entirely.

🧪 Practice

1. Pick partition keys for: bank transactions, chat messages, IoT sensor readings, page views. Justify the parallelism each allows.
2. Explain why a poison message stalls a Kafka partition but not a standard SQS queue, and what that implies for DLQ design.
3. Interview: You need ordered processing but one key produces 60% of traffic. What do you do? (Hint: either the key must be split, or the ordering requirement for that key must be re-examined.)

Competing Consumers Pattern

The competing consumers pattern is the standard way to scale queue processing: attach many consumer instances to one queue and let them compete for messages. The broker hands each message to exactly one of them, so throughput scales with consumer count without any coordination between consumers.

What makes it work is that the broker is already the coordinator. Each consumer simply asks for work; the broker's in-flight tracking guarantees that no two consumers get the same message at the same time. No leader election, no partition assignment, no shared state — which is why this is usually the first scaling pattern reached for and rarely the wrong one.

It also gives self-balancing load distribution for free. Consumers pull only when ready, so a slow consumer naturally takes fewer messages while a fast one takes more. Compare that with a push-based system, which must estimate consumer capacity and will overload the slow one.

 COMPETING CONSUMERS

              [ queue: m1 m2 m3 m4 m5 m6 m7 ]
                   /         |         \
                  v          v          v
            consumer1   consumer2   consumer3
            (fast)      (slow)      (fast)
              m1          m2          m3
              m4                      m5
              m6                      m7

  Fast consumers pull more. No coordination needed: PULL is the balancer.
  Add consumer4 -> it starts pulling immediately, no rebalance, no config.

# Scaling knob 1: prefetch. This single number controls fairness.
channel.basic_qos(prefetch_count=1)
# prefetch=1   -> a consumer holds one unacked message. Perfect fairness,
#                 highest per-message overhead. Right for long, variable jobs.
# prefetch=100 -> a consumer grabs 100 up front. High throughput, but a slow
#                 consumer HOARDS 100 messages while others idle, and its crash
#                 delays 100 messages by a full visibility timeout.
#
# Rule of thumb: long jobs (seconds+) -> low prefetch;
#                short jobs (millis)  -> high prefetch to amortize round trips.

# Scaling knob 2: how many consumers. Drive it from queue depth, not CPU.
def desired_consumers(queue_depth, arrival_rate, msgs_per_sec_per_consumer,
                      target_drain_seconds=300):
    # Enough capacity to keep up with arrivals...
    steady = arrival_rate / msgs_per_sec_per_consumer
    # ...plus enough to clear the existing backlog within the target window.
    catch_up = queue_depth / (msgs_per_sec_per_consumer * target_drain_seconds)
    return max(1, math.ceil(steady + catch_up))

print(desired_consumers(queue_depth=45_000, arrival_rate=120,
                        msgs_per_sec_per_consumer=8))
#: 34 consumers -> 15 to keep up, 19 to drain the backlog in 5 minutes.
#
# Queue depth is the correct auto-scaling signal for workers: it is the only
# metric that reflects both arrival rate AND service rate. CPU does not.

Three consequences of the pattern deserve attention.

Consumers must be idempotent. Competing consumers combined with at-least-once delivery means a message will occasionally be processed twice — after a timeout expires while the first consumer is still alive, for example. This is not an edge case; it is routine at scale.

Ordering is gone. As the previous topic showed, this is inherent. If you need order, you need partitioned consumers, which is a different pattern with different scaling properties.

The bottleneck moves. Twenty consumers hitting one database do not process work twenty times faster; they produce twenty times the database load. The queue's depth tells you whether you need more consumers; the downstream's saturation tells you whether more consumers will help.

Aspect	Competing consumers (queue)	Partitioned consumers (log)
Work assignment	Broker, per message	Static partition assignment
Adding a consumer	Instant, no rebalance	Triggers a rebalance
Max parallelism	Unbounded	Number of partitions
Ordering	None	Per partition
Slow consumer	Takes less work naturally	Its partition falls behind

Key Takeaways

• Many consumers on one queue scale throughput with zero coordination — the broker's in-flight tracking is the coordination.
• Pull-based delivery self-balances: fast consumers take more work, slow ones less.
• Prefetch trades fairness against throughput; use low prefetch for long jobs and high prefetch for short ones.
• Scale consumers on queue depth, and check that the downstream can absorb the added concurrency before adding it.

🧪 Practice

1. Compute the consumers needed for a 200,000-message backlog, 500 msg/s arrival, 12 msg/s per consumer, drained within ten minutes.
2. Explain the exact harm of prefetch_count=500 for jobs averaging 45 seconds each.
3. Interview: You triple your consumers and throughput improves by 10%. What happened? (Hint: the queue was not the constraint — what did all three times as many consumers start doing simultaneously?)

<a id="72-event-streaming"></a>

7.2 Event Streaming

A queue delivers a message and forgets it. A stream keeps every message in an ordered, replayable log. That single change — retention instead of deletion — produces a different set of capabilities and a different set of problems.

Log-Based Streaming Architecture

Traditional brokers treat a message as a task in transit: it arrives, it is delivered, it is deleted. Log-based streaming systems treat a message as a fact appended to a permanent record. Consumers read from a position in that record and advance their own pointer; nothing is removed because someone read it.

The reason this matters is that deletion destroys information you frequently need later. With a queue, once the analytics consumer has processed an event, it is gone — so a bug in the analytics code means the data is unrecoverable, a new service cannot learn what happened before it existed, and rebuilding a corrupted database means asking every producer to resend history nobody kept.

The analogy is the difference between a mailbox and a ledger. Mail is removed when read; each letter has one reader and one life. A ledger is appended to and never erased; any number of readers can start at any page, read at their own pace, and re-read last year's entries whenever they need to.

 QUEUE (destructive read)            LOG (positional read)

  [ m1 m2 m3 ]                        offset:  0    1    2    3    4    5
     |                                        [e0] [e1] [e2] [e3] [e4] [e5]
     v  consumer reads m1                              ^         ^
  [ m2 m3 ]     m1 is GONE                             |         |
                                             analytics(2)   billing(4)
  One consumer, one chance.
  Bug in the consumer = data lost.    Independent positions. Analytics can be
                                      reset to 0 and replay everything without
                                      affecting billing at all.

Three structural properties follow from "append-only, never delete":

Sequential I/O. Writes only ever append to the end of a file, and reads walk forward through it. Sequential disk access is orders of magnitude faster than random access, which is why a log broker on spinning disks can outrun an in-memory broker doing random deletes. There is no per-message index to update, no tombstone to write, no compaction of a mutable structure on the write path.

Zero-copy delivery. Because the on-disk format and the wire format are the same bytes, the broker can hand a range of the file directly to the network card without ever copying it into application memory.

Consumer independence. The consumer's position is the only per-consumer state, and it is just a number. Ten consumers reading the same data cost the broker almost nothing extra, because there is only one copy of the data and ten integers.

# The mental model of the log, in about fifteen lines.
class Log:
    def __init__(self):
        self.records = []                 # append-only; nothing is ever removed
        self.positions = {}               # consumer_group -> next offset to read

    def append(self, record):
        self.records.append(record)
        return len(self.records) - 1      # the offset IS the identity

    def read(self, group, max_records=10):
        start = self.positions.get(group, 0)
        batch = self.records[start:start + max_records]
        return start, batch               # NOTE: reading does not advance

    def commit(self, group, offset):
        # The consumer decides when it has durably handled up to `offset`.
        # This separation is what makes replay possible: commit is a choice,
        # not a side effect of reading.
        self.positions[group] = offset

    def reset(self, group, offset=0):
        self.positions[group] = offset    # replay: just move the pointer back

The architectural consequence is larger than a performance improvement. Because the log retains history, it can serve as the system of record for state changes rather than a transport for notifications. A database becomes a materialized view of the log, derivable at any time by replaying it. That is the foundation of event sourcing, change data capture, and stream processing, all of which depend on the log being the durable truth rather than a pipe.

Property	Queue broker	Log-based stream
Read effect	Removes the message	Advances a position
Retention	Until consumed	Time or size based
Replay	Impossible	Move the offset
Adding a new consumer	Sees only new messages	Can start from the beginning
Per-consumer cost	A copy of the message	One integer
Access pattern	Random (delete anywhere)	Sequential append and scan

Key Takeaways

• A log retains messages after they are read; consumers track a position rather than consuming destructively.
• Append-only sequential I/O and zero-copy transfer are what make log brokers extremely fast on ordinary disks.
• Multiple consumers are nearly free because per-consumer state is a single offset.
• Retention turns the log into a system of record — replayable, and able to serve consumers that did not exist when the events were written.

🧪 Practice

1. Name three capabilities a log has that a queue cannot provide, and one thing a queue does better.
2. Extend the Log class with time-based retention and explain what breaks for a consumer that falls behind.
3. Interview: Why can a log broker be faster than an in-memory queue broker? (Hint: compare the disk access pattern, not the storage medium.)

Topics, Partitions, and Offsets

Three concepts define how a log is organized, and together they explain both its scalability and its constraints.

A topic is a named stream of related records — order-events, page-views. It is a logical grouping, not a physical one.

A partition is the physical unit: an independent, ordered, append-only log file. A topic is split into partitions, and this split is the source of all parallelism in the system. Each partition lives on a broker (with replicas elsewhere), is written to independently, and is consumed by at most one member of a consumer group.

An offset is a record's position within its partition — a monotonically increasing integer that never changes and is never reused. Offsets are meaningful only within a partition; offset 500 in partition 0 has no relationship to offset 500 in partition 1.

 TOPIC "order-events" WITH 3 PARTITIONS

  partition 0:  [0][1][2][3][4][5]   <- append here
  partition 1:  [0][1][2]            <- append here
  partition 2:  [0][1][2][3][4]      <- append here
                 ^        ^
                 |        consumer-group-B committed offset 3
                 consumer-group-A committed offset 0

  * Order is guaranteed WITHIN a partition, never across partitions.
  * Offsets are per-partition; (partition, offset) is the unique address.
  * Parallelism within a consumer group is capped at 3 -- the partition count.

The producer decides the partition, and how it decides is the most consequential choice in the design:

# 1. Keyed: hash(key) % partitions -> same key ALWAYS lands in the same
#    partition, which is what gives per-entity ordering.
producer.send("order-events", key=b"order-991", value=payload)

# 2. Keyless: round-robin / sticky batching across partitions.
#    Maximum spread, zero ordering guarantee.
producer.send("order-events", value=payload)

# 3. Explicit: you choose. Useful for co-partitioning two topics so a stream
#    join can be done locally without a network shuffle.
producer.send("order-events", partition=2, value=payload)

Partition count is a capacity decision that is painful to change later, because increasing it re-maps keys: hash(k) % 6 is a different partition than hash(k) % 3, so existing keys move and their historical ordering is broken across the boundary. The practical guidance is to over-provision modestly at creation time.

# Sizing partitions: take the maximum of the throughput and parallelism needs.
def partition_count(target_mb_s, per_partition_mb_s,
                    peak_consumer_instances, growth=2.0):
    by_throughput = target_mb_s / per_partition_mb_s
    by_parallelism = peak_consumer_instances   # a consumer group cannot exceed
                                               # one consumer per partition
    return math.ceil(max(by_throughput, by_parallelism) * growth)

print(partition_count(target_mb_s=180, per_partition_mb_s=10,
                      peak_consumer_instances=24))
# 48 -> 18 needed for throughput, 24 for parallelism, doubled for growth.
#
# Do not simply pick a huge number: every partition costs open file handles,
# memory for buffers, replication traffic, and leader-election time on failure.

Offsets carry one subtlety that causes real bugs. The committed offset is the position a consumer group has durably recorded as processed, and it is normally stored as "the next offset to read", not "the last offset handled". Committing before processing means a crash skips records (at-most-once); committing after processing means a crash reprocesses them (at-least-once). There is no third option that a commit alone can give you.

Term	Meaning
Log end offset	The next offset to be written by a producer
Committed offset	Where a consumer group will resume
Consumer lag	Log end offset minus committed offset
Retention bound	The earliest offset still available to read

Consumer lag is the single most important metric in any streaming system. It is the exact analogue of queue depth: it measures whether consumers are keeping up, in units of records not yet processed. Alert on lag trending upward, not merely on an absolute value, because a steadily rising lag predicts an outage long before the number itself looks alarming.

Key Takeaways

• A topic is logical; a partition is the physical, ordered log and the unit of parallelism and ordering.
• Offsets are per-partition immutable positions; (partition, offset) uniquely addresses a record.
• Partition count caps consumer-group parallelism and is disruptive to change, because it remaps keys — size it with headroom.
• Consumer lag is the streaming equivalent of queue depth and is the primary health signal.

🧪 Practice

1. A topic needs 250 MB/s, each partition sustains 12 MB/s, and peak consumers are 30. How many partitions, and which constraint dominates?
2. Explain precisely what breaks for keyed ordering when partitions go from 4 to 8, and describe a migration that avoids it.
3. Interview: A consumer group has 12 consumers and the topic has 8 partitions. What is throughput, and what are the other consumers doing? (Hint: what is the maximum number of consumers per partition within one group?)

Consumer Groups and Rebalancing

A consumer group is a set of consumer instances that cooperate to read a topic. The group's defining rule is that each partition is assigned to exactly one member, so every record is processed once by the group as a whole while work is spread across its members.

This gives you both delivery models from one mechanism. Multiple instances in the same group is competing consumers — the work is divided. Multiple groups reading the same topic is publish-subscribe — each group gets every record, independently, with its own offsets. A single topic can feed a billing group, an analytics group, and a search-indexing group, none aware of the others.

 ONE TOPIC, TWO GROUPS

  topic (4 partitions)     group "billing" (2)     group "analytics" (4)
     p0 ------------------> consumer-A ----------> consumer-W
     p1 ------------------> consumer-A ----------> consumer-X
     p2 ------------------> consumer-B ----------> consumer-Y
     p3 ------------------> consumer-B ----------> consumer-Z

  Within a group: work is DIVIDED (each partition to one member).
  Across groups: work is DUPLICATED (each group sees everything).
  Groups keep separate offsets, so analytics can replay without touching
  billing's position.

Rebalancing is the process of reassigning partitions when group membership changes — an instance joins, leaves, crashes, or is deemed dead. It is the mechanism that makes the group self-healing, and it is also the primary source of operational pain in streaming systems.

The pain comes from the classic protocol being stop-the-world: every member revokes its partitions, the group coordinator computes a new assignment, and every member resumes. During that window, no member is processing anything. If rebalances are frequent — a deploy, an autoscaling event, a long GC pause — the group can spend more time rebalancing than working.

 STOP-THE-WORLD (eager) vs COOPERATIVE (incremental) REBALANCE

 EAGER:   all consumers revoke ALL partitions
          |<---------- everything stopped ---------->|
          reassign -> resume       (seconds of zero throughput)

 COOPERATIVE: only the partitions that actually MOVE are revoked
          p0,p1 keep running --------------------------->
                        p2 revoked -> reassigned -> resumes
          (throughput dips slightly instead of stopping)

# The settings that determine rebalance frequency and duration.
consumer = KafkaConsumer(
    "order-events",
    group_id="billing",
    partition_assignment_strategy=[CooperativeStickyAssignor],
    # Cooperative: revoke only what moves. Sticky: keep existing assignments
    # where possible, so local state and caches stay warm.

    session_timeout_ms=45_000,       # how long the coordinator waits for a
                                     # heartbeat before declaring us dead
    heartbeat_interval_ms=3_000,     # ~1/3 of the session timeout
    max_poll_interval_ms=300_000,    # MAX TIME BETWEEN poll() CALLS.
                                     # This is the one that bites: if handling
                                     # a batch takes longer than this, the
                                     # coordinator assumes we hung and kicks
                                     # us out -- triggering a rebalance, which
                                     # slows everyone, which causes more
                                     # timeouts. A classic feedback loop.
    max_poll_records=100,            # bound the batch so processing time stays
                                     # comfortably under max_poll_interval_ms
    enable_auto_commit=False,        # commit explicitly, after processing
)

Two refinements are worth knowing because they eliminate most unnecessary rebalances.

Static group membership gives each instance a stable group.instance.id. When a pod restarts during a routine deploy, it rejoins with the same identity and reclaims its previous partitions without a rebalance, as long as it returns within the session timeout. This turns a rolling deploy from N rebalances into zero.

Decoupling polling from processing protects max.poll.interval.ms. Poll on one thread, hand work to a bounded executor, and pause/resume partitions to apply backpressure. The consumer keeps heartbeating while the work proceeds.

# Long-running handlers: never block the poll loop.
records = consumer.poll(timeout_ms=1000)
if backlog_full():
    consumer.pause(*consumer.assignment())   # stop fetching, KEEP heartbeating
else:
    consumer.resume(*consumer.assignment())
submit_to_worker_pool(records)               # process off the poll thread

Key Takeaways

• Within a group, partitions are divided among members; across groups, every group receives everything with independent offsets.
• Rebalancing reassigns partitions on membership change and is the main source of stalls in streaming systems.
• max.poll.interval.ms exceeded by slow processing is the most common cause of a rebalance storm — bound batch size or process off the poll thread.
• Cooperative sticky assignment plus static membership removes most rebalances entirely, including those from rolling deploys.

🧪 Practice

1. A group of 6 consumers reads an 18-partition topic. Two crash. Describe the assignment before, during, and after the rebalance.
2. Handling a batch takes 6 minutes and max.poll.interval.ms is 300,000. Explain the failure loop and give two independent fixes.
3. Interview: How would you deploy a 20-instance consumer group with no processing gap? (Hint: what does the coordinator need to believe about an instance that is restarting?)

Stream Retention and Compaction

Because a log never deletes on read, something else must decide when records go away. Retention policy is that decision, and there are two fundamentally different kinds.

Time or size based retention deletes whole segments once they exceed an age or the partition exceeds a size. This treats records as a stream of events — things that happened at a time and become irrelevant afterwards. Keep seven days of page views; nobody needs the ones from March.

Log compaction takes a different view: it retains the most recent record per key, forever, deleting only superseded versions. This treats the log as a stream of state changes, where the latest value for each key is the current truth. A compacted topic is a durable, replayable key-value snapshot that also happens to be a change feed.

 BEFORE COMPACTION (offsets preserved, keys repeated)

  off: 0      1      2      3      4      5      6
  key: u1     u2     u1     u3     u2     u1     u4
  val: A      B      A'     C      B'     A''    D

 AFTER COMPACTION

  off: 3      4      5      6
  key: u3     u2     u1     u4
  val: C      B'     A''    D

  * The LATEST value for every key survives; earlier versions are removed.
  * Offsets are NOT renumbered -- gaps are normal and expected.
  * Replaying a compacted topic from 0 reconstructs the full current state.

The practical power of compaction is that it makes the log a valid source for rebuilding state from scratch. A new service, a wiped cache, or a corrupted materialized view can be repopulated by replaying the compacted topic from the beginning: bounded by the number of distinct keys rather than the total history. Kafka uses this internally for consumer offsets — the __consumer_offsets topic is compacted, so only the newest offset per group and partition is kept.

# Event stream: keep a week, then discard.
kafka-topics.sh --create --topic page-views \
  --config retention.ms=604800000 \
  --config cleanup.policy=delete

# State stream: keep the latest value per key forever.
kafka-topics.sh --create --topic user-profiles \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5 \
  --config delete.retention.ms=86400000   # how long tombstones survive, so
                                          # offline consumers can still see
                                          # the deletion when they return

# Both: compact, but drop anything older than 30 days regardless.
kafka-topics.sh --create --topic account-state \
  --config cleanup.policy=compact,delete \
  --config retention.ms=2592000000

Deletion in a compacted topic requires an explicit tombstone: a record with the key and a null value. Compaction keeps the tombstone for delete.retention.ms so that consumers replaying later observe the removal, and then removes both it and the key's history. Publishing nothing does not delete a key — absence is indistinguishable from "no update".

# Deleting a key from a compacted topic.
producer.send("user-profiles", key=b"u-42", value=None)   # tombstone
# A consumer replaying sees: ... u-42=<data> ... u-42=None
# and must interpret None as "this key was deleted", not as a malformed record.

Policy	Keeps	Bounded by	Use for
delete (time/size)	Everything within the window	Time or bytes	Events, metrics, logs
compact	Latest value per key	Number of distinct keys	Entity state, config, offsets
compact,delete	Latest per key, within window	Both	State with a regulatory max age

Two operational cautions. First, compaction is asynchronous and never instantaneous: a consumer reading the tail sees every version, including duplicates that will later be compacted away. Compaction guarantees eventual convergence, not a de-duplicated read. Second, a compacted topic's size is driven by key cardinality, so a high-cardinality key (a UUID per event) makes compaction useless — nothing is ever superseded, and the topic grows without bound.

Key Takeaways

• Time/size retention treats records as events to expire; compaction treats them as state and keeps the latest value per key.
• A compacted topic can rebuild full current state on replay, bounded by distinct key count rather than total history.
• Deletion requires an explicit null-valued tombstone, retained long enough for lagging consumers to observe it.
• Compaction is asynchronous — tail consumers still see superseded versions — and is useless for high-cardinality keys.

🧪 Practice

1. Choose a policy and settings for: audit logs (7-year legal hold), current inventory levels, IoT telemetry, feature flags. Justify each.
2. Explain why a compacted topic keyed by event_id never shrinks.
3. Interview: How would you rebuild a corrupted read model with zero downtime using a compacted topic? (Hint: build the new one alongside the old and switch only when it has caught up.)

Kafka vs. Traditional Brokers

The choice between a log-based system and a traditional message broker is not a matter of one being newer or faster. They optimize for different things, and picking the wrong one produces either unnecessary complexity or missing capability.

Traditional brokers (RabbitMQ, ActiveMQ, and the AMQP family) are built around flexible routing and per-message handling. The broker maintains per-message state, supports rich topologies through exchanges and bindings, can route on header values, can requeue a single message, and offers per-message TTLs, priorities, and delayed delivery. The unit of thought is the individual message.

Log brokers (Kafka, Pulsar, Redpanda) are built around high-throughput ordered retention. Routing is minimal — a partition key and nothing else — but throughput is enormous, history is retained, and any number of consumers can read independently. The unit of thought is the stream.

 RABBITMQ: routing is the feature

   producer -> [ exchange ] --binding: "order.*.eu"--> [ queue A ]
                    |      --binding: "order.high.*"-> [ queue B ]
                    |      --binding: header x-vip=1-> [ queue C ]
                    v
             smart broker, per-message decisions, per-message state

 KAFKA: retention and throughput are the feature

   producer -> [ topic: partition by key ] -> retained 7 days
                     |          |         |
                consumer-gp-A  gp-B     gp-C   (independent offsets, replay)

             dumb broker, smart consumers, no per-message state

Dimension	Traditional broker (RabbitMQ)	Log broker (Kafka)
Throughput	Tens of thousands msg/s	Millions msg/s
After consumption	Message deleted	Retained until policy expires
Replay	No	Yes, reset the offset
Routing	Rich (topic, header, fanout)	Partition key only
Per-message TTL	Yes	No (retention is per topic)
Priority queues	Yes	No
Delayed delivery	Yes (plugin/TTL + DLX)	No (needs an external scheduler)
Selective ack	Yes, any message	No, offsets advance in order
Ordering	Per queue	Per partition
Consumer scaling limit	Unbounded	Partition count
Operational weight	Moderate	Higher (though improving)

The decision usually reduces to four questions:

1. Do you need history or replay? Only the log can give it. This alone decides many cases.
2. Do you need per-message routing, priority, delay, or selective retry? Only the traditional broker does these natively; on Kafka each is a project.
3. What is the throughput? Below roughly 50k msg/s, both work comfortably; above it, the log pulls decisively ahead.
4. How many independent consumers will read the same data? Every additional consumer group is nearly free on a log and is another queue and another copy on a traditional broker.

# The capability that has no traditional-broker equivalent.
# Reprocessing three days of history through fixed code:
consumer = KafkaConsumer(group_id="billing-v2", enable_auto_commit=False)
partitions = [TopicPartition("order-events", p) for p in range(8)]
consumer.assign(partitions)

target_ms = int((time.time() - 3 * 86400) * 1000)
offsets = consumer.offsets_for_times({tp: target_ms for tp in partitions})
for tp, meta in offsets.items():
    consumer.seek(tp, meta.offset)      # rewind to a POINT IN TIME
# Now reprocess with the corrected logic. A queue-based system would have to
# ask every producer to resend history that nobody retained.

A pragmatic note: these are not mutually exclusive, and many mature systems run both — Kafka as the event backbone carrying state changes and analytics, and RabbitMQ or SQS for task queues that need priorities, delays, and per-message retry. Choosing "one messaging system for everything" is a common source of awkward workarounds in both directions.

Key Takeaways

• Traditional brokers optimize for routing flexibility and per-message control; log brokers optimize for throughput, retention, and replay.
• Replay, history, and cheap additional consumers are log-only capabilities.
• Priority, per-message TTL, delayed delivery, and selective ack are traditional-broker capabilities that require extra machinery on a log.
• Running both for different workloads is normal and usually cheaper than forcing one to do everything.

🧪 Practice

1. Choose a broker for: password-reset emails, clickstream ingestion, video transcoding jobs, CDC from a database. Justify each in one sentence.
2. Describe how you would implement delayed delivery on a log broker and what it costs.
3. Interview: When is a log broker the wrong choice? (Hint: think about a workload where one slow message must not hold up the ones behind it.)

Event Replay

Event replay is reprocessing records that have already been consumed. It is the capability that most distinguishes a log from a queue, and it turns several otherwise-expensive operations into routine ones.

The reason it is so valuable is that it changes what a bug costs. In a queue-based system, a defect in a consumer means the affected data is gone — recovery requires reconstructing it from other sources, if that is even possible. With replay, the fix is: deploy corrected code, rewind the offset, let it run. The log holds the truth; derived state is disposable.

Four situations call for it:

Situation	What you do
Bug fix	Rewind to before the bug and reprocess with new code
New consumer	Start at offset 0 and build state from full history
Rebuilding a store	Wipe a corrupted read model and regenerate it
Testing	Replay real production traffic into a staging system

 REPLAY WITH A PARALLEL CONSUMER GROUP (the safe pattern)

  topic: [==================== history ====================>]
              ^                                          ^
              |                                          |
    group "billing-v2" starts here               group "billing-v1"
    (rewound, reprocessing with the fix)         (still live, still serving)

  v2 catches up to v1's position, output is compared, then traffic is
  switched. v1 is never interrupted, and a bad v2 is discarded harmlessly.

# Three ways to position a consumer for replay.
# 1. From the beginning: full history.
consumer.seek_to_beginning(*partitions)

# 2. From a specific offset, per partition.
consumer.seek(TopicPartition("order-events", 0), 1_500_000)

# 3. From a timestamp -- almost always what you actually want, because you
#    know WHEN the bug was deployed, not which offset that corresponds to.
bug_deployed_ms = 1_724_500_000_000
offsets = consumer.offsets_for_times({tp: bug_deployed_ms for tp in partitions})
for tp, meta in offsets.items():
    consumer.seek(tp, meta.offset if meta else 0)

Replay is powerful precisely because it re-executes real logic, which is also why it is dangerous. Four hazards must be handled deliberately:

Side effects re-fire. Replaying OrderPlaced through a consumer that sends email will send every email again. Consumers intended to be replayable must either be side-effect free or guard external calls with idempotency keys — and during a replay, non-idempotent side effects should be disabled outright.

def handle(event, replaying=False):
    update_read_model(event)                 # idempotent: safe to repeat
    if not replaying:
        send_email(event)                    # NOT idempotent: suppress on replay
    charge_card(event, idempotency_key=event["event_id"])   # guarded: safe

Downstream load multiplies. Replaying three days of events in ten minutes compresses three days of write load into ten minutes. Rate-limit the replay consumer, or it becomes an outage of the database it is writing to.

Old events meet new code. A record written eighteen months ago has the schema of eighteen months ago. Replay is the operation that punishes schema carelessness most severely, which is why consumers should tolerate missing fields and unknown fields rather than assume the current shape.

Time semantics shift. Code that uses now() will compute different results during replay than it did originally. Anything time-dependent must read the event's own timestamp, not the wall clock.

The safest procedure is the parallel-group pattern shown above: run the replay as a new consumer group writing to a new destination, compare its output against the live one, and switch over only when it has caught up and matches. Never rewind a live consumer group in production as a first move — that is a change you cannot undo, made under the pressure of an incident.

Key Takeaways

• Replay makes derived state disposable and rebuildable, which changes a consumer bug from data loss into a redeploy.
• Position by timestamp rather than raw offset: you know when the bug shipped, not which offset that was.
• Guard side effects — suppress non-idempotent ones during replay and use idempotency keys for the rest.
• Rate-limit replays and run them as a parallel group writing to a new destination, rather than rewinding a live group.

🧪 Practice

1. List every side effect in a typical order-processing consumer and classify each as replay-safe, guardable, or must-suppress.
2. Replaying 30 days at full speed would produce 400x normal write load. Design the throttling.
3. Interview: You must replay six months of events, but the schema changed three times. How do you proceed? (Hint: something has to know how to read the old shapes — either the consumer or a step before it.)

<a id="73-delivery-semantics"></a>

7.3 Delivery Semantics

Every messaging system must answer one question: if something fails between sending and processing, does the message get lost or repeated? There is no third option, and the whole of this subchapter follows from that fact.

At-Most-Once Delivery

At-most-once delivery guarantees that a message is processed zero or one times — never more. Duplicates are impossible; loss is possible.

The reason such a weak guarantee exists is that it is the cheapest and fastest one available. There are no retries, no acknowledgment round trips, no stored delivery state, and no dedup bookkeeping. The producer sends and moves on; the consumer acknowledges on receipt and processes afterwards. If anything fails in between, the message is simply gone.

You get at-most-once by acknowledging before doing the work — which is exactly what auto-acknowledge mode does, and why it is a bug in most systems and a deliberate choice in a few.

 AT-MOST-ONCE: ack first, work second

  broker ---- message ----> consumer
  broker <--- ack ---------|            message deleted NOW
                           |
                           v  process()
                           X  crash here -> the work never happened, and
                                            nobody will ever know

# At-most-once, deliberately: the ack precedes the work.
channel.basic_consume(queue="metrics", on_message_callback=handle, auto_ack=True)

def handle(ch, method, properties, body):
    # The broker already considers this delivered. A crash here loses it.
    metrics_store.record(json.loads(body))

# Producer side: fire and forget, no confirmation waited for.
producer.send("metrics", value=payload)   # note: no .get(), no callback,
                                          # no acks -- send and continue

This is the right choice only when the cost of a duplicate exceeds the cost of a loss, or when losses are self-correcting. In practice that means:

Workload	Why at-most-once fits
High-frequency metrics	The next sample arrives in a second; one gap is invisible
Live sensor telemetry	A stale reading is worse than a missing one
Cache-warming hints	A miss is handled by the normal cache-miss path
Debug/trace sampling	Already sampled; losing more changes nothing
Non-critical UI notifications	A duplicate notification is more annoying than a missed one

The configuration that produces it spans producer and consumer, and both halves must agree:

# Producer: acks=0 means "do not wait for the broker to confirm anything".
producer = KafkaProducer(
    acks=0,          # no confirmation; a broker restart loses in-flight records
    retries=0,       # a retry could produce a duplicate -- forbid it
    linger_ms=100,   # batch aggressively; throughput is the whole point here
)

# Consumer: commit the offset BEFORE processing.
records = consumer.poll()
consumer.commit()                 # committed first -> a crash below skips these
for record in records:
    process(record)               # if this fails, the record is not retried

The honest framing is that at-most-once is a decision to lose data under failure in exchange for throughput and simplicity. That is defensible for a telemetry pipeline and indefensible for a payment. The failure mode to guard against is choosing it by accident — auto_ack=True and acks=0 are both defaults or one-word settings in various clients, and neither announces that it has traded away durability.

Key Takeaways

• At-most-once means zero or one processing: no duplicates, possible loss.
• It is produced by acknowledging or committing before the work is done.
• It is the fastest and simplest option, appropriate only when a lost message is cheaper than a repeated one or is self-correcting.
• Both producer (acks=0, no retries) and consumer (commit first) must be configured for it — and it is easy to select unintentionally.

🧪 Practice

1. Give two workloads where at-most-once is correct and two where it would be a serious defect. Explain the deciding factor.
2. Identify the specific line in a consumer that turns at-least-once into at-most-once, and describe the failure window it creates.
3. Interview: When is a duplicate worse than a loss? (Hint: think about actions whose repetition is visible to a person or moves money.)

At-Least-Once Delivery

At-least-once delivery guarantees a message is processed one or more times — it is never lost, but it may repeat. This is the default of nearly every messaging system in production, because for most work, doing something twice is recoverable and never doing it is not.

The mechanism is the mirror image of at-most-once: acknowledge after the work succeeds, and retry anything unacknowledged. Since an acknowledgment can be lost in flight, or a consumer can crash in the window between finishing the work and sending the ack, the broker sometimes redelivers work that was already done. It has no way to distinguish "not done" from "done but unconfirmed".

 THE UNAVOIDABLE DUPLICATE WINDOW

  broker ---- message ----> consumer
                            |
                            v  process()   <- SUCCEEDS, side effects applied
                            |
                            |  ack ------> X  network drops the ack
                            X  or consumer crashes right here
                            |
  broker: no ack received -> redeliver -> the work runs a SECOND time

  The broker cannot tell these apart:
    (a) consumer died before doing the work   -> must redeliver
    (b) consumer died after doing the work    -> must not redeliver
  Since it cannot know, it chooses (a). That choice IS at-least-once.

# At-least-once on both sides.
producer = KafkaProducer(
    acks="all",              # wait for all in-sync replicas to persist
    retries=10,              # retry on failure -- this is where duplicates
                             # are born: a retried send whose first attempt
                             # actually succeeded produces two records
    max_in_flight_requests_per_connection=5,
)

consumer = KafkaConsumer(group_id="billing", enable_auto_commit=False)
for record in consumer:
    process(record)                 # work FIRST
    consumer.commit()               # commit only after success
    # A crash between these two lines means `process` runs again on restart.

The essential consequence is that at-least-once shifts the burden to the consumer: since duplicates are guaranteed to occur eventually, the consumer must be built so that repeating an operation is harmless. That property is idempotency, and the next topic covers how to obtain it.

It helps to know where duplicates come from, because each source has a different frequency and a different mitigation:

Source	Trigger	Frequency
Producer retry after a lost ack	Network blip during send	Common under load
Consumer crash before commit	Deploy, OOM, node failure	Every deploy
Visibility timeout expiring early	Job slower than the configured timeout	Constant if misconfigured
Rebalance mid-batch	Scaling, GC pause, max.poll.interval hit	Common
Deliberate redrive from a DLQ	Operator replays messages	Occasional

# Estimating how often duplicates actually occur -- worth doing, because the
# answer determines how much you should invest in dedup infrastructure.
def duplicate_rate(msgs_per_day, deploys_per_day, consumers,
                   in_flight_per_consumer, crash_rate_per_day=0.02):
    # Each deploy interrupts every consumer's in-flight batch.
    from_deploys = deploys_per_day * consumers * in_flight_per_consumer
    from_crashes = crash_rate_per_day * consumers * in_flight_per_consumer
    total = from_deploys + from_crashes
    return total, total / msgs_per_day

dupes, rate = duplicate_rate(msgs_per_day=40_000_000, deploys_per_day=6,
                             consumers=30, in_flight_per_consumer=100)
print(f"{dupes:.0f} duplicates/day ({rate:.6%} of messages)")
# 1860 duplicates/day (0.004650% of messages)
#
# A tiny percentage -- and 1,860 double-charged customers per day. The rate is
# irrelevant; the absolute number and the per-event cost are what matter.

That calculation is the point of the topic. "Duplicates are rare" is not a design position, because rare multiplied by high volume is a daily incident. At any meaningful scale, at-least-once without idempotent consumers is not a system that occasionally misbehaves — it is a system that misbehaves on schedule.

Key Takeaways

• At-least-once never loses messages but guarantees occasional duplicates.
• It arises from acknowledging after processing: the broker cannot distinguish "not done" from "done but unconfirmed".
• Duplicates come from producer retries, consumer crashes, expired visibility timeouts, and rebalances — deploys alone guarantee them.
• The guarantee is only usable if consumers are idempotent; compute the absolute daily duplicate count, not the percentage.

🧪 Practice

1. For a system at 500 msg/s with 12 daily deploys across 20 consumers holding 50 in-flight each, estimate the daily duplicate count.
2. For each duplicate source in the table, state the mitigation that reduces (not eliminates) it.
3. Interview: A consumer commits its offset immediately after poll() "to avoid duplicates". What has the team actually chosen? (Hint: name the guarantee, then name what they now lose.)

Exactly-Once Semantics

Exactly-once is the guarantee everyone wants: each message is processed once, with no loss and no duplication. It is also the most misunderstood topic in messaging, because in the general case exactly-once delivery is impossible, while exactly-once processing is achievable under specific conditions.

The impossibility is not an engineering limitation. It is the Two Generals Problem: a sender and receiver communicating over a lossy channel can never both be certain that a message was received. The sender that gets no acknowledgment cannot know whether the message or the acknowledgment was lost, and must choose between resending (risking a duplicate) and not resending (risking a loss). No protocol removes that choice, because any confirmation can itself be lost.

 WHY DELIVERY CANNOT BE EXACTLY-ONCE

  sender                     receiver
    |--- message ----------->|   received
    |<--- ack --------X      |   ack lost in transit
    |                        |
  sender's dilemma: was the message delivered?
    resend -> possible duplicate     don't resend -> possible loss
  Add an "ack of the ack" and the same problem simply moves one hop later.

What is achievable is exactly-once effect: at-least-once delivery combined with deduplication or idempotency, so that repeated delivery produces the same end state as a single delivery. The message may arrive five times; the effect occurs once. Every real "exactly-once" system works this way, and knowing that tells you what to check when a vendor claims it.

There are three mechanisms, in increasing order of scope:

1. Idempotent producer. Each producer gets an ID and each record a sequence number, so a broker receiving a duplicate from a retried send recognizes and discards it. This removes producer-side duplicates only.

2. Transactional writes. The offset commit and the output records are written in a single atomic transaction, so a consume-transform-produce pipeline either commits both or neither. This is what Kafka means by exactly-once semantics, and its scope is strictly Kafka-to-Kafka.

3. Idempotent consumer with a shared store. The consumer writes its business change and its processing marker in one transaction of the destination database. This is the only mechanism that extends exactly-once to systems outside the broker.

# Kafka transactional read-process-write: atomic across the pipeline.
producer = KafkaProducer(
    transactional_id="enricher-1",   # stable per instance; enables fencing so
    enable_idempotence=True,          # a zombie old instance cannot write
)
producer.init_transactions()

consumer = KafkaConsumer(
    "raw-events", group_id="enricher",
    isolation_level="read_committed", # never see records from an aborted txn
    enable_auto_commit=False,         # offsets are committed IN the transaction
)

for batch in consumer:
    producer.begin_transaction()
    try:
        for record in batch:
            producer.send("enriched-events", enrich(record))
        # The critical line: offsets are part of the SAME transaction as the
        # output records. Either both are visible or neither is.
        producer.send_offsets_to_transaction(offsets_of(batch), "enricher")
        producer.commit_transaction()
    except Exception:
        producer.abort_transaction()   # output records never become visible

# The general case: exactly-once EFFECT via the destination database.
# This works with any broker, and extends the guarantee outside Kafka.
def handle(message):
    with db.transaction():                    # ONE transaction, TWO writes
        already = db.execute(
            "INSERT INTO processed_messages (message_id) VALUES (?) "
            "ON CONFLICT DO NOTHING RETURNING 1", message.id)
        if not already:
            return                            # duplicate: the effect already
                                              # happened, do nothing
        db.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?",
                   message.amount, message.account_id)
    # Atomicity is what makes this correct. If the marker and the business
    # change could commit separately, a crash between them reintroduces
    # either a duplicate or a lost update.

Scope of guarantee	Mechanism	Works across systems?
No producer-side duplicates	Idempotent producer	Broker-internal
Atomic consume-transform-produce	Broker transactions	Same broker only
Exactly-once business effect	Idempotent consumer + shared txn	Yes
Exactly-once delivery	None — provably impossible	—

The costs are real: transactions reduce throughput (commonly 20-50% on Kafka depending on batch size), add latency because read_committed consumers wait for transaction boundaries, and require careful handling of transaction timeouts. The correct question is therefore not "can we have exactly-once?" but "which specific effects must not be duplicated?" — and then applying idempotency there rather than transactional machinery everywhere.

Key Takeaways

• Exactly-once delivery is impossible (Two Generals); exactly-once effect is achievable and is what all real systems provide.
• Idempotent producers remove send-retry duplicates; broker transactions make consume-transform-produce atomic within one broker.
• Only an idempotent consumer writing its marker and its business change in one destination transaction extends the guarantee to external systems.
• Transactions cost throughput and latency — apply them to the effects that genuinely require them, not to the whole pipeline.

🧪 Practice

1. Explain in two sentences why an "ack of the ack" does not solve the Two Generals Problem.
2. A pipeline reads Kafka, writes to Postgres, and calls a payment API. Which parts can be exactly-once, and how do you handle the part that cannot?
3. Interview: A vendor advertises exactly-once delivery. What do you ask? (Hint: find the boundary of the claim — between which two components, and what happens at the edge of that boundary?)

Idempotent Consumers

An operation is idempotent when performing it multiple times has the same effect as performing it once. Since at-least-once delivery guarantees eventual duplicates, idempotency is what converts an unavoidable delivery weakness into a non-issue — it is the practical answer to exactly-once.

The key realization is that idempotency is a property of the operation, not of the messaging system. Many operations are naturally idempotent and need no machinery at all, and recognizing them saves a great deal of work.

# Naturally idempotent -- absolute assignment.
db.execute("UPDATE users SET email = ? WHERE id = ?", email, uid)
cache.set(key, value)
s3.put_object(Bucket=b, Key=k, Body=data)
db.execute("INSERT INTO x (id, ...) VALUES (?, ...) ON CONFLICT DO NOTHING", id)

# NOT idempotent -- relative modification or external side effect.
db.execute("UPDATE accounts SET balance = balance - 50 WHERE id = ?", uid)
db.execute("INSERT INTO orders (...) VALUES (...)")     # a second row appears
send_email(user)                                        # a second email arrives
queue.publish(downstream_event)                         # fans out twice
counter.increment()

The pattern is clear: setting a value to an absolute state is idempotent; modifying relative to the current state is not. Where you can express an operation as "make the world look like this" instead of "change the world by this much", you get idempotency for free.

When you cannot, there are three general techniques.

1. Deduplication by message identity. Record which message IDs have been processed and skip the ones you have seen. This is the most general approach and is covered in detail in the next topic.

2. Conditional updates with a version or state guard. Make the write itself refuse to apply twice, by encoding the precondition into the statement.

-- The transition is only valid from one specific prior state, so a
-- redelivery affects zero rows instead of shipping the order twice.
UPDATE orders
   SET status = 'shipped', shipped_at = NOW()
 WHERE id = :order_id
   AND status = 'paid';        -- guard: already-shipped rows do not match

-- Version-based (optimistic concurrency): a stale or repeated event no-ops.
UPDATE inventory
   SET quantity = :qty, version = :version
 WHERE sku = :sku
   AND version < :version;     -- older or equal versions are ignored

3. Natural keys and upserts. Give the derived record a deterministic identity computed from the event, so a duplicate collides with itself.

# The row's primary key is derived from the event, not generated per insert.
payment_id = f"{order_id}:{attempt_no}"      # deterministic, not a random UUID
db.execute("""
    INSERT INTO payments (id, order_id, amount_cents, status)
    VALUES (?, ?, ?, 'captured')
    ON CONFLICT (id) DO NOTHING              -- the duplicate simply vanishes
""", payment_id, order_id, amount_cents)

For side effects you do not own — a payment gateway, an email provider — the mechanism is an idempotency key passed to the external service, which then deduplicates on its side and returns the original result.

# The external call is made safe by the key, not by our retry logic.
stripe.PaymentIntent.create(
    amount=amount_cents, currency="usd", customer=customer_id,
    idempotency_key=f"order-{order_id}-capture",   # deterministic per intent:
)                                                  # NOT a fresh uuid4() per
# Retrying with the same key returns the ORIGINAL result rather than charging
# again. Generating a new key per attempt would defeat the entire mechanism.

Two subtleties are worth stating explicitly. First, idempotency has a time scope: most external services honour idempotency keys for a limited window (commonly 24 hours), so a redrive from a DLQ two days later may not be deduplicated. Second, a handler with several steps is only idempotent if every step is — one non-idempotent write among five idempotent ones makes the whole handler unsafe, which is why the correct design usually wraps the marker and the effects in a single transaction.

Key Takeaways

• Idempotency is a property of the operation; absolute assignment is idempotent, relative modification is not.
• Prefer natural techniques — upserts on deterministic keys, state-guarded updates, version comparisons — over external dedup infrastructure.
• Use idempotency keys for third-party side effects, and derive the key from the event so retries reuse it.
• A handler is idempotent only if every step is; check the whole path, and note that external dedup windows expire.

🧪 Practice

1. Rewrite these to be idempotent: balance = balance - 50, INSERT INTO audit_log, send_sms(user), counter.increment().
2. Write the state-guarded SQL for an order lifecycle (created -> paid -> shipped -> delivered) where each transition is safe to replay.
3. Interview: A handler updates a row, publishes an event, and sends an email. How do you make it idempotent? (Hint: the three steps have three different answers, and one of them may need to move out of the handler.)

Deduplication Strategies

When an operation cannot be made naturally idempotent, you deduplicate explicitly: remember which messages have been processed and skip repeats. The engineering problem is that "remember everything forever" is not viable at volume, so every strategy is a trade between accuracy, memory, and the window over which duplicates can be caught.

The first decision is what identity to deduplicate on. A producer-assigned message ID is the usual choice, but it only catches redelivery of the same message — not the case where a producer's own retry created two records with two IDs describing the same event. For that you need a content-derived key: a deterministic hash of the business fields that define the event's identity.

import hashlib

# Producer-assigned: catches broker redelivery.
message_id = str(uuid.uuid4())

# Content-derived: also catches a producer that sent the same event twice.
def business_key(event):
    material = f"{event['order_id']}:{event['type']}:{event['sequence']}"
    return hashlib.sha256(material.encode()).hexdigest()
# Note what is EXCLUDED: timestamps, hostnames, retry counters. Anything that
# varies between two sends of the "same" event must not enter the key.

The four practical strategies:

Strategy	Memory	Accuracy	Window
Database unique key	Grows with messages	Exact	Unbounded
Redis SET with TTL	Bounded by TTL	Exact within the TTL	Minutes-hours
Bloom filter	Tiny and fixed	No false negatives, some false positives	Configurable
Broker-native dedup	Managed by the broker	Exact within a window	Fixed (e.g. 5 min)

# Strategy 1: the destination database is the dedup store. Strongest option,
# because the marker and the business change commit atomically.
def handle(message):
    with db.transaction():
        try:
            db.execute(
                "INSERT INTO processed (message_id, processed_at) VALUES (?, ?)",
                message.id, now())
        except UniqueViolation:
            return "duplicate"          # atomic: the effect below is skipped
        apply_business_change(message)  # same transaction as the marker

# Bound the table's growth -- otherwise it becomes the largest table you own.
# DELETE FROM processed WHERE processed_at < NOW() - INTERVAL '7 days';

# Strategy 2: Redis SET NX with a TTL. Fast, bounded, but NOT atomic with the
# business change -- there is a window where the marker exists and the work
# did not happen.
def handle(message):
    # NX = only set if absent; EX = expire. One round trip, atomic in Redis.
    if not redis.set(f"dedup:{message.id}", "1", nx=True, ex=86400):
        return "duplicate"
    try:
        apply_business_change(message)
    except Exception:
        redis.delete(f"dedup:{message.id}")   # release the claim so a retry
        raise                                 # can proceed -- otherwise a
                                              # transient failure permanently
                                              # suppresses the message

# Strategy 3: Bloom filter -- constant memory, probabilistic.
# 100M messages at 0.1% false-positive rate needs ~180 MB, versus tens of GB
# for an exact set. The trade: a false positive DISCARDS a real message.
from pybloom_live import ScalableBloomFilter

seen = ScalableBloomFilter(initial_capacity=10_000_000, error_rate=0.001)

def handle(message):
    if message.id in seen:
        return "probably duplicate"    # 0.1% chance this is WRONG and we just
                                       # dropped a legitimate message
    seen.add(message.id)
    apply_business_change(message)
# Acceptable for analytics and metrics. Never for payments: silent data loss
# at a known rate is still silent data loss.

The most important design parameter is the deduplication window — how far back you can detect a repeat. It must be at least as long as the maximum realistic gap between two deliveries of the same message, which is set by your retry policy, your DLQ retention, and your redrive practice.

def dedup_window(max_retries, base_delay_s, dlq_retention_days, redrive_lag_h):
    retry_span = sum(base_delay_s * 2 ** i for i in range(max_retries))
    # The binding constraint is almost always the DLQ: a message redriven
    # after five days must still be recognized as already processed.
    return max(retry_span, dlq_retention_days * 86400 + redrive_lag_h * 3600)

print(dedup_window(5, 30, 14, 4) / 86400)   # 14.17 days
# A 24-hour Redis TTL against a 14-day DLQ is a dedup system that quietly
# fails exactly when it is needed -- during recovery from a real incident.

Key Takeaways

• Deduplicate on a content-derived key when producer retries can create distinct IDs for the same event.
• A unique constraint in the destination database is the strongest strategy because the marker commits atomically with the effect.
• Redis with TTL is fast and bounded but not atomic with the business change — release the claim on failure.
• Bloom filters trade a known rate of silent message loss for constant memory; size the dedup window from DLQ retention, not from the retry policy alone.

🧪 Practice

1. Compute the memory for exact dedup of 500M message IDs over 30 days, and compare it with a Bloom filter at 0.01% false positives.
2. Explain the failure that occurs if the Redis strategy omits the redis.delete on the exception path.
3. Interview: Your dedup window is 24 hours and your DLQ retains 14 days. What goes wrong, and when? (Hint: the failure only appears during a redrive, which is to say during an incident.)

Poison Message Handling

A poison message is one that fails every time it is processed. It is not a transient failure that will succeed on retry — it is structurally unprocessable: malformed JSON, a schema the consumer cannot parse, a reference to a deleted entity, or a payload that triggers a bug.

The danger is disproportionate to the single message. Under at-least-once delivery with retries, a poison message is an infinite loop that consumes capacity forever. In an ordered system it is worse: a poison message at the head of a Kafka partition halts that partition entirely, and every message behind it — possibly millions — waits indefinitely.

 HEAD-OF-LINE BLOCKING IN AN ORDERED PARTITION

  partition 2:  [ BAD ][ m1 ][ m2 ][ m3 ] ... [ m900000 ]
                   ^
                   consumer retries forever; the offset never advances

  Lag on this partition grows without bound while the other partitions are
  perfectly healthy -- which is why per-partition lag, not topic-average lag,
  is the metric that catches this.

The first requirement is to classify the failure correctly, because the right response is opposite for the two categories:

Category	Examples	Correct response
Transient	Timeout, 503, deadlock, connection reset, 429	Retry with backoff
Permanent	Parse error, schema violation, 400, 404, business-rule rejection	Fail immediately to the DLQ

Retrying a permanent failure wastes capacity and delays the inevitable. Dead-lettering a transient failure discards recoverable work during an outage — the worst possible moment, because the DLQ then fills with thousands of messages that would have succeeded five minutes later.

class Permanent(Exception): pass
class Transient(Exception): pass

def classify(exc):
    # Anything the payload itself causes is permanent: a retry re-runs the
    # same bytes through the same code and gets the same result.
    if isinstance(exc, (json.JSONDecodeError, SchemaError, ValidationError)):
        return Permanent(str(exc))
    if isinstance(exc, HTTPError):
        if exc.status in (400, 404, 409, 422):
            return Permanent(f"http {exc.status}")
        if exc.status in (429, 502, 503, 504):
            return Transient(f"http {exc.status}")
    if isinstance(exc, (TimeoutError, ConnectionError, DeadlockDetected)):
        return Transient(str(exc))
    return Transient(str(exc))      # default to transient: a wrong retry is
                                    # recoverable, a wrong discard is not

For ordered streams, where dead-lettering the message is a decision to break ordering, the standard resolution is to skip forward and record the failure, accepting an ordering gap rather than an indefinite stall:

def consume_ordered(record):
    try:
        process(record)
        consumer.commit()
    except Exception as exc:
        err = classify(exc)
        if isinstance(err, Permanent) or attempts(record) >= MAX_ATTEMPTS:
            # Explicit choice: preserve LIVENESS over ORDERING. The gap is
            # recorded so it can be repaired later from the DLQ.
            dlq.send(record, reason=str(err), partition=record.partition,
                     offset=record.offset)
            consumer.commit()          # advance past the poison message
            alert("poison_message", partition=record.partition,
                  offset=record.offset, key=record.key)
        else:
            raise                      # transient: do not advance; retry

Three practices keep poison handling from becoming a silent data-loss channel.

Fail fast on parse errors. Validate the schema before any expensive work, so an unprocessable message costs microseconds rather than a full timeout cycle.

Cap retries per message, not just globally. A message that has failed five times with a transient error is behaving like a permanent failure regardless of what the exception class says.

Alert with the address, not the count. A DLQ alert that says "47 messages" is useless at three in the morning. Include the partition, offset, key, error, and consumer version so the on-call engineer can reproduce it immediately.

The deeper point is that poison messages are usually a schema or contract problem presenting as an operational one. A recurring class of poison message means a producer is emitting something the consumer never agreed to accept, and the durable fix is schema validation at the producer with a registry enforcing compatibility — not a larger DLQ.

Key Takeaways

• A poison message fails deterministically and, without bounded retries, loops forever or blocks an ordered partition entirely.
• Classify failures: retry transient ones with backoff, dead-letter permanent ones immediately, and default to transient when unsure.
• In ordered streams, skipping and recording the gap preserves liveness at the cost of ordering — make that trade explicitly.
• Alert with partition, offset, key, and error; recurring poison messages indicate a producer-side schema problem, not a DLQ-sizing problem.

🧪 Practice

1. Classify as transient or permanent: connection reset, 401, 429, foreign-key violation, out-of-memory, NoSuchBucket.
2. Write the consumer logic that stalls a partition for transient failures but skips past permanent ones, and explain why the two paths differ.
3. Interview: A poison message blocks a Kafka partition with 2 million messages behind it. Walk through your response. (Hint: stop the bleeding first, then diagnose — and decide what you are willing to give up to advance the offset.)

<a id="74-background-work-and-scheduling"></a>

7.4 Background Work and Scheduling

Messaging moves work off the request path; this subchapter is about executing that work reliably — how workers consume it, how failures are retried, and how work that must happen at a particular time gets scheduled across a fleet where no single machine is trustworthy.

Worker Pools and Task Queues

A task queue is a message queue whose payloads describe jobs to execute, and a worker pool is the set of processes that execute them. The distinction from generic messaging is one of intent, and it shows up in the surrounding machinery: task queues care about retries, timeouts, results, progress, and scheduling in a way that raw messaging does not.

The reason to separate workers from the web tier at all is that the two have opposite resource profiles and opposite scaling signals. A web server is latency-sensitive, short-lived per request, and scales on request rate. A worker is throughput-sensitive, potentially long-running, and scales on queue depth. Running them in one process means a burst of heavy jobs starves the request path of exactly the threads it needs to stay responsive.

 TASK QUEUE ARCHITECTURE

  web tier                 broker                worker pool
  +--------+           +-----------+          +-----------+
  | API    |--enqueue->| default   |--------->| worker 1  | generic jobs
  +--------+           +-----------+          +-----------+
                       | emails    |--------->| worker 2  |
                       +-----------+          +-----------+
                       | video     |--------->| worker 3  | CPU-heavy,
                       +-----------+          +-----------+ few, big boxes
                            |
                       +-----------+
                       | result    |<-- workers write status/results here
                       +-----------+     so the API can answer "is it done?"

  Separate queues per workload class -- NOT one queue for everything. A
  40-minute video job and a 200 ms email must not compete for the same
  workers, or every email waits behind a video.

That last point is the most commonly violated rule in task-queue design. Mixing job durations in one queue means short jobs inherit the latency of long ones — head-of-line blocking at the workload level. Separate queues (or separate worker pools bound to specific queues) give each class its own capacity and its own scaling behaviour.

# Celery: workload isolation expressed as routing.
app.conf.task_routes = {
    "tasks.send_email":     {"queue": "fast"},     # ms, high volume
    "tasks.generate_report":{"queue": "slow"},     # minutes, low volume
    "tasks.transcode":      {"queue": "cpu"},      # long, resource-hungry
}

@app.task(
    bind=True,
    acks_late=True,          # ack AFTER execution: a worker crash redelivers
                             # (at-least-once -- the task must be idempotent)
    reject_on_worker_lost=True,
    time_limit=600,          # hard kill: prevents a hung job holding a slot
    soft_time_limit=540,     # raises an exception first, so the job can clean up
    max_retries=5,
)
def generate_report(self, report_id):
    with progress(self, report_id):
        return build_report(report_id)

# Each pool is sized and shaped for its workload.
celery -A app worker -Q fast --concurrency=50 --pool=gevent    # I/O-bound
celery -A app worker -Q cpu  --concurrency=4  --pool=prefork   # CPU-bound
#                                        ^
# Concurrency for CPU work should track core count. Setting it to 50 on a
# 4-core box does not do more work -- it thrashes the scheduler and inflates
# every job's wall-clock time.

Two design questions come up in every task-queue system.

Where do results go? Fire-and-forget jobs need nothing. Jobs whose outcome the user waits for need a result backend — a row in a database or a key in Redis — that the API polls or pushes from. Storing large results in the broker itself is a common mistake: brokers are optimized for small messages in transit, not for holding megabyte payloads.

How big is a job? Prefer many small jobs over one large one. A job that processes 100,000 records in a single execution restarts from zero on any failure, cannot be parallelized, and will eventually exceed any timeout you choose. Splitting it into batches makes progress durable and parallelism possible.

# Anti-pattern: one enormous job. One failure at record 97,000 loses everything.
@app.task
def process_all_users():
    for user in User.objects.all():        # 2 million rows, hours of work
        process(user)

# Better: a coordinator fans out bounded chunks that fail independently.
@app.task
def process_all_users():
    for chunk in chunked(User.objects.values_list("id", flat=True), 500):
        process_user_chunk.delay(list(chunk))   # each chunk retries alone

@app.task(max_retries=3)
def process_user_chunk(user_ids):
    for uid in user_ids:
        process(uid)                        # ~500 units of lost work worst case

Key Takeaways

• Separate workers from the web tier: opposite resource profiles, opposite scaling signals.
• Give each workload class its own queue and pool, or short jobs queue behind long ones.
• Use late acknowledgment plus hard and soft time limits, and match concurrency to the workload type (threads for I/O, cores for CPU).
• Prefer many small jobs to one large one so failure loses a bounded amount of work and parallelism is possible.

🧪 Practice

1. Design the queue topology for a system with password-reset emails, nightly invoicing, image thumbnails, and ML inference. State each queue's concurrency model.
2. Explain what happens when a 4-core box runs --concurrency=64 for CPU-bound jobs.
3. Interview: A job takes 6 hours and fails at hour 5 for the third day running. How do you restructure it? (Hint: what would let day four start from hour 5 instead of hour 0?)

Job Retry and Exponential Backoff

Most job failures are transient — a timeout, a brief deadlock, a dependency restarting — so retrying is usually correct. The question is how, because a naive retry loop is one of the most reliable ways to turn a small incident into a large one.

The failure mode is the retry storm. A dependency slows down, every in-flight job fails, and every worker immediately retries. The dependency now receives its normal load plus all the retries, so it slows further, so more jobs fail, so more retries arrive. The system has built a positive feedback loop around its own failure, and it will not recover on its own even after the original trigger passes.

Exponential backoff breaks the loop by making each successive retry wait longer, giving the dependency room to recover:

 IMMEDIATE RETRY                    EXPONENTIAL BACKOFF

 t=0.0  1000 requests fail          t=0    1000 fail
 t=0.0  1000 retries                t=1s   1000 retries
 t=0.0  1000 retries                t=2s   1000 retries
 ...    load is UNBOUNDED           t=4s   1000 retries
                                    t=8s   1000 retries
 The dependency never gets a        t=16s  1000 retries
 quiet moment to recover.
                                    Load decays geometrically; the dependency
                                    gets progressively longer quiet windows.

Backoff alone is not enough, because all the failed jobs retry at the same time — the retries are synchronized by the shared failure. Every retry wave arrives as a spike. Jitter — randomizing each delay — spreads them out, and it matters as much as the backoff itself.

import random

def backoff_delay(attempt, base=1.0, cap=300.0, strategy="full_jitter"):
    """Delay before retry number `attempt` (0-indexed)."""
    exponential = min(cap, base * (2 ** attempt))     # 1, 2, 4, 8, ... capped
    if strategy == "none":
        return exponential                 # synchronized waves: avoid
    if strategy == "full_jitter":
        return random.uniform(0, exponential)         # best general choice:
                                                      # maximum spread
    if strategy == "equal_jitter":
        half = exponential / 2
        return half + random.uniform(0, half)         # some spread, less
                                                      # variance in worst case
    if strategy == "decorrelated":
        prev = backoff_delay.last if hasattr(backoff_delay, "last") else base
        d = min(cap, random.uniform(base, prev * 3))
        backoff_delay.last = d
        return d

# Retry schedule with full jitter, base 1s, cap 300s:
#   attempt 0 -> 0.0-1s     attempt 3 -> 0-8s
#   attempt 1 -> 0.0-2s     attempt 4 -> 0-16s
#   attempt 2 -> 0.0-4s     attempt 8 -> 0-256s
# Total worst-case span for 8 attempts: ~8.5 minutes.

# Celery: retry with backoff and jitter declared, not hand-rolled.
@app.task(
    bind=True,
    autoretry_for=(TransientError, requests.Timeout),
    retry_backoff=2,          # exponential base
    retry_backoff_max=600,    # cap the delay at 10 minutes
    retry_jitter=True,        # randomize -- do not skip this
    max_retries=6,
)
def call_partner_api(self, payload):
    resp = requests.post(PARTNER_URL, json=payload, timeout=10)
    if resp.status_code in (429, 502, 503, 504):
        # Honour the server's own guidance when it gives any.
        raise self.retry(countdown=int(resp.headers.get("Retry-After", 30)))
    if 400 <= resp.status_code < 500:
        raise PermanentError(resp.text)     # do NOT retry a client error
    resp.raise_for_status()

Three constraints bound any retry policy.

A retry budget. Even with backoff, unlimited retries across many jobs can keep a dependency saturated. A budget — "retries may not exceed 10% of total requests" — is a global cap that backoff alone does not provide.

A deadline, not just an attempt count. If a job's result is worthless after five minutes, retrying it at minute forty is pure waste. Carry a deadline in the job and abandon work that can no longer matter.

def should_retry(job, attempt, max_attempts=6):
    if time.time() > job["deadline"]:
        return False, "deadline_exceeded"     # the result is now worthless
    if attempt >= max_attempts:
        return False, "attempts_exhausted"
    if retry_budget.exhausted():              # global protection for the
        return False, "budget_exhausted"      # dependency, across all jobs
    return True, None

A circuit breaker for the hard-down case. When a dependency is fully down, even backed-off retries are wasted work and add load to something that cannot serve it. A breaker that trips after a threshold of consecutive failures fails subsequent jobs instantly, then probes periodically to detect recovery.

Key Takeaways

• Immediate retries create feedback loops that keep a struggling dependency down; exponential backoff decays the load geometrically.
• Jitter is not optional — without it, all retries stay synchronized and arrive as waves.
• Retry only transient failures; a 4xx repeated six times is still a 4xx.
• Bound retries by attempts, a deadline, a global retry budget, and a circuit breaker for the hard-down case.

🧪 Practice

1. Compute the total elapsed time for 8 attempts with base 2s and a 120s cap, with and without full jitter (give the range).
2. Explain concretely why 10,000 jobs retrying with identical backoff still overload a dependency.
3. Interview: A partner API is down for two hours. Your retry policy is 10 attempts with exponential backoff. What happens to the queue, and what should happen instead? (Hint: how much work accumulates, and what does it all do the moment the partner recovers?)

Delayed and Scheduled Jobs

Not all work should run immediately. A trial ending in 14 days, a reminder in an hour, a retry in 30 seconds, an abandoned-cart email 3 hours after the last activity — all require executing a job at a future time, which is a different problem from executing it as soon as capacity allows.

The naive approach — a worker that polls a table every second for due jobs — works and is often good enough, but it has three weaknesses that matter at scale: polling load proportional to poll frequency rather than to work, timing precision bounded by the poll interval, and a full table scan or index sweep on every tick.

There are four mechanisms, and the right one depends on delay length and volume.

Mechanism	Precision	Max delay	Scale
Broker-native delay (SQS)	Seconds	15 minutes	Excellent, fully managed
Redis sorted set	Sub-second	Unbounded	Very good (millions)
Database polling table	Poll interval	Unbounded	Moderate; indexes matter
Time-bucketed queues	Bucket width	Unbounded	Excellent

# 1. Broker-native: simplest, but note the 15-minute ceiling.
sqs.send_message(QueueUrl=Q, MessageBody=body, DelaySeconds=900)   # max 900

# 2. Redis sorted set: the general-purpose delayed-job structure.
#    Score = execution timestamp. Due jobs are always a range query at the
#    front of the set, which is O(log n) regardless of how many are scheduled.
def schedule(job, run_at_epoch):
    redis.zadd("delayed_jobs", {json.dumps(job): run_at_epoch})

def poll_due(now=None, batch=100):
    now = now or time.time()
    # Atomically claim due jobs so multiple pollers cannot take the same one.
    lua = """
    local due = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1],
                           'LIMIT', 0, tonumber(ARGV[2]))
    if #due > 0 then redis.call('ZREM', KEYS[1], unpack(due)) end
    return due
    """
    for raw in redis.eval(lua, 1, "delayed_jobs", now, batch):
        work_queue.enqueue(json.loads(raw))   # move it to the ready queue

-- 3. Database table: durable and queryable, but the index is what makes it
--    viable. Without it, every poll scans the whole table.
CREATE TABLE scheduled_jobs (
    id          BIGSERIAL PRIMARY KEY,
    run_at      TIMESTAMPTZ NOT NULL,
    payload     JSONB       NOT NULL,
    status      TEXT        NOT NULL DEFAULT 'pending',
    attempts    INT         NOT NULL DEFAULT 0,
    locked_by   TEXT,
    locked_at   TIMESTAMPTZ
);
-- Partial index: only pending rows are ever queried, so the index stays small
-- even when the table holds years of completed jobs.
CREATE INDEX idx_due ON scheduled_jobs (run_at) WHERE status = 'pending';

-- Claiming rows safely across many pollers. SKIP LOCKED is the key: each
-- poller takes different rows instead of blocking on the same ones.
UPDATE scheduled_jobs SET status = 'running', locked_by = :worker,
                          locked_at = NOW()
 WHERE id IN (
     SELECT id FROM scheduled_jobs
      WHERE status = 'pending' AND run_at <= NOW()
      ORDER BY run_at
      LIMIT 100
      FOR UPDATE SKIP LOCKED
 )
RETURNING id, payload;

 4. TIME-BUCKETED QUEUES (the hierarchical timing wheel, simplified)

  Bucket by minute; one queue per bucket. A scheduler moves whole buckets
  into the ready queue as their time arrives.

   now=10:00      10:01        10:02        10:03
  [ ready ]  <-- [ 47 jobs ] [ 3 jobs ]  [ 1200 jobs ]

  Cost per tick is O(1) in the number of SCHEDULED jobs -- you touch one
  bucket, never the whole set. This is how schedulers handle tens of
  millions of pending timers.

Three details separate a working scheduler from a broken one.

Long delays must survive restarts. An in-process setTimeout for 14 days is lost on the next deploy. Anything beyond seconds belongs in durable storage.

"Due" jobs arrive in bursts. Ten thousand trials expiring at midnight all become due in the same second. Either spread the scheduled times deliberately (add jitter when scheduling) or rate-limit the dispatcher, or the scheduler becomes a load generator.

Cancellation must be possible. A reminder scheduled for a task that gets completed early must not fire. Store a job ID that can be cancelled, and — since cancellation races with dispatch — re-verify the precondition at execution time rather than trusting the schedule.

def send_trial_reminder(user_id, scheduled_for):
    user = db.get_user(user_id)
    # The world may have changed in the 14 days since scheduling. Always
    # re-check preconditions at execution time; cancellation is best-effort.
    if user.subscription_status != "trial" or user.trial_ends_at != scheduled_for:
        return "stale, skipped"
    send_email(user, "trial_ending")

Key Takeaways

• Delayed jobs need durable storage; in-process timers do not survive deploys.
• Redis sorted sets handle general delayed execution efficiently; time-bucketed queues scale to tens of millions of pending timers.
• For database-backed schedulers, use a partial index on pending rows and FOR UPDATE SKIP LOCKED so pollers do not contend.
• Scheduled times cluster — jitter them — and always re-verify preconditions at execution time, because cancellation races with dispatch.

🧪 Practice

1. Design storage for 50 million pending reminders spread over two years, with one-minute precision. Estimate the memory or storage required.
2. Explain why SKIP LOCKED is essential when several pollers share one table, and what happens without it.
3. Interview: 100,000 subscription renewals all fall due at midnight UTC. What breaks and how do you fix it? (Hint: the fix can be applied when the jobs are scheduled, not only when they are run.)

Distributed Cron

Cron on a single machine is simple and dependable. Across a fleet it becomes a genuinely hard problem, because the property you need — "this runs exactly once at this time" — collides with the property you built the fleet for — "there are many identical machines".

Put a crontab on ten application servers and the nightly billing job runs ten times. Put it on one designated machine and you have reintroduced a single point of failure: when that machine is down at 02:00, billing silently does not happen, and nobody finds out until a customer asks.

 THE DISTRIBUTED CRON PROBLEM

  cron on every node          cron on ONE node
  +------+ +------+ +------+  +------+ +------+ +------+
  | n1 * | | n2 * | | n3 * |  | n1 * | | n2   | | n3   |
  +------+ +------+ +------+  +------+ +------+ +------+
   runs     runs     runs      runs     idle     idle
      \       |       /
       3 EXECUTIONS             1 execution -- unless n1 is down at 02:00,
       (triple billing)         in which case: 0 executions, silently

Three solutions exist, in increasing order of robustness.

1. A distributed lock. Every node tries to acquire a lock keyed by job and scheduled time; the winner runs. Simple and effective, with one important caveat: a lock with a TTL can expire while the holder is still working, letting a second node start. For jobs where a double execution is unacceptable, the lock must be paired with an idempotent job or a fencing token.

def run_with_lock(job_name, scheduled_slot, ttl_seconds, fn):
    # Keying the lock by the SLOT, not just the job name, means a job that
    # runs long cannot be started twice for the same scheduled time even
    # after the lock expires.
    key = f"cron:{job_name}:{scheduled_slot}"
    token = str(uuid.uuid4())
    if not redis.set(key, token, nx=True, ex=ttl_seconds):
        return "another node holds this slot"
    try:
        # Extend the lease periodically for long jobs rather than setting a
        # huge TTL, which would block the next slot after a crash.
        with lease_renewal(key, token, ttl_seconds):
            return fn()
    finally:
        # Release only if we still hold it -- never delete another node's lock.
        redis.eval("if redis.call('GET', KEYS[1]) == ARGV[1] then "
                   "return redis.call('DEL', KEYS[1]) else return 0 end",
                   1, key, token)

2. Leader election. One node is elected leader through a consensus system (etcd, ZooKeeper, Consul) and runs all scheduled jobs; if it fails, a new leader is elected within seconds. This is more robust than ad-hoc locks and gives a single coherent place to reason about scheduling.

3. A dedicated scheduler service. A managed or purpose-built scheduler (Kubernetes CronJob, cloud schedulers, Temporal, Quartz clustered mode) owns the schedule and enqueues a message; ordinary workers execute it. This is the strongest pattern because it separates deciding to run from running, and the execution then inherits all the reliability of the task queue — retries, DLQ, observability, backpressure.

# Kubernetes CronJob: scheduling is infrastructure, not application code.
apiVersion: batch/v1
kind: CronJob
metadata: { name: nightly-billing }
spec:
  schedule: "0 2 * * *"
  timeZone: "UTC"                    # ALWAYS pin this: DST makes local time
                                     # schedules run twice or zero times a year
  concurrencyPolicy: Forbid          # never start if the previous run is going
  startingDeadlineSeconds: 600       # if the controller was down, still start
                                     # within 10 minutes -- otherwise SKIP the
                                     # slot rather than run it hours late
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 5
  jobTemplate:
    spec:
      backoffLimit: 2
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: billing
              image: billing:v4.2
              args: ["--run-date=$(RUN_DATE)"]   # pass the logical date in, so
                                                 # a re-run reproduces exactly

Beyond the mechanism, four properties distinguish scheduled jobs that survive production:

Idempotency by scheduled slot. Whatever the locking, assume a job may run twice. Key its effects on the slot — billing_run_2026_08_25 — so a second execution is a no-op.

Missed-run policy. If the scheduler was down at 02:00, should the 02:00 run happen at 04:00? For billing, probably yes. For "send the daily digest", probably no. Decide explicitly; startingDeadlineSeconds is that decision.

Overrun policy. If the hourly job takes 90 minutes, do you skip, queue, or overlap? Forbid prevents pile-ups, which is the sane default.

Alerting on absence. The characteristic failure of scheduled work is that it does not run, which produces no errors and no logs. A dead-man's-switch — alert if job X has not reported success in the last 26 hours — is the only monitor that catches it.

def run_scheduled(job_name, slot):
    # Idempotency keyed on the logical slot, enforced by the database.
    try:
        db.execute("INSERT INTO cron_runs (job, slot, started_at) "
                   "VALUES (?, ?, NOW())", job_name, slot)
    except UniqueViolation:
        return "already ran for this slot"
    result = execute(job_name, slot)
    db.execute("UPDATE cron_runs SET finished_at = NOW(), status = ? "
               "WHERE job = ? AND slot = ?", result.status, job_name, slot)
    heartbeat.report(job_name)      # feeds the dead-man's-switch alert
    return result

Key Takeaways

• Cron on every node multiplies executions; cron on one node reintroduces a single point of failure.
• Distributed locks keyed by scheduled slot are the simple answer; leader election is more robust; a scheduler service that enqueues work is strongest.
• Make jobs idempotent on the scheduled slot regardless of locking, since a lock TTL can expire mid-run.
• Decide missed-run and overrun policies explicitly, pin the timezone, and alert on absence — a job that never runs produces no errors.

🧪 Practice

1. A lock has a 5-minute TTL and the job occasionally takes 8 minutes. Describe the failure and give two fixes.
2. Design the missed-run policy for: nightly billing, an hourly cache refresh, a daily digest email, a weekly database vacuum.
3. Interview: How do you find out that a nightly job silently stopped running three weeks ago? (Hint: you cannot alert on an error that was never raised — what can you alert on instead?)

Batch Processing Windows

Batch processing groups many items into one execution. It exists because per-item overhead — a network round trip, a transaction, an API handshake, a file open — is often far larger than the per-item work, so processing a thousand items together can cost barely more than processing one.

The gain is easy to quantify and usually surprising:

# Per-item cost dominated by round trips.
ITEMS, RTT_MS, PER_ITEM_MS = 100_000, 2.0, 0.05

one_at_a_time = ITEMS * (RTT_MS + PER_ITEM_MS)        # 205,000 ms = 3.4 min
batched_1000  = (ITEMS / 1000) * (RTT_MS + 1000 * PER_ITEM_MS)  # 5,200 ms

print(f"individual: {one_at_a_time/1000:.0f}s   batched: {batched_1000/1000:.0f}s"
      f"   speedup: {one_at_a_time/batched_1000:.0f}x")
# individual: 205s   batched: 5s   speedup: 39x
#
# Nothing about the work got faster. The overhead was amortized, and that is
# where nearly all batching wins come from.

The cost is latency: an item that arrives at the start of a batch window waits for the window to close. Every batching decision is therefore a throughput versus freshness trade, and the right window is set by how stale the result is allowed to be, not by what is technically possible.

A batch should close on whichever of three conditions comes first:

class BatchAccumulator:
    """Flush on size, age, or byte limit -- whichever triggers first."""
    def __init__(self, max_items=1000, max_age_s=5.0, max_bytes=1_000_000):
        self.max_items, self.max_age_s, self.max_bytes = max_items, max_age_s, max_bytes
        self.items, self.bytes, self.opened_at = [], 0, None

    def add(self, item, size):
        if not self.items:
            self.opened_at = time.monotonic()   # the clock starts on the FIRST
        self.items.append(item)                 # item, not on the last one
        self.bytes += size
        return self.should_flush()

    def should_flush(self):
        if len(self.items) >= self.max_items:
            return "size"
        if self.bytes >= self.max_bytes:
            return "bytes"      # protects against a few huge items
        if self.opened_at and time.monotonic() - self.opened_at >= self.max_age_s:
            return "age"        # ESSENTIAL: without it, a slow trickle of
                                # items is never flushed at all
        return None

 BATCH WINDOW TRADE-OFF

  window   throughput        p99 latency      good for
  --------------------------------------------------------------
  10 ms    moderate          ~15 ms           interactive writes
  1 s      high              ~1.1 s           near-real-time aggregation
  1 min    very high         ~65 s            dashboards, metrics rollups
  1 hour   maximum           ~65 min          reporting, ETL
  1 day    maximum           ~24 h            billing, reconciliation

  Pick the window from the freshness requirement, then verify the throughput
  it yields is sufficient -- not the other way around.

Three problems recur in batch systems.

Partial failure. If item 700 of 1000 fails, what happens to the other 999? Three policies exist and each is right somewhere: all-or-nothing (transactional, simplest to reason about), best-effort (apply what works, collect failures), and split-and-retry (bisect the batch to isolate the bad item).

def process_batch(items):
    try:
        with db.transaction():
            bulk_insert(items)                # all-or-nothing
    except BulkError:
        if len(items) == 1:
            dead_letter(items[0])             # isolated the culprit
            return
        # Bisect: two more attempts localize the failure logarithmically
        # instead of retrying 1,000 items one at a time.
        mid = len(items) // 2
        process_batch(items[:mid])
        process_batch(items[mid:])

Boundary correctness for time windows. A batch covering "yesterday" must define what happens to an event that occurred at 23:59:58 and arrived at 00:00:03. Use event time rather than processing time, allow a grace period for late arrivals, and make the job re-runnable so a corrected result can replace an earlier one.

-- Windowing on event time with a watermark for late data.
SELECT date_trunc('hour', event_time) AS window_start,
       COUNT(*), SUM(amount_cents)
  FROM events
 WHERE event_time  >= :window_start          -- the logical window
   AND event_time  <  :window_end
   AND ingested_at <  :window_end + INTERVAL '15 minutes'  -- grace period
 GROUP BY 1;
-- The job is keyed by window_start and is idempotent: re-running it after
-- the grace period replaces the earlier result rather than adding to it.

Batch size versus resource limits. Larger batches are not monotonically better. Memory, transaction log size, lock duration, and request payload limits all impose ceilings, and a batch that holds a database transaction open for minutes blocks other writers and inflates replication lag. The right size is usually found empirically and is smaller than the theoretical optimum.

Key Takeaways

• Batching amortizes fixed per-item overhead and commonly yields order-of- magnitude gains without making any individual operation faster.
• Its cost is latency; size the window from the freshness requirement.
• Flush on size, age, or bytes — whichever comes first — because the age trigger is what saves a slow trickle from never flushing.
• Decide a partial-failure policy explicitly, window on event time with a grace period, and cap batch size by transaction and memory limits, not theory.

🧪 Practice

1. Compute the speedup for 1M items with a 5 ms round trip and 0.02 ms per item at batch sizes 1, 100, and 10,000. Where do the returns flatten?
2. Implement bisecting retry and compute the number of attempts needed to isolate one bad item in a 1,024-item batch.
3. Interview: A daily aggregation job produces different numbers each time it is re-run. What is wrong? (Hint: consider which clock decides whether an event belongs to yesterday.)

Long-Running Workflow Orchestration

Some processes span minutes, days, or months and involve many steps across many services: onboarding a customer, fulfilling an order, running a loan application. These cannot be a single job, because no process should hold state in memory for three days, and any failure at step seven must not restart from step one.

The naive implementation is a chain of queue messages: step one enqueues step two, which enqueues step three. It works until you need to answer ordinary questions — where is order 991 right now, why did it stop, what happens if step four fails after step three moved money — and discover that the workflow's state exists only as scattered messages with no coherent record anywhere.

Workflow orchestration makes the process itself a durable, inspectable object. Two architectural styles exist:

 ORCHESTRATION (central brain)        CHOREOGRAPHY (event chain)

     +--------------+                  [ OrderPlaced ]
     | orchestrator |                        |
     +--------------+                        v
      /     |      \                   payment service
     v      v       v                        |
  payment inventory shipping                 v
                                       [ PaymentCaptured ]
  + State is in ONE place                    |
  + Easy to see and to change                v
  - The orchestrator is a               inventory service
    component to run                         |
                                             v
                                       [ StockReserved ] -> shipping ...

                                       + No central component
                                       - The process exists nowhere; you
                                         reconstruct it from logs
                                       - Changing the order means changing
                                         several services

Neither is universally right: choreography suits loosely coupled reactions, orchestration suits processes with a defined outcome that someone must be able to inspect and debug. Business processes with money in them are almost always better orchestrated.

The critical requirement for long-running workflows is durable execution: the workflow's progress is persisted after every step, so a crash resumes from the last completed step rather than the beginning. Modern engines (Temporal, Cadence, AWS Step Functions, Azure Durable Functions) provide this by recording an event history and replaying it to rebuild in-memory state after a restart.

# Temporal: ordinary-looking code that survives process death, deploys, and
# multi-day sleeps. The engine persists each step's result.
@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order: Order) -> str:
        # Every activity result is durably recorded. A crash after this line
        # resumes AFTER it -- the payment is never captured twice.
        payment = await workflow.execute_activity(
            capture_payment, order,
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=5,
                                     backoff_coefficient=2.0),
        )
        try:
            await workflow.execute_activity(
                reserve_inventory, order,
                start_to_close_timeout=timedelta(seconds=30))
        except ActivityError:
            # COMPENSATION: no distributed transaction exists across these
            # services, so the workflow undoes the earlier step explicitly.
            await workflow.execute_activity(refund_payment, payment)
            return "failed: out of stock"

        await workflow.execute_activity(ship_order, order,
            start_to_close_timeout=timedelta(minutes=5))

        # A three-day durable sleep. No thread, no memory, no timer process --
        # the engine simply schedules the continuation.
        await asyncio.sleep(timedelta(days=3).total_seconds())
        await workflow.execute_activity(request_review, order)
        return "completed"

Because there is no transaction spanning payment, inventory, and shipping, a workflow that fails partway must compensate — apply a semantic undo for each completed step, in reverse order. This is the saga pattern, and its defining constraint is that compensation is not rollback: a refund is a new transaction that leaves both the charge and the refund visible, and some steps (an email already sent) cannot be undone at all.

# The saga structure, independent of any engine.
async def run_saga(steps, context):
    completed = []
    try:
        for step in steps:
            await step.execute(context)
            completed.append(step)         # remember what needs undoing
    except Exception as exc:
        for step in reversed(completed):   # compensate in REVERSE order
            try:
                await step.compensate(context)
            except Exception:
                # A failed compensation is an operational emergency: the
                # system is now in a state no code path intended. Alert; do
                # not swallow.
                alert("compensation_failed", step=step.name, context=context)
        raise

Three design rules follow.

Separate deterministic orchestration from side-effecting work. Engines rebuild workflow state by replaying its history, which only works if the workflow code produces the same decisions every time. Anything non-deterministic — now(), random(), a network call — must live in an activity, not in the workflow body.

Design compensations up front. Ask for every step: if a later step fails, what undoes this? Steps with no possible compensation should be ordered last, so they only occur once everything reversible has succeeded.

Make the workflow observable. The value of an orchestrator is partly that "where is order 991?" has an answer. Ensure state, current step, attempt counts, and failure reasons are queryable by support staff, not only by engineers with log access.

Key Takeaways

• Long processes need durable execution: progress persisted per step so a crash resumes rather than restarts.
• Orchestration centralizes process state and makes it inspectable; choreography avoids a central component but leaves the process implicit.
• Without distributed transactions, partial failure is handled by sagas — compensating actions applied in reverse order.
• Keep workflow code deterministic and push all side effects into activities; order irreversible steps last.

🧪 Practice

1. Write the compensation for each step of: reserve inventory, charge card, generate label, send confirmation email. Which cannot be compensated?
2. Explain why calling datetime.now() inside workflow code breaks replay-based durable execution.
3. Interview: A saga's compensation step itself fails. What now? (Hint: no code path can fix this automatically — what must the system guarantee instead?)

---

<a id="8-distributed-systems-theory"></a>