Part 27 Project Complete

"The Backbone" — Kafka for Analytics

Rohan has run Kafka for years, for services. Analytics asks different questions of it: which key keeps order, how long the log remembers, which schema change a registry must refuse, and whether a reader sees an aborted transaction. Topics as code, Avro through Apicurio Registry from Java 25 producers, a refused rename, the Iceberg Kafka Connect sink built from source, and Flink SQL reading the same bytes.

October 31, 2026 20 min read GitHub

Skim Card

One thing to remember: Kafka for analytics is a contract, not a pipe: keys, retention and schemas are designed before the first record, and every reader asks for committed data.

Key by the entity whose order matters (tills for POS, visitors for clicks); partitions can only grow, and growing remaps keys.

Under FULL_TRANSITIVE, the registry refused the page → page_path rename before one event was written; the additive v2 passed.

Transactions protect only read_committed readers: an aborted upload’s 1,000 basket events reached Kafka’s default consumer, the Iceberg sink and Flink SQL.

acks=all and idempotence are the Java client’s defaults; a conflicting setting silently switches idempotence off unless you enabled it explicitly.

Apicurio’s default serializer writes Confluent framing with contentIds; Flink’s avro-confluent format reads it through /apis/ccompat/v7.

The Iceberg sink landed every committed record; on a quiet topic, Connect’s offset flush interval, not the commit interval, paced commits.

Share groups (production-ready since Kafka 4.2) suit independent, slow work; ordered analytics state needs consumer groups.

Use … When …
Iceberg Kafka Connect sink Append-only event topics into Iceberg, minutes of latency, Connect already running
Flink’s Iceberg sink You deduplicate, join or aggregate on the way in (Part 28)
Spark Structured Streaming + MERGE CDC topics that fold into current state (Part 16)
A managed topic-to-table product You already run that vendor’s Kafka and accept its catalog and pricing

Verified against: Apache Kafka 4.4.0 (Java 25 clients), Apicurio Registry 3.3.3, Kafka Connect 4.3.1 (Debezium 3.7.0.Final image), Apache Iceberg 1.12.0 Kafka Connect sink, Apache Flink 2.2.1 with flink-sql-avro-confluent-registry 2.2.1 · vendor docs · 30 October 2026

Jump to: Concepts · Hands-On · Landscape · Design Drill

Use …	When …
Iceberg Kafka Connect sink	Append-only event topics into Iceberg, minutes of latency, Connect already running
Flink’s Iceberg sink	You deduplicate, join or aggregate on the way in (Part 28)
Spark Structured Streaming + `MERGE`	CDC topics that fold into current state (Part 16)
A managed topic-to-table product	You already run that vendor’s Kafka and accept its catalog and pricing

Story Opening

Season 3’s streaming jobs need Larkspur’s tills and clicks on Kafka. Rohan did it the order-service way (topics created on first use, whatever key was handy), replayed the till uploads of 15 August and pointed the lake’s new sink at the topic.

The lake had 23,276 basket events; the raw zone, 22,276. One upload had failed half-way and been retried, and the producer had aborted the first attempt, exactly as written. The sink landed it anyway.

“Kafka did what you asked,” Ingrid said. “The aborted records are in the log; a reader decides whether to see them. For a service, a duplicate is a retry. For finance, it’s a thousand baskets that never happened.”

The Quick Map

Backend instinct	Data engineering reality	Why it matters
Topics appear when a producer first writes	Topics are code: partitions, key, retention and cleanup decided before the first record	The key fixes ordering forever; partitions can only grow, and growing remaps keys
JSON with a `version` field; consumers cope	A schema registry with a compatibility rule refuses a breaking schema before a record exists	A lake table reads years of history with today’s schema
A transaction makes the write safe	A transaction protects only `read_committed` readers; Kafka’s default reader sees aborted records	The lab’s sink and Flink SQL both read them until told not to
`acks=1` is fast enough	`acks=all` + idempotence (Java client defaults since 3.0; idempotence effective from 3.0.1/3.1.1/3.2, KAFKA-13598) + `min.insync.replicas`	A lost or doubled event is a wrong number, not a retry
Kafka is the database	Kafka keeps days; the lake keeps years	Retention is a replay budget, not an archive

Concepts, with use cases

Of Part 2’s five promises, this part strengthens completeness: every event the source sent, once, in a shape every reader can still read next year. The Log Never Lies covers the log itself, ordering per key and registries for CDC topics; this part starts where analytics differs.

Keys and partitions

A topic is split into partitions; a record’s key is hashed to pick one, and Kafka keeps order only within a partition. So the key answers one question: whose events must stay in order? The partition count caps how many consumers in a group read in parallel.

What breaks with the wrong key. A null key spreads records with no order at all, and a till’s re-send can land far from its original. Key by store, and Larkspur’s flagship stores make hot partitions. Larkspur: POS is keyed by till_id (a till’s uploads stay in order; a re-sent basket comes from the same till, so its copies share a partition), clickstream by anonymous_id (Part 6’s sessions need one visitor’s events in order). Elsewhere: a ride-hailing platform keys trip events by trip ID; keyed by city, one Saturday night in one city saturates a partition.

Partitions can be added, never removed, and adding them sends existing keys to new partitions, breaking per-key order at that moment. Size for the parallelism of two years from now.

Retention, compaction and tiered storage

Retention (cleanup.policy=delete, retention.ms) drops whole segments older than the limit. Compaction (cleanup.policy=compact) keeps at least the latest record per key, which suits a changelog of current state (prices, store attributes) and nothing that must be summed. Tiered storage, production-ready since Kafka 3.9, moves old segments to object storage behind the same topic; it doesn’t support compacted topics, and you supply a RemoteStorageManager plugin.

What breaks without a policy. Infinite retention makes Kafka an expensive, unindexed archive; too short, and a sink down for a weekend has nothing to replay. Larkspur: POS keeps 7 days, clickstream 3: long enough to replay after an outage, while the lake keeps the history. Elsewhere: a telecom keeps network events for 72 hours on brokers and 30 days in a tier, for incident forensics.

Schemas, a registry and compatibility

A schema registry stores versioned schemas (Avro, Protobuf or JSON Schema) per subject, usually <topic>-value. Producers write a 5-byte prefix in the value (magic byte + 4-byte ID) instead of the schema; readers fetch the writer’s schema by that ID. A compatibility rule decides which new versions the registry accepts: BACKWARD (new readers read old data), FORWARD (old readers read new data), FULL (both), each against the latest version or, with _TRANSITIVE, against every version.

What breaks without one. The producer’s release is the consumer’s outage, found weeks later: Part 20 met the 15 August rename of page to page_path after it had landed. Larkspur: both topics run FULL_TRANSITIVE, because analytics has both readers at once: the lake and every replay read years of data with the newest schema (backward, transitive), while last month’s Flink jobs read today’s data with their old one (forward). Optional additions pass; renaming or removing a required field doesn’t. Elsewhere: an ad exchange’s bid logs feed billing, so a dropped field silently zeroes an advertiser’s invoice.

Producer settings that decide correctness

acks=all: a write succeeds once every in-sync replica has it; with min.insync.replicas=2 on three replicas, losing one broker loses nothing acknowledged.
Idempotence (enable.idempotence=true): the broker drops a retried batch it already wrote (by producer ID and sequence number), keeping order with up to 5 requests in flight. Both are the 4.4 client’s defaults, but a conflicting setting (say acks=1) silently switches idempotence off; set it explicitly and the conflict becomes an error.
Transactions (transactional.id): records to several partitions commit or abort together, and a restarted producer with the same ID fences its zombie predecessor.

What breaks without them. With acks=1 a failed leader loses its last writes; retries without idempotence duplicate them. Larkspur: a till’s upload is a transaction, so a crash half-way leaves nothing a careful reader would count. Elsewhere: a payments ledger writes a debit and a credit to two topics in one transaction.

A consumer group splits partitions among its members: per-key order, at most one active consumer per partition, progress as one offset per partition. A share group (KIP-932, “production-ready in Apache Kafka 4.2”; dead-letter topics in 4.4) lets many consumers read the same partition: each record is acquired, then acknowledged (ACCEPT), released for redelivery or rejected, like a queue.

What breaks if you pick the wrong one. A share group gives up order, so a sessionizer or an upsert fed by one is wrong. A consumer group can’t scale past the partition count, which caps a slow per-record job (sending receipts, calling an API). Larkspur: every analytics reader is a consumer group; the loyalty team’s “email the receipt” workers could be a share group. Elsewhere: a video platform’s transcoding workers, where each upload is independent and slow.

Getting topics into the lakehouse

The Iceberg Kafka Connect sink (part of Apache Iceberg) appends records to Iceberg tables through a commit coordinator; with no upsert or delete writer in 1.12 it suits event topics, not CDC (Part 16 folds those with Spark). Flink’s Iceberg sink commits on checkpoints (Part 28; the SinkV2-based variant is experimental). Managed products do the same for their own Kafka (Landscape).

What breaks without a landing. Every analyst reads Kafka with their own consumer and their own idea of “the day”. Larkspur: POS and clickstream land in stream.* tables, partition and offset kept as lineage by the sink’s KafkaMetadataTransform (experimental). Elsewhere: an IoT fleet lands telemetry into day-partitioned tables, and nobody queries the brokers.

Under the Hood

What a transaction leaves in the log

A transactional producer writes records into the partitions like any producer; committing or aborting appends a control record (a marker) to every partition the transaction touched, and the broker keeps an index of aborted transactions per segment. A read_uncommitted consumer returns every data record. A read_committed consumer reads only up to the last stable offset, just before the first transaction still open, and drops records whose transaction aborted. Aborted records are skipped, never deleted; they leave with retention (or compaction).

So offsets aren’t counts. In the lab’s run, 24 transactions over 6 partitions left 144 markers: end offsets sum to 23,420 for 23,276 data records, of which 22,276 are committed.

What the registry puts in each record

Apicurio’s 3.3.3 serializer, at its defaults, writes the Confluent wire layout (0x0, the 4-byte ID, the Avro body). The ID is the schema’s contentId (apicurio.registry.use-id=contentId), in the payload, not in headers (apicurio.registry.headers.enabled=false). Apicurio’s Confluent-compatible API (/apis/ccompat/v7) serves the schema under subject <topic>-value with id = contentId, nested records as Confluent references, which Confluent clients that support references resolve (Flink’s did).

How the Iceberg sink commits

Each sink task writes Parquet files as records arrive, not yet visible. The task holding the first partition also runs a coordinator: every iceberg.control.commit.interval-ms it puts START_COMMIT on the control topic. Each worker answers with the files it wrote (DATA_WRITTEN, then DATA_COMPLETE) in one Kafka transaction that also commits its consumer offsets, so from then on those records won’t be read again and their file list is safe in the log. The coordinator commits one Iceberg snapshot per table and records in the snapshot summary how far it has read the control topic, so after a crash it never commits the same files twice. A round that doesn’t hear from every worker within iceberg.control.commit.timeout-ms (30 s) commits nothing.

Hands-On

The code is in the every-record-counts folder of the companion-workspaces repository (tag every-record-counts/part-27). New in the stream profile: Apicurio Registry 3.3.3, its schemas in the lab PostgreSQL (database apicurio; the default in-memory store forgets them on restart), and the Iceberg sink in the existing Connect worker. Java 25 producers live in the Gradle module kafka-backbone. Peak container memory was 3.48 GiB (the Flink step).

JAVA_HOME=/path/to/jdk21 make iceberg-sink   # build the sink once (Iceberg's build needs JDK 17 or 21)
make up-backbone                             # core + Kafka + Apicurio + Connect with the sink plugin
uv run python -m scripts.part27_backbone reset
uv run python -m scripts.part27_backbone topics
uv run python -m scripts.part27_backbone registry
./gradlew -q :kafka-backbone:run --args="pos --day 2026-08-15 --abort 3"
./gradlew -q :kafka-backbone:run --args="count --topic larkspur.pos.basket-events"
./gradlew -q :kafka-backbone:run --args="clicks --day 2026-08-14 --schema v1"
./gradlew -q :kafka-backbone:run --args="clicks --day 2026-08-15 --schema v2_rename"
./gradlew -q :kafka-backbone:run --args="clicks --day 2026-08-15 --schema v2"
uv run python -m scripts.part27_backbone versions
uv run python -m scripts.part27_backbone sink --isolation read_uncommitted
uv run python -m scripts.part27_backbone lake
uv run python -m scripts.part27_backbone sink   # read_committed, from empty tables again
uv run python -m scripts.part27_backbone lake
docker compose --profile core --profile stream stop connect   # Kafka and Apicurio stay up
make up-flink
uv run python -m scripts.part27_backbone flink

Step 0: the sink, built from source

Iceberg publishes no runtime package for the sink, so kafka/iceberg-sink/build.sh builds only that module from the checksum-verified 1.12.0 source release (3 min 19 s; a 190 MB plugin, mounted read-only into the Debezium image’s Connect worker).

Step 1: topics as code, a strict registry

Topic: larkspur.pos.basket-events PartitionCount: 6 ReplicationFactor: 1 Configs: min.insync.replicas=1,cleanup.policy=delete,retention.ms=604800000
Topic: larkspur.clickstream.events PartitionCount: 6 ReplicationFactor: 1 Configs: min.insync.replicas=1,cleanup.policy=delete,retention.ms=259200000
Topic: control-iceberg PartitionCount: 1 ReplicationFactor: 1 Configs: min.insync.replicas=1,retention.ms=604800000
global rule VALIDITY: FULL
global rule COMPATIBILITY: FULL_TRANSITIVE

Auto-creation is off on the broker, so a mistyped topic name fails instead of creating a topic with defaults. Replication factor 1 is the one-broker lab; production runs 3 with min.insync.replicas=2.

Step 2: a transactional upload, and what readers see

PosProducer replays the 15 August uploads from the raw zone, one transaction per 1,000 basket events, keyed by till. --abort 3 sends the third batch, aborts it as a crashed uploader would leave it, and sends it again. The settings from Concepts:

c.put(ProducerConfig.ACKS_CONFIG, "all");                 // written to every in-sync replica before success
c.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);    // broker drops a retried batch it already has
c.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AvroKafkaSerializer.class.getName());
c.put("apicurio.registry.url", REGISTRY);
c.put("apicurio.registry.auto-register", true);   // lab convenience; production registers from CI

22,276 basket events from upload_date=2026-08-15 -> larkspur.pos.basket-events, keyed by till_id
transactions: 23 committed, 1 aborted (1,000 records written, then aborted and re-sent)
took 1.5 s

larkspur.pos.basket-events: 6 partitions, end offsets sum to 23,420
end offset per partition: 3,865 3,868 3,789 3,917 4,117 3,864
read_uncommitted consumer reads 23,276 records
read_committed   consumer reads 22,276 records

Same topic, two answers: TopicCounter changes only isolation.level. Keyed by till, no partition runs hot (3,789 to 4,117 offsets).

Step 3: the registry refuses a rename

Clickstream v1 has a required page. The web team’s first v2 renames it; the compatible v2 keeps page and adds an optional page_path (null default), the expand step of an expand-and-contract migration:

schema v1: 61,554 of 61,554 events from event_date=2026-08-14 -> larkspur.clickstream.events (0 failed)
schema v2_rename refused on the first record:
  HTTP 400: Incompatible artifact: larkspur.clickstream.events-value [AVRO], num of incompatible diffs: {2}, list of diff types: [page at /fields/8, page_path at /fields/8]
  - page at /fields/8
  - page_path at /fields/8
schema v2_rename: 0 of 62,958 events from event_date=2026-08-15 -> larkspur.clickstream.events (0 failed)
schema v2: 62,958 of 62,958 events from event_date=2026-08-15 -> larkspur.clickstream.events (0 failed)

The serializer runs inside send(), so the refusal is an exception before any record exists: zero events written. Why a rename fails in each direction, and what no mode catches, is in The Log Never Lies.

registry v3: larkspur.clickstream.events-value version 1  globalId 23  contentId 5
registry v3: larkspur.clickstream.events-value version 2  globalId 24  contentId 7
ccompat v7:  subject larkspur.clickstream.events-value version 2  id 7

The refused schema never became a version. The Confluent-compatible ID is the contentId the producers wrote; with apicurio.registry.use-id=globalId, Confluent clients would resolve another schema or none.

Step 4: land both topics in Iceberg

One connector reads both topics. KafkaMetadataTransform (experimental) adds each record’s topic, partition, offset and Kafka timestamp as columns, and the sink routes on the topic column (kafka/connectors/iceberg-sink.json; trimmed: tasks.max 1, the key converter, the catalog and S3 keys):

{
  "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
  "topics": "larkspur.pos.basket-events,larkspur.clickstream.events",
  "consumer.override.isolation.level": "read_committed",
  "value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
  "value.converter.apicurio.registry.url": "http://apicurio:8080/apis/registry/v3",
  "transforms": "kafka",
  "transforms.kafka.type": "org.apache.iceberg.connect.transforms.KafkaMetadataTransform",
  "iceberg.tables.route-field": "_kafka_metadata_topic",
  "iceberg.tables": "stream.pos_basket_events,stream.click_events",
  "iceberg.table.stream.pos_basket_events.route-regex": "larkspur\\.pos\\.basket-events",
  "iceberg.table.stream.click_events.route-regex": "larkspur\\.clickstream\\.events",
  "iceberg.tables.auto-create-enabled": "true",
  "iceberg.tables.evolve-schema-enabled": "true",
  "iceberg.tables.default-partition-by": "day(event_ts)",
  "iceberg.control.commit.interval-ms": "10000"
}

Key	What it does here
`consumer.override.isolation.level`	Connect’s sink consumer reads `read_uncommitted` unless told otherwise (the worker’s default override policy, `All`, permits this)
`value.converter`	The Apicurio converter in the Debezium image (`ENABLE_APICURIO_CONVERTERS=true`): reads the contentId, fetches the schema
`auto-create-enabled`, `evolve-schema-enabled`, `default-partition-by`	Tables created from the first record’s schema, partitioned by event day; new optional fields become columns

sink deploys it on empty tables and waits; lake compares them with the raw zone. First without the override, then with it:

# sink --isolation read_uncommitted, then lake
table                       raw lines  lake rows  event_ids snapshots  files
stream.pos_basket_events       22,276     23,276     22,276         1      2
stream.click_events           124,512    124,512    124,160         1      3

# sink (read_committed, from empty tables), then lake
stream.pos_basket_events       22,276     22,276     22,276         1      2

stream.click_events: page_path is field 20, optional=True, added by the sink
  2026-08-14: 61,554 events, 0 with page_path
  2026-08-15: 62,958 events, 34,726 with page_path

The thousand aborted basket events landed under the default and vanished with the override; they repeat real events’ IDs, so a dedup on event_id would have hidden the bug, not fixed it. (The 352 repeated clickstream IDs are the generator’s documented SDK retries.) Each row keeps _kafka_metadata_partition, _kafka_metadata_offset and _kafka_metadata_timestamp (the producer’s send time, so ingestion time) beside the till’s event_ts: two clocks, two columns, as in Part 26.

Both runs landed everything in a single commit, by 121 s after the connector was created (checked every 10 s), with a 10-second commit interval. Meanwhile the worker logged Commit timeout reached and committed to 0 table(s) every 30 s while the task sat in poll() (Gotchas). With the worker’s flush interval at 5 s (Connect recreated with CONNECT_OFFSET_FLUSH_MS=5000, which compose.yaml passes on), the same load landed by 30 s.

Step 5: Flink SQL reads the same bytes

Part 28 builds on Flink SQL reading these topics, so this step settles how. The lab Flink image now also carries flink-sql-avro-confluent-registry-2.2.1.jar (checksum-pinned in flink/Dockerfile), and the table points the format at Apicurio’s Confluent-compatible API:

CREATE TABLE pos_basket_events ( … ) WITH (
  'connector' = 'kafka',
  'topic' = 'larkspur.pos.basket-events',
  'properties.bootstrap.servers' = 'kafka:29092',
  'scan.startup.mode' = 'earliest-offset',
  'scan.bounded.mode' = 'latest-offset',
  'format' = 'avro-confluent',
  'avro-confluent.url' = 'http://apicurio:8080/apis/ccompat/v7'
);

-- The same table read the other way: only records of committed transactions.
CREATE TABLE pos_committed WITH ('properties.isolation.level' = 'read_committed') LIKE pos_basket_events;

+-------------------+--------+-----------+---------+
|            reader | events | event_ids | baskets |
+-------------------+--------+-----------+---------+
| as declared above |  23276 |     22276 |   22197 |
|    read_committed |  22276 |     22276 |   22197 |
+-------------------+--------+-----------+---------+

Flink decoded every record, nested references included, with no Confluent server in the stack. The connector page says read_committed is “the default value”, but that sentence swapped its two values around connector 3.3 (Flink 1.17’s page named read_uncommitted); the 5.0.0 source sets nothing, so Kafka’s read_uncommitted applied and Flink counted the aborted batch. The other number to keep: 22,197 baskets for 22,276 basket events, 79 till re-sends that day, which Part 28 removes in Flink SQL.

The Landscape

Current product names as of 30 October 2026. A cell that couldn’t be verified says so.

Need	Open source	AWS	GCP	Azure	Snowflake / Databricks	Notes
Event log	Apache Kafka 4.4	Amazon MSK; Amazon Kinesis Data Streams	Google Cloud Managed Service for Apache Kafka; Pub/Sub	Azure Event Hubs (Kafka endpoint)	Neither runs a log; both read Kafka	Diskless topics (KIP-1150) are accepted but in no Kafka release
Schema registry	Apicurio Registry 3.3 (Apache 2.0); Karapace (Apache 2.0); Confluent Schema Registry (Confluent Community License)	AWS Glue Schema Registry (Avro, JSON Schema, Protobuf)	Schema registry in Managed Service for Apache Kafka (Confluent-compatible API; Avro, Protobuf; status not verified); Pub/Sub schemas	Event Hubs Schema Registry (Avro, JSON Schema, Protobuf; evolution checks for Avro only, modes Backward, Forward, None)	Databricks `from_avro` reads a Confluent-compatible registry; Snowflake: not verified (2026-10-30)	The wire format (magic byte + 4-byte ID) is the interoperability point
Topics into lakehouse tables	Iceberg Kafka Connect sink (build from source, append-only); Flink Iceberg sink; Spark Structured Streaming	Amazon Data Firehose (MSK source → Apache Iceberg Tables; not MSK Serverless)	BigQuery Sink connector on Managed Service for Apache Kafka’s Connect clusters (BigQuery tables, or Iceberg tables in BigQuery)	Fabric eventstreams (Apache Kafka source (preview) → Lakehouse, Delta)	Snowflake Connector for Kafka v4 (GA; into Snowflake-managed Iceberg tables, no schema evolution for them); Databricks Structured Streaming and Lakeflow pipelines	Confluent Tableflow (Delta GA, Iceberg “available”); Redpanda Iceberg Topics (enterprise licence)

Tips & Gotchas

Gotcha — every reader defaults to read_uncommitted. Kafka’s consumer, Connect’s sink and Flink SQL all read the aborted batch; set read_committed in every reader and test with an aborted batch.

Gotcha — a quiet topic slows the Iceberg sink’s commits. An idle Connect sink task waits in poll() until its next offset commit (offset.flush.interval.ms, 60 s in the Debezium image) and only answers the coordinator between polls, so commit rounds time out and commit nothing. Lower the worker’s flush interval, or expect a minute or two, whatever commit.interval-ms says.

Gotcha — Apicurio registers nested records as artifacts of their own. The 3.3.3 serializer created com.larkspur.pos.Tender, …BasketLine and com.larkspur.clickstream.Consent beside the topic subjects. A second topic using a record named Tender shares that artifact and its compatibility history. apicurio.registry.dereference-schema=true registers one self-contained schema per subject instead (from the 3.3.3 source; not run in the lab).

Tip — register schemas from CI, not from producers. The lab’s producers auto-register for brevity. In production, CI registers the schema (and fails the build on a refusal) and producers run with apicurio.registry.auto-register=false, so an unreviewed schema can’t appear at 2 a.m.

Troubleshooting

Symptom	Likely cause	How to confirm	Fix
Connect log: `committed to 0 table(s)` every 30 s, rows appear minutes later	Sink tasks idle in `poll()` (see Gotchas)	`Commit timeout reached` lines; `offset.flush.interval.ms` in the worker’s properties	Lower the worker’s `offset.flush.interval.ms` (here: recreate `connect` with `CONNECT_OFFSET_FLUSH_MS=5000`)
Producer fails with `RuntimeException: …RuleViolationProblemDetails` and no reason	The registry refused the schema; Apicurio’s client exception carries only its class name	The exception’s `getCauses()`, or the registry’s log	Log the causes (`ClickProducer` does), then fix the schema

Design Drill

The question: “We’re a payments company processing 50,000 events a second (authorisations, captures, refunds, chargebacks) in the EU, India and the US. Fraud scoring needs them in seconds, the ledger must never miss or double one, and analysts want years of history. Design the event backbone.”

Clarify first

Peak versus average, and event size? At 1 KB, 50,000 a second is 50 MB/s, and replication factor 3 makes it 150 MB/s of broker writes across the regions, before any peak.
Which data may leave its region (card data, personal data, India’s payment-data storage rules)?
Ordering: per payment, per card, per merchant? Which consumer needs which?
Recovery targets: may a region lose acknowledged events if it fails? How long may fraud scoring lag?
Who produces: the payment services directly, or an outbox read by CDC?

The shape of a strong answer

One Kafka cluster per region, three availability zones, replication factor 3, min.insync.replicas=2, producers with acks=all and idempotence: no acknowledged event is lost to one broker or zone.
Producers publish through the transactional outbox and CDC (The Log Never Lies), so a payment and its event commit together.
Key by payment ID: every state change of one payment in order, no hot merchants. Partition for peak consumer parallelism (say 3× the average), knowing partitions can only grow and growing remaps keys.
Avro with a registry under FULL_TRANSITIVE, schemas registered from CI; one topic per event family with an envelope that carries event time, ingestion time and an event ID.
Fraud reads in-region in a consumer group, and its per-card features re-key by card inside the stream processor; notifications and other slow side effects use share groups; the ledger loader reads read_committed and is idempotent by event ID.
Cross-region: replicate only what may leave (MirrorMaker 2, part of Apache Kafka, or a vendor’s cluster linking), into an analytics region; mask or tokenise before it crosses.
Retention 7 days on brokers, longer in tiered storage for forensic replays; years in Iceberg tables by event day, landed exactly once with partition and offset as lineage, and reconciled daily against the ledger (Part 21).

Trade-offs to say out loud

Per-region clusters keep latency and residency right, at the price of a global view that is minutes old; a stretched cluster across continents pays cross-ocean latency on every acknowledgement.
FULL_TRANSITIVE slows schema changes; the alternative is breaking a fraud model on a Friday.
More partitions buy parallelism and cost rebalance time, file handles and small files downstream.

Common weak answers

One global cluster, and “Kafka handles replication”.
Key by merchant: one large merchant becomes one hot partition.
“Exactly-once is on” without saying that consumers read read_committed and the ledger write is idempotent.

Story Closing

Rohan reran the day with read_committed everywhere: 22,276 basket events in the raw zone, 22,276 in the lake, and a registry that had refused the rename before a single renamed event existed.

“The topic is right,” Ingrid said. “The baskets aren’t: seventy-nine are re-sends from tills that reconnected.”

“Part 21 drops them in the nightly run,” Rohan said.

“Nightly. The stock-out alerts can’t wait for the night.” In Part 28, Rohan meets Flink SQL: re-sent baskets deduplicated as they stream, each line joined to the price in force when it sold, the result written to Iceberg.

This is Part 27 of “Every Record Counts: Data Engineering for Backend Engineers” — Season 3: Real Time, Serving and AI.

Series Progress 36/36 parts

0 Part 0: "The Field Guide" — How to Read This Series Complete 1 Part 1: "Three Houses for Data" — Warehouse, Lake, Lakehouse Complete 2 Part 2: "Think in Pipelines" — The Data Engineer's Mental Model Complete 3 Part 3: "Bytes on Disk" — File Formats, Parquet and Object Storage Complete 4 Part 4: "Grain Before Glory" — Dimensional Modeling Complete 5 Part 5: "Other Shapes" — Data Vault, Wide Tables and Modeling Trade-offs Complete 6 Part 6: "SQL You Will Actually Use" — Analytical SQL for Data Engineers Complete 7 Part 7: "The Table That Remembers" — Apache Iceberg in Depth Complete 8 Part 8: "Format Wars, Ceasefire" — Delta, Hudi, Paimon, DuckLake and Interoperability Complete 9 Part 9: "Who Owns the Truth" — Catalogs and the Iceberg REST Protocol Complete 10 Part 10: "Where the Bytes Live" — Partitioning, File Layout and Table Maintenance Complete 11 Part 11: "You Might Not Need a Cluster" — DuckDB, Polars and Single-Node Analytics Complete 12 Part 12: "Shuffle, Spill and Skew" — Distributed Batch with Spark 4.1 Complete 13 Part 13: "Engines at Scale" — Trino, Cloud Warehouses and Real-Time OLAP Complete 14 Part 14: "Design Review I" — The Storage and Compute Layer Complete 15 Part 15: "Getting Data In" — Batch Ingestion: Files, APIs and Databases Complete 16 Part 16: "Changes, Not Snapshots" — CDC into the Lakehouse Complete 17 Part 17: "Transformations as Code" — dbt Fundamentals Complete 18 Part 18: "dbt at Scale" — Incremental Models, CI and Environments Complete 19 Part 19: "Who Runs What, When" — Orchestration with Airflow 3 Complete 20 Part 20: "Trust, but Test" — Data Quality and Data Contracts Complete 21 Part 21: "Every Record Counts" — Completeness and Correctness Engineering Complete 22 Part 22: "Know Before They Do" — Observability, Lineage and SLAs Complete 23 Part 23: "Locks on the Lake" — Security, Privacy and Compliance Complete 24 Part 24: "The Bill Always Comes" — Cost and Performance Engineering Complete 25 Part 25: "Design Review II" — The Batch Reporting Platform, End to End Complete 26 Part 26: "Time Is a Lie" — Streaming Fundamentals Complete 27 Part 27: "The Backbone" — Kafka for Analytics Complete 28 Part 28: "Streaming SQL" — Apache Flink I Complete 29 Part 29: "Stateful Streams in Java" — Apache Flink II Complete 30 Part 30: "The Streaming Lakehouse" — Real-Time Tables and OLAP Complete 31 Part 31: "Last Mile" — Serving Data: Semantic Layers, BI and Data APIs Complete 32 Part 32: "Feeding the Models" — Data Engineering for ML and AI Complete 33 Part 33: "Context Is the Product" — Metadata, Agents and AI-Assisted Data Engineering Complete 34 Part 34: "Platform, Not Pipelines" — Data Platform Engineering and Organisation Complete 35 Part 35: "Design Review III" — The Whole Platform, and the Interview Playbook Complete