Kafka Fundamentals: The Log Between Producers and Consumers

In plain terms

An order service publishes an event once; a fraud service and a warehouse loader each read it at their own pace. Kafka retains records according to topic policies, independently of these readers, so they can resume or replay the history that remains available. Its central structure is a distributed log: records are appended in order within each partition, with durability determined by replication and acknowledgement settings.

Earlier articles introduced streaming collection and log collectors. Kafka can be the retained handoff between those collectors and downstream processing. Here we follow partition placement, acknowledgements, consumer progress, and replay, then trace selected failure cases in a small Python model.

A log, not a queue

In a conventional work queue, acknowledgement usually removes a message from that queue. Fan-out queues and messaging services with retention can also support multiple readers or replay. Kafka makes the retained log explicit: each partition assigns offsets to appended records, and consumer groups track progress independently. Reading does not delete a record. Replay starts at an available offset, not necessarily offset zero; deletion and compaction policies determine which history remains.

Three properties follow, and they are the reasons a platform puts a log between its sources and its consumers. Decoupling: the checkout service writes once and does not know or care that the fraud model, the warehouse loader, and the analytics job all read it. Buffering: when a consumer is down or slow, records wait in the log, not in the producer’s memory, provided retention and storage capacity cover the outage; a broker outage still requires producer-side retry or buffering. Replay: a new consumer, a fixed consumer, or a backfill reads history from the log through the same code that reads the present, which is what the Kappa idea in the architecture article depended on.

Topics, partitions, offsets, keys

A topic is a named log, such as events.checkout or cdc.orders. A topic is split into partitions, each an independent append-only log with its own offsets, spread across the brokers, the servers that store them. Partitions are the unit of everything: of parallelism, because different consumers read different partitions; of ordering, because records are ordered within a partition and nowhere else; and of placement, because a record goes to one partition and stays there.

With stable key serialization, partitioner, and partition count, key-based placement sends the same key to the same partition. Kafka preserves that partition’s append order, not global event-time order. Concurrent producers can interleave, and retries without suitable ordering settings can reorder sends. Different keys may share a partition; they then share its log order. Null-key placement depends on the client and partitioner and often batches records using sticky placement rather than strict round-robin. Choose a key for the entity whose updates must stay together. The lab uses CRC32, not the Java producer’s default key hash: growing from six to twelve partitions moves 241 of its 500 customer keys. The general risk is splitting one key’s old and new history across partitions.

A record is a key, a value, a timestamp, and optional headers. Identify its log position by topic, partition, and offset; offset 10 in partition 0 is unrelated to offset 10 in partition 1. The broker stores the value as bytes rather than enforcing the application’s business schema; that schema is the producer’s and consumer’s agreement.

Replication and acknowledgement

A partition has a leader replica and follower replicas on other brokers. Writes go to the leader; consumer reads normally do too, although configured follower reads are possible. The in-sync replica set (ISR) includes the leader and followers meeting Kafka’s catch-up criteria; it does not mean every follower has every new record at every instant. On leader failure, a suitable replica can take over. Safe election policy and surviving replicas matter to whether acknowledged data remains available.

acks=0 waits for no broker acknowledgement, so the producer cannot confirm acceptance. acks=1 confirms the leader’s append; failure before replication can lose that record. acks=all waits for the ISR, while min.insync.replicas=2 rejects such writes when fewer than two replicas are in sync. With three replicas, this can tolerate one broker failure if the remaining replicas and controller quorum are available. Leader election and retries may still interrupt requests; it is not a zero-downtime or fixed-latency promise. An acknowledgement confirms the configured replication condition, not an fsync on every disk or completion in an external database. A timeout is ambiguous: the append may have succeeded.

Producers

A producer batches records per partition and can compress batches before sending them. linger.ms allows time to fill batches; throughput also depends on record size, network, broker capacity, and workload. Lost acknowledgements make retries ambiguous. Kafka’s idempotent producer uses producer identity, epochs, and per-partition sequences to recognize retries within its protocol. For the Java 4.1 client, idempotence is enabled by default when settings do not conflict; it requires acks=all, retries, and at most five in-flight requests per connection. Check the actual client and configuration, and handle terminal send failures. This is not deduplication of a business event sent again as a new record or after an application restart.

Consumers, groups, and offsets

A conventional consumer group divides assigned partitions among members: a partition has at most one assigned member at a time, and members may own several partitions. More members than assigned partitions leaves some idle. This assignment does not guarantee one processing attempt: crashes and rebalances can cause redelivery. A separate group has separate progress and can read the same retained records. This article uses conventional consumer groups, not Kafka share groups with per-record acknowledgement.

Distinguish position, the next offset this consumer will fetch, from the committed offset, the restart position stored for the group and partition. Polling advances position; committing saves progress and does not itself seek the reader. After processing offset 49 durably, commit 50: the next offset to read. Reassignment resumes from committed progress. Commit a whole batch first and a crash can skip unfinished work; commit after durable output and a crash before commit can repeat work. Under these assumptions, these are at-most-once and at-least-once processing strategies, respectively. The lab demonstrates 50 skipped versus 50 repeated records. Automatic commit is not inherently commit-before-processing: safety depends on the client’s poll/commit behavior and whether previously fetched work has completed. Asynchronous handoff can commit ahead of durable work. Use explicit commits when coordinating output, and commit only through the contiguous completed prefix of each partition.

Suppose one partition returns offsets 50, 51, and 52. If 50 and 52 finish but 51 is still running, commit 51, not 53: the committed value names the next record to resume. Committing 53 would skip unfinished 51 after a restart. Once 51 finishes durably, the completed prefix extends through 52 and can advance to 53. Keep completion tracking per partition, and stop stale workers from committing or producing uncontrolled effects after reassignment.

Committed-offset lag is the distance from a partition’s end to the group’s committed offset. Inspect it per partition, together with processing latency, oldest unfinished work, and error rates. Offset distance need not equal a count of readable records after compaction or transactions. One hot partition can dominate completion time. Speeding up that partition’s consumer or output can help; adding group members alone cannot divide its assignment. Splitting a hot key trades away the original per-key order unless another mechanism restores it. The lab holds processing capacity fixed to isolate skew.

Retention, compaction, replay

Delete retention removes eligible old log segments according to time and per-partition size limits, independently of consumer progress. Verify topic overrides and broker defaults; deletion is not an exact per-record TTL. If a saved offset is no longer available, auto.offset.reset determines whether to start at the earliest available offset, at the end, or raise an error, depending on the selected policy. It also applies when no committed offset exists; it does not rewind a valid committed offset on every restart. Resetting to the earliest preserves only the remaining history. A gap requires recovery from another source or explicit acceptance of missing data.

Compaction asynchronously removes superseded values for a key. It preserves record offsets and order, leaving gaps; it does not instantly turn a topic into a one-row-per-key table. A null-valued record, a tombstone, represents deletion, and delete markers can later expire. A state-building consumer must account for that retention window. With compaction alone, the latest non-deleted state can be retained; combining compact,delete also permits time/size deletion. Replaying a compacted topic reconstructs retained state, not every historical event. Tiered storage changes where segments live, not whether retention can remove them; replay also needs compatible historical schemas and controlled output effects.

If a consumer retains an old local row but misses the tombstone before it expires, resuming the surviving compacted log may leave that stale row in place. Restoring a current-state view may require rebuilding an empty target from a complete retained state or reconciling against an authoritative snapshot. Replaying a suffix into an old target is not equivalent to rebuilding from scratch. The dictionary helper below starts empty and does not simulate an expired tombstone.

Delivery, end to end

Transactions can atomically publish output records and consumed offsets within Kafka. Downstream consumers need isolation.level=read_committed to exclude aborted transaction records. The transaction boundary is the key: a separate database transaction followed by a Kafka offset commit still leaves a crash window. For an external sink, use a durable event ID with atomic deduplication and the business update, or a supported coordination protocol. Producer idempotence suppresses protocol retries, while consumer retries and business duplicates require their own handling. Eventual delivery also assumes successful recovery before the required data expires.

Kafka, Redpanda, Kinesis, Pub/Sub

These systems all decouple event producers and consumers, but they do not expose identical delivery or replay models. The table is a starting comparison, not a feature-compatibility guarantee. Confirm client support, quotas, retention configuration, and regional availability for a deployment.

Choose by required retention, replay procedure, client compatibility, and operational ownership. A managed service runs the brokers, but the application team still owns schemas, keys, consumers, permissions, and recovery. Pub/Sub acknowledgement and seek cannot be treated as renamed Kafka group offsets.

Operating a cluster

Monitor under-replicated and offline partitions, per-group lag, request errors and latency, and storage headroom. Retained bytes roughly follow ingress rate times retention times replication factor, with compression, indexes, segment behavior, and spare capacity accounted for; consumer fan-out adds read and network load. Kafka 4.x uses a KRaft controller quorum for cluster metadata and leadership coordination, distinct from data-partition replicas. The controller quorum needs an available majority; plan its failures separately from data replication. TLS encrypts traffic and can authenticate clients with certificates; SASL provides configured authentication, and ACLs authorize resource operations. Apply producer/consumer quotas and record topic ownership and configuration.

The subscription company’s topics

Keys follow what must be ordered, partition counts follow expected throughput with room to grow, retention follows who needs to replay and how far back, and the compacted topic is the one that carries state.

Anti-patterns

• Ordering assumed across partitions. A consumer that expects the checkout to arrive after the product view for a different key.
• Growing partitions casually. Doubling the count on a keyed topic and splitting some customers’ histories across partitions.
• acks=1 for events that matter. A leader failover during a busy hour and a few thousand orders that were acknowledged by the leader but lost before replication.
• Committing ahead of durable work. A fetched batch handed to asynchronous workers and committed before their outputs finish.
• More consumers than partitions. Scaling the group to twenty on a six-partition topic and wondering why lag did not move.
• One hot key. The biggest customer, or a null key replaced by a constant, in one partition that decides every job’s finish time.
• Retention shorter than the longest outage. A consumer down for a weekend on a topic retained for two days.
• The queue mindset. Deleting a topic’s records after reading them, and losing the replay that was the point.

Lab

This lab is an in-memory teaching model, not a Kafka broker or client implementation. It uses CRC32 string keys, contiguous list offsets, prefix expiration, a single outstanding record per producer-partition, and round-robin assignment. It omits replication, storage durability, epochs, transactions, networking, and real rebalance fencing. Unlike Kafka’s consumer-wide max.poll.records, its max_records applies per partition. The retention fixture has increasing timestamps; real Kafka deletes segments. Run the setup once before each independent exercise. It generates 3,000 orders over 500 possible customer IDs, with a 30% chance of C1 on each draw.

import random, zlib, copy
from collections import Counter, defaultdict

class OffsetOutOfRange(Exception):
    pass

def nonnegative_int(value, name):
    if type(value) is not int or value < 0:
        raise ValueError(f"{name} must be a nonnegative integer")

class Topic:
    """In-memory partition logs; no replication, transactions, or durable storage."""

    def __init__(self, partitions):
        nonnegative_int(partitions, "partitions")
        if partitions == 0:
            raise ValueError("partitions must be positive")
        self.logs = [[] for _ in range(partitions)]
        self.earliest = [0] * partitions
        self.last_sequence = {}

    def partition_for(self, key):
        if not isinstance(key, str):
            raise ValueError("this model requires a string key")
        return zlib.crc32(key.encode()) % len(self.logs)

    def check_partition(self, p):
        nonnegative_int(p, "partition")
        if p >= len(self.logs):
            raise ValueError("unknown partition")

    def append(self, key, value, ts, producer_id=None, sequence=None):
        p = self.partition_for(key)
        nonnegative_int(ts, "timestamp")
        record = {"key": key, "value": copy.deepcopy(value), "ts": ts}
        if (producer_id is None) != (sequence is None):
            raise ValueError("producer_id and sequence must be supplied together")
        if producer_id is not None:
            if not isinstance(producer_id, str) or not producer_id:
                raise ValueError("producer_id must be a nonempty string")
            nonnegative_int(sequence, "sequence")
            previous = self.last_sequence.get((producer_id, p))
            if previous is not None:
                last, original_offset, original_record = previous
                if sequence == last:
                    if record != original_record:
                        raise ValueError("same sequence with different content")
                    return original_offset
                if sequence != last + 1:
                    raise ValueError("stale or skipped sequence")
            elif sequence != 0:
                raise ValueError("first sequence must be zero")
        offset = self.earliest[p] + len(self.logs[p])
        self.logs[p].append(record)
        if producer_id is not None:
            self.last_sequence[(producer_id, p)] = (sequence, offset, copy.deepcopy(record))
        return offset

    def fetch(self, p, offset, max_records=100):
        self.check_partition(p)
        nonnegative_int(offset, "offset")
        nonnegative_int(max_records, "max_records")
        if offset < self.earliest[p]:
            raise OffsetOutOfRange(f"partition {p}: offset {offset} is before the earliest retained offset {self.earliest[p]}")
        if offset > self.end_offsets()[p]:
            raise OffsetOutOfRange(f"partition {p}: offset {offset} is beyond the end")
        start = offset - self.earliest[p]
        return [(offset + i, copy.deepcopy(r)) for i, r in enumerate(self.logs[p][start:start + max_records])]

    def end_offsets(self):
        return [self.earliest[p] + len(log) for p, log in enumerate(self.logs)]

    def expire(self, before_ts):
        nonnegative_int(before_ts, "before_ts")
        for p, log in enumerate(self.logs):
            cut = 0
            while cut < len(log) and log[cut]["ts"] < before_ts:
                cut += 1
            self.earliest[p] += cut
            self.logs[p] = log[cut:]

class ConsumerGroup:
    """Separate volatile read positions and saved restart offsets in a toy group."""

    def __init__(self, topic, members):
        self.topic, self.committed = topic, {}
        self.rebalance(members)

    def rebalance(self, members):
        members = list(members)
        if any(not isinstance(m, str) or not m for m in members) or len(set(members)) != len(members):
            raise ValueError("members must be unique nonempty names")
        self.members = members
        self.assignment = {m: [p for p in range(len(self.topic.logs)) if p % len(members) == i] for i, m in enumerate(members)}
        self.position = {p: self.committed.get(p, self.topic.earliest[p]) for ps in self.assignment.values() for p in ps}

    def poll(self, member, max_records=100):
        records, next_positions = [], {}
        for p in self.assignment[member]:
            batch = self.topic.fetch(p, self.position[p], max_records)
            records += [(p, offset, r) for offset, r in batch]
            if batch:
                next_positions[p] = batch[-1][0] + 1
        self.position.update(next_positions)
        return records

    def seek(self, p, offset):
        self.topic.fetch(p, offset, 0)
        if p not in self.position:
            raise ValueError("partition is not assigned")
        self.position[p] = offset

    def commit(self, p, next_offset):
        self.topic.fetch(p, next_offset, 0)
        self.committed[p] = next_offset

    def lag(self):
        return {p: end - self.committed.get(p, self.topic.earliest[p]) for p, end in enumerate(self.topic.end_offsets())}

rng = random.Random(0)
orders = Topic(partitions=6)
for n in range(1, 3001):
    customer = "C1" if rng.random() < 0.3 else f"C{rng.randint(2, 500)}"
    orders.append(customer, {"order": n, "status": rng.choice(["paid", "shipped", "refunded"])}, ts=n)
print("records per partition:", [len(log) for log in orders.logs])
# records per partition: [319, 294, 351, 1206, 375, 455]
print("C1 lands in partition", orders.partition_for("C1"), "every time; C2 in", orders.partition_for("C2"))
# C1 lands in partition 3 every time; C2 in 5

In this model, expiry removes only a leading run of old records; an older timestamp behind a retained record waits, preserving offsets. A new group starts at the earliest retained offset. Every simulated rebalance resets assigned read positions to saved offsets, and commits are caller-controlled rather than fenced by group generation. These are explicit teaching policies, not a complete Kafka state machine.

Five partitions hold roughly 300–460 records each and the sixth holds 1,206, because the key hash sends every one of C1’s orders to the same place. That skew is a fact about the data, not the broker, and it reappears in the last exercise.

1. Ordering and the partition count. Read one customer’s records back and check their order. Read the topic round-robin across partitions and look at the order numbers. Then create a twelve-partition topic and count how many customers would hash to a different partition.

p = orders.partition_for("C7")
c7 = [r["value"]["order"] for _, r in orders.fetch(p, 0, 10_000) if r["key"] == "C7"]
print("C7's orders in partition", p, "read back as", c7, "- in order:", c7 == sorted(c7))
# C7's orders in partition 4 read back as [687, 1362, 1832] - in order: True
interleaved = []
for offset in range(3):
    for p in range(6):
        interleaved.append(orders.fetch(p, offset, 1)[0][1]["value"]["order"])
print("reading all partitions round-robin gives order numbers", interleaved)
# reading all partitions round-robin gives order numbers [1, 15, 8, 2, 7, 10, 4, 26, 18, 3, 9, 19, 5, 29, 38, 6, 14, 21]

wider = Topic(partitions=12)
moved = [f"C{n}" for n in range(1, 501) if orders.partition_for(f"C{n}") != wider.partition_for(f"C{n}")]
print("after growing the topic from 6 to 12 partitions,", len(moved), "of 500 customers hash to a different partition;",
      moved[0], "moves from", orders.partition_for(moved[0]), "to", wider.partition_for(moved[0]))
# after growing the topic from 6 to 12 partitions, 241 of 500 customers hash to a different partition; C6 moves from 2 to 8

2. Consumer groups. Form a group of three over the six partitions and print the assignment, then a group of eight, then a second group. Consume one round with commits, remove one member, rebalance, and check where the survivors continue.

group = ConsumerGroup(orders, ["worker-a", "worker-b", "worker-c"])
print("3 members over 6 partitions:", group.assignment)
# 3 members over 6 partitions: {'worker-a': [0, 3], 'worker-b': [1, 4], 'worker-c': [2, 5]}
crowd = ConsumerGroup(orders, [f"w{i}" for i in range(8)])
print("8 members over 6 partitions:", sum(1 for ps in crowd.assignment.values() if not ps), "members get nothing to read")
# 8 members over 6 partitions: 2 members get nothing to read
analytics = ConsumerGroup(orders, ["analytics-1"])
print("a second group starts at offset", analytics.committed.get(0, 0), "on every partition, regardless of the first group's progress")
# a second group starts at offset 0 on every partition, regardless of the first group's progress

for member in group.members:
    for p, offset, r in group.poll(member, max_records=200):
        group.commit(p, offset + 1)
print("committed after one round:", group.committed)
# committed after one round: {0: 200, 3: 200, 1: 200, 4: 200, 2: 200, 5: 200}
group.rebalance(["worker-a", "worker-c"])
print("worker-b died; new assignment:", group.assignment)
# worker-b died; new assignment: {'worker-a': [0, 2, 4], 'worker-c': [1, 3, 5]}
print("worker-c now continues partition 1 from offset", group.committed[1], "- nothing lost, nothing repeated")
# worker-c now continues partition 1 from offset 200 - nothing lost, nothing repeated

3. When to commit. Consume the first hundred offsets of every partition in batches, committing each partition’s batch either before or after processing it, and kill the process after 150 records on the first run. Restart from the committed offsets and count what was handled.

def run(strategy, crash_at=150):
    """Consume offsets 0-99 of every partition, committing per batch, and die after 150 records on the first run."""
    if strategy not in {"commit first", "commit after"}:
        raise ValueError("unknown commit strategy")
    nonnegative_int(crash_at, "crash_at")
    if crash_at > 600:
        raise ValueError("crash_at exceeds this 600-record exercise")
    committed, handled = {p: 0 for p in range(6)}, []
    for attempt in ("first run", "after restart"):
        try:
            for p in range(6):
                batch = orders.fetch(p, committed[p], 100 - committed[p])
                if strategy == "commit first":
                    committed[p] = 100
                for offset, r in batch:
                    if attempt == "first run" and len(handled) == crash_at:
                        raise RuntimeError("process killed")
                    handled.append((p, offset))
                committed[p] = 100
        except RuntimeError:
            continue
    return len(handled), len(set(handled))

for strategy in ("commit first", "commit after"):
    total, distinct = run(strategy)
    print(f"{strategy}: {total} records handled, {distinct} distinct, of 600; lost {600 - distinct}, repeated {total - distinct}")
# commit first: 550 records handled, 550 distinct, of 600; lost 50, repeated 0
# commit after: 650 records handled, 600 distinct, of 600; lost 0, repeated 50

4. The idempotent producer. Send 100 records to a fresh topic with a producer that resends every tenth record because its acknowledgement was lost, once without and once with a producer id and per-partition sequence numbers.

def produce_with_retries(topic, idempotent):
    sequence = defaultdict(int)
    for n in range(1, 101):
        key = f"C{n % 10}"
        for attempt in range(2):
            topic.append(key, {"order": n}, ts=n, producer_id="p1" if idempotent else None,
                         sequence=sequence[topic.partition_for(key)] if idempotent else None)
            if n % 10 != 0 or attempt == 1:
                break
        sequence[topic.partition_for(key)] += 1
    return sum(len(log) for log in topic.logs)

print("acks lost on every tenth send, plain producer:     ", produce_with_retries(Topic(3), idempotent=False), "records stored for 100 sent")
# acks lost on every tenth send, plain producer:      110 records stored for 100 sent
print("acks lost on every tenth send, idempotent producer:", produce_with_retries(Topic(3), idempotent=True), "records stored for 100 sent")
# acks lost on every tenth send, idempotent producer: 100 records stored for 100 sent

5. Retention, the slow consumer, and replay. On a copy of the topic, let a nightly consumer commit after 50 records per partition, then expire everything before timestamp 1500. Poll again, handle the error by resetting to the earliest retained offset, and count what the consumer will never see. Then replay what is retained to rebuild each order’s current status, and build a compacted view with the latest record per key.

import copy

log = copy.deepcopy(orders)
slow = ConsumerGroup(log, ["nightly"])
for p, offset, r in slow.poll("nightly", max_records=50):
    slow.commit(p, offset + 1)
log.expire(before_ts=1500)
print("after retention removed everything before ts 1500, earliest offsets are", log.earliest)
# after retention removed everything before ts 1500, earliest offsets are [148, 144, 183, 604, 188, 232]
try:
    slow.poll("nightly")
except OffsetOutOfRange as error:
    print("the slow consumer's next poll fails:", error)
# the slow consumer's next poll fails: partition 0: offset 50 is before the earliest retained offset 148
for p in range(6):
    reset = max(slow.committed[p], log.earliest[p])
    slow.seek(p, reset)
    slow.commit(p, reset)
print("reset to earliest and carry on; records it will never see:", sum(log.earliest) - 300)
# reset to earliest and carry on; records it will never see: 1199

status = {}
for p in range(6):
    for _, r in log.fetch(p, log.earliest[p], 10_000):
        status[r["value"]["order"]] = r["value"]["status"]
print("replaying what is retained rebuilds the status of", len(status), "orders:", Counter(status.values()))
# replaying what is retained rebuilds the status of 1501 orders: Counter({'refunded': 510, 'paid': 497, 'shipped': 494})

def compacted(topic):
    latest = {}
    for p in range(len(topic.logs)):
        for _, r in topic.fetch(p, topic.earliest[p], len(topic.logs[p])):
            if r["value"] is None:
                latest.pop(r["key"], None)
            else:
                latest[r["key"]] = r["value"]
    return latest

print("a compacted view keeps the last record per key:", len(compacted(log)), "customers from", sum(len(l) for l in log.logs), "records")
# a compacted view keeps the last record per key: 441 customers from 1501 records

6. Lag and the hot key. Let a group of three drain the topic at 100 records per partition per poll, printing lag by partition as it goes, and compare the number of polls with what an even spread would have needed.

group = ConsumerGroup(orders, ["worker-a", "worker-b", "worker-c"])
polls = 0
while any(group.lag().values()):
    polls += 1
    for member in group.members:
        for p, offset, r in group.poll(member, max_records=100):
            group.commit(p, offset + 1)
    if polls in (1, 4, 8):
        print(f"after {polls} polls of 100 per partition, lag by partition: {group.lag()}")
# after 1 polls of 100 per partition, lag by partition: {0: 219, 1: 194, 2: 251, 3: 1106, 4: 275, 5: 355}
# after 4 polls of 100 per partition, lag by partition: {0: 0, 1: 0, 2: 0, 3: 806, 4: 0, 5: 55}
# after 8 polls of 100 per partition, lag by partition: {0: 0, 1: 0, 2: 0, 3: 406, 4: 0, 5: 0}
print("drained after", polls, "polls; the same 3000 records spread evenly would have taken", (sum(orders.end_offsets()) + 599) // 600)
# drained after 13 polls; the same 3000 records spread evenly would have taken 5
print("partition", max(range(6), key=lambda p: len(orders.logs[p])), "holds the hot key C1; at this fixed rate, extra members cannot split its assignment")
# partition 3 holds the hot key C1; at this fixed rate, extra members cannot split its assignment

7. Choose the service. For the subscription company’s four topics, decide between self-managed Kafka, a managed Kafka, Redpanda, Kinesis, and Pub/Sub, and write the paragraph that justifies the choice in terms of retention, replay, ecosystem, and who operates it.

References

Kafka behavior and Java client defaults above are scoped to the linked 4.1 documentation. Service documentation checked on 2026-09-09; verify the version and service settings used in your deployment.

• Kafka 4.1 design: replication, transactions, compaction
• Kafka 4.1 producer configuration
• Kafka 4.1 consumer configuration
• KafkaConsumer API: position, commits, rebalances
• Kafka 4.1 KRaft operations
• Kinesis Data Streams concepts and retention
• Pub/Sub topic retention
• Pub/Sub replay and seek
• Pub/Sub message ordering
• Redpanda introduction