In this lab, you will publish an OrderPlaced event as JSON from Java. You will then test how two consumer versions handle three contract changes. You will add an optional field and confirm that nothing breaks, rename a required field and watch a consumer stop, then repair the damage with a second topic and dual publication. You will also separate a CapturePayment command from the PaymentCaptured event it causes, compare a thin ProductUpdated notification with a state-carrying version, and move the record key from order to customer to see what the ordering guarantee becomes.

The order examples keep the same amounts and customer assignments while their representation changes: ord-1042 and ord-1043 belong to cust-208, and ord-1044 belongs to cust-311. Amounts use integer currency minor units. For the versioning experiment, eventId identifies one immutable representation, occurrenceId identifies the underlying order acceptance, and orderId identifies the order. Retrying a representation preserves its event ID and payload; another version gets its own event ID and shares the occurrence ID. The command and catalog phases define their own contracts below.

Use the single-broker kafka-local container for the broker and the CLI tools, which run inside the container with docker exec. The Java code runs on the host with Java 21 and Maven, using kafka-clients:4.3.1, slf4j-simple:1.7.36, and jackson-databind:2.20.0 for JSON. Jackson is a separate library, not a Kafka serializer; we pin the version so the record and unknown-property behavior you observe is reproducible. There is no Schema Registry. The consumers test how these specific records behave in application code. A registry would separately compare schema definitions; these are different checks.

Use fresh lab topics, fresh consumer groups, and an empty observations directory. If you ran an earlier lab on kafka-local, finish that lab’s cleanup first. The expected counts assume no earlier records or committed progress. Start a new broker:

Create the three-partition order topics:

Create the single-partition payment and catalog topics:

Start from the producer project layout from the producers module. Add com.fasterxml.jackson.core:jackson-databind:2.20.0 to the dependencies and set <mainClass> to com.example.kafka.Lab. Lab reads its first argument as a subcommand and calls the matching class, so every program runs as mvn compile exec:java -Dexec.args="<subcommand> ...".

Every consumer uses StringDeserializer, enable.auto.commit=false, auto.offset.reset=earliest, and commits with commitSync(records.nextOffsets()) after processing each batch. Every producer uses StringSerializer, acks=all, and enable.idempotence=true, and prints the partition and offset from RecordMetadata for each send. Share one ObjectMapper across the project:

Write schemas/order-placed-v1.json as a JSON Schema document. It is the written contract; the lab runs no validator against it, so your producer must enforce the same rules in code.

Bind it to a Java record. Use Long rather than long for the total so that a missing field decodes as null instead of a silent zero:

OrderProducer publishes three v1 representations to orders.placed: evt-7001 for ord-1042 (129900 USD minor units), evt-7002 for ord-1043 (49900), and evt-7003 for ord-1044 (259900). Their occurrence IDs are placement-1042, placement-1043, and placement-1044. Use data.orderId as the record key. Before each send, validate the required fields, nonnegative total, UTC occurredAt, and matching key. Running produce-orders v1 again must reuse those event IDs, occurrence IDs, and payloads.

Save the printed partitions and offsets to observations/phase1-produce.txt, then confirm the stored keys:

What you should see: three records with partitions between 0 and 2, and keys equal to data.orderId. With the same key serialization, keyed routing settings, and partition count, a second producer run sends each order to the same partition at a higher offset. Run that second v1 publication only when Phase 2 asks for it.

Copy the v1 schema to schemas/order-placed-v2.json and add one optional property to data: salesChannel, a string whose documented values are WEB, MOBILE, and UNKNOWN. Leave it out of required. Create OrderPlacedV2 with the extra String salesChannel field.

OldOrderConsumer joins group order-reporting, decodes into OrderPlacedV1, and prints OLD evt-7001 ord-1042 total=129900. NewOrderConsumer joins group order-fulfillment, decodes into OrderPlacedV2, maps a null channel to UNKNOWN, and prints NEW evt-7001 ord-1042 total=129900 channel=UNKNOWN. Both validate the required fields after decoding and throw if one is missing. Start old-consumer and new-consumer in two terminals:

Then mix producers. produce-orders v2 publishes the same three order acceptances as evt-7004 to evt-7006, with channels WEB, MOBILE, WEB. Preserve each order’s occurrenceId, time, amount, and other fields. These are new representations of existing facts, not new order placements. Follow this with a second produce-orders v1 run:

Save each consumer's output to its observations/phase2-*.txt file.

What you should see: both consumers process all nine records and neither stops. The old consumer prints the v2 records without a channel because Jackson discarded the unknown property; that is the forward direction, old reader with new data. The new consumer prints channel=UNKNOWN for every v1 record; that is the backward direction, and the UNKNOWN mapping is your reader-side default. These specific old and new JSON readers both handle the sample records. Do not infer a registry FULL result: compare the complete JSON Schemas under the registry’s rules. FAIL_ON_UNKNOWN_PROPERTIES=false is the JSON reader's way of ignoring an added writer field; a JSON Schema with additionalProperties: false would have rejected the same payload.

Copy the v2 schema to schemas/order-placed-v3.json, rename totalMinorUnits to total in both properties and required, and create OrderPlacedV3. Preserve occurrenceId and the other envelope fields. produce-orders v3 uses evt-7007 to evt-7009 for the three orders. Keep both consumers running, then deliberately publish these incompatible representations to the old topic:

What you should see: the old consumer decodes the first v3 record without a parse error, because total is only an unknown property, but its validation finds totalMinorUnits equal to null and throws. The poll loop's error path prints the message and the consumer exits without committing. Record the exception text, including the orders.placed-<partition> offset <n> location and the eventId, in observations/phase3-failure.txt. Restart the old consumer: it rereads uncommitted records and fails again on an incompatible record, because rereading the same bytes with the same reader cannot repair a contract mismatch. Confirm the stuck position:

The new consumer fails the same way. With a primitive long, both would have printed total=0 and kept going, turning a visible failure into wrong data.

The incompatible v3 representation needs its own topic. First, preserve the three contract violations already in orders.placed. On validation failure, each consumer writes the topic, partition, offset, and raw value to its own file: observations/phase3-quarantine-order-reporting.txt or observations/phase3-quarantine-order-fulfillment.txt. Write at most one entry per topic/partition/offset in each file, print QUARANTINED, and continue. Commit batch progress only after every record in the batch has either completed normally or reached its quarantine file. Restart both consumers and verify that each file contains the same three failed record coordinates.

Make produce-orders dual publish each order in both representations: v2 to orders.placed using evt-7004 through evt-7006, and v3 to orders.placed.v3 using evt-7007 through evt-7009. Reuse the exact payload for each representation, including its occurrence ID. Extend new-consumer with a v3 argument that uses group order-fulfillment-v3, subscribes to orders.placed.v3, and decodes OrderPlacedV3. Print the event ID and occurrence ID in both consumers so you can compare the representations:

Run new-consumer v3 in a third terminal, then the dual producer:

What you should see: during the dual run, the old consumer prints three additional OLD lines from orders.placed, and the v3 consumer prints three lines from orders.placed.v3. Matching orders share occurrence IDs but have different event IDs across versions. The two topics have independent offsets; use occurrence IDs to match facts.

send-command writes a CapturePayment command to payments.commands. For cmd-9001, use key and paymentId pay-501, correlationId checkout-1042, and a data object containing orderId ord-1042, amountMinorUnits 129900, and currency USD. For cmd-9002, use pay-502 and the corresponding order data. PaymentService joins payment-execution and checks a file of handled command IDs before executing each command. A matching ID prints DUPLICATE <commandId> ignored. Otherwise, it prints CAPTURED to simulate capture, appends the command ID to observations/handled-commands.txt, publishes PaymentCaptured, waits for publication to succeed, and then commits progress. The event uses a new event ID, preserves the key and correlation ID, and sets causationId to the command ID. This deliberately incomplete implementation exposes the failure windows in Q5; it does not perform a real payment.

Start payment-service, then send cmd-9001 twice and cmd-9002 for pay-502 once:

Stop the service with Ctrl-C, start it again, and send cmd-9001 a third time. Read payments.events with the console consumer from the beginning and save everything to observations/phase4-payments.txt.

What you should see: payments.commands holds four records, but payments.events holds exactly two PaymentCaptured events, one with causationId cmd-9001 and one with cmd-9002, each carrying its correlationId. The service prints DUPLICATE twice, once before and once after the restart, because the handled set lives in a file rather than in memory. Explain in the report what happens if the process crashes between appending to the file and the publish succeeding.

stub/prod-501.json stands in for the catalog API. product-producer notify <rev> <price> writes that revision and price into the stub file, then publishes a thin ProductUpdated event to catalog.product-events with key prod-501 and only productId and productRevision in data. product-producer state <rev> <price> writes the same file but publishes the full snapshot, including priceMinorUnits and currency.

ProductConsumer joins group search-updater. If an event carries no price, it reads stub/prod-501.json and counts one lookup; otherwise, it uses the carried values and counts no lookup. It prints the event revision, any fetched revision, and the lookup count. For example: NOTIFIED rev=17 FETCHED rev=18 lookups=1 or CARRIED rev=19 price=64900 lookups=0.

Leave product-consumer stopped. Publish both notifications below and wait for both producer commands to complete. The stub must contain revision 18 before the consumer starts:

Now start product-consumer and wait until it prints and commits both notification records. Both lookups read revision 18, so the revision 17 event demonstrates a stale historical lookup without relying on a sleep. Stop the consumer, publish state 19 64900 and state 20 69900, wait for both commands to complete, and restart it. Save all four output lines to observations/phase5-lookups.txt. The snapshot records carry their own values regardless of the stub’s current revision.

What you should see: with revision 18 already in the stub file before the consumer performs its lookup, the revision 17 notification prints NOTIFIED rev=17 FETCHED rev=18. The lookup returned a different revision than the one notified, even though every component worked correctly. Both snapshot events print CARRIED with their own revision and lookups=0.

keyed-orders order publishes three lifecycle events to orders.events keyed by orderId: OrderPlaced for ord-1042, OrderPlaced for ord-1043, and OrderCancelled for ord-1042, all for customer cust-208. keyed-orders customer publishes the same three events to orders.by-customer keyed by customerId. Do not switch the key in place on orders.events; a key change is a migration, and mixing key encodings in one topic would split one order's history across partitions.

Record the partition and offset of each send in observations/phase6-keys.txt, then verify with the console consumer on each topic using print.partition=true and print.key=true.

What you should see: in orders.by-customer, all three records sit in one partition at consecutive offsets, so a consumer sees placement of ord-1042, placement of ord-1043, and cancellation of ord-1042 in that sequence. In orders.events, the two ord-1042 records share a partition in append order, but ord-1043 lands wherever its own key hashes. If it happens to share the partition, that is coincidence rather than a guarantee.

Stop every consumer and producer with Ctrl-C, then remove the broker and its data:

Verify with docker ps -a, which should list no kafka-local container. Keep the project directory, observations, and report; delete observations/handled-commands.txt if you intend to rerun Phase 4 from scratch.

The lab separates runtime decoding, business validation, and registry schema compatibility. The optional addition works with these two readers and sample records; a registry comparison remains a separate check. Renaming the total field exposes the old reader’s missing required value. The migration preserves those failures in per-consumer quarantine files, moves v3 to its own topic, and uses occurrence IDs to connect different representations of the same accepted order.

The command experiment uses separate command and event identities and remembers handled IDs across a normal restart. It also leaves crash windows between simulated capture, the file write, and publication. The catalog experiment shows why fetching current state cannot reconstruct an older revision, while carried snapshots preserve each revision’s price. Finally, switching from order keys to customer keys changes which events share a partition and an ordering boundary.