Kafka/Produce API extension

course-definition.yml

@@ -71,6 +71,15 @@ extensions:
 
       [fetch-api]: https://kafka.apache.org/protocol.html#The_Messages_Fetch
 
+  - slug: "producing-messages"
+    name: "Producing Messages"
+    description_markdown: |
+      In this challenge extension you'll add support for producing messages by implementing the [Produce][produce-api] API.
+
+      Along the way you'll learn about how Kafka's Produce API works, how Kafka stores messages on disk and more.
+
+      [produce-api]: https://kafka.apache.org/protocol.html#The_Messages_Produce
+
 stages:
   - slug: "vi6"
     name: "Bind to a port"
@@ -199,3 +208,54 @@ stages:
     difficulty: hard
     marketing_md: |-
       In this stage, you'll implement the Fetch response for a topic with multiple messages, reading them from disk.
+
+  # Producing Messages
+
+  - slug: "um3"
+    primary_extension_slug: "producing-messages"
+    name: "Include Produce in APIVersions"
+    difficulty: easy
+    marketing_md: |-
+      In this stage, you'll add the Produce API to the APIVersions response.
+
+  - slug: "ck2"
+    primary_extension_slug: "producing-messages"
+    name: "Produce to an invalid topic"
+    difficulty: medium
+    marketing_md: |-
+      In this stage, you'll implement the Produce response for an invalid topic.
+
+  - slug: "dp1"
+    primary_extension_slug: "producing-messages"
+    name: "Respond to Produce requests"
+    difficulty: medium
+    marketing_md: |-
+      In this stage, you'll implement the Produce response for a valid topic.
+
+  - slug: "ps7"
+    primary_extension_slug: "producing-messages"
+    name: "Produce a single record"
+    difficulty: hard
+    marketing_md: |-
+      In this stage, you'll implement producing a single record to disk.
+
+  - slug: "sb8"
+    primary_extension_slug: "producing-messages"
+    name: "Produce multiple records"
+    difficulty: hard
+    marketing_md: |-
+      In this stage, you'll implement producing multiple records.
+
+  - slug: "mf2"
+    primary_extension_slug: "producing-messages"
+    name: "Produce to multiple partitions"
+    difficulty: hard
+    marketing_md: |-
+      In this stage, you'll implement producing to multiple partitions.
+
+  - slug: "ar4"
+    primary_extension_slug: "producing-messages"
+    name: "Produce to multiple topics"
+    difficulty: hard
+    marketing_md: |-
+      In this stage, you'll implement producing to multiple topics.

stage_descriptions/producing-messages-01-um3.md

@@ -0,0 +1,36 @@
+In this stage, you'll add an entry for the `Produce` API to the APIVersions response.
+
+## The Produce API
+
+The [Produce API](https://kafka.apache.org/protocol#The_Messages_Produce) (API key `0`) is used to produce messages to a Kafka topic.
+
+We've created an interactive protocol inspector for the request & response structures for `Produce`:
+
+- 🔎 [Produce Request (v11)](https://binspec.org/kafka-produce-request-v11)
+- 🔎 [Produce Response (v11)](https://binspec.org/kafka-produce-response-v11)
+
+In this stage, you'll only need to add an entry for the `Produce` API to the APIVersions response you implemented in earlier stages. This lets clients know that your broker supports the `Produce` API. We'll get to responding to `Produce` requests in later stages.
+
+## Tests
+
+The tester will execute your program like this:
+
+```bash
+./your_program.sh /tmp/server.properties
+```
+
+It'll then connect to your server on port 9092 and send a valid `APIVersions` (v4) request.
+
+The tester will validate that:
+
+- The first 4 bytes of your response (the "message length") are valid.
+- The correlation ID in the response header matches the correlation ID in the request header.
+- The error code in the response body is `0` (NO_ERROR).
+- The response body contains at least one entry for the API key `0` (Produce).
+- The `MaxVersion` for the Produce API is at least 11.
+
+## Notes
+
+- You don't have to implement support for handling `Produce` requests in this stage. We'll get to this in later stages.
+- You'll still need to include the entry for `APIVersions` in your response to pass earlier stages.
+- The `MaxVersion` for `Produce` and `APIVersions` are different. For `APIVersions`, it is 4. For `Produce`, it is 11.

stage_descriptions/producing-messages-02-ck2.md

@@ -0,0 +1,42 @@
+In this stage, you'll add support for handling Produce requests to invalid topics or partitions.
+
+## Produce API Response for Invalid Topics or Partitions
+
+When a Kafka broker receives a Produce request, it needs to validate that both the topic and partition exist. If either the topic or partition doesn't exist, it returns an appropriate error code and response.
+
+For this stage, you can hardcode the error response - assume that all Produce requests are for invalid topics or partitions and return the error code `3` (UNKNOWN_TOPIC_OR_PARTITION). In the next stage, you'll implement handling success responses.
+
+We've created an interactive protocol inspector for the request & response structures for `Produce`:
+
+- 🔎 [Produce Request (v11)](https://binspec.org/kafka-produce-request-v11)
+- 🔎 [Produce Response (v11) - Invalid Topic](https://binspec.org/kafka-produce-error-response-v11-invalid-topic)
+
+## Tests
+
+The tester will execute your program like this:
+
+```bash
+./your_program.sh /tmp/server.properties
+```
+
+It'll then connect to your server on port 9092 and send a `Produce` (v11) request with either an invalid topic name or a valid topic but invalid partition.
+
+The tester will validate that:
+
+- The first 4 bytes of your response (the "message length") are valid.
+- The correlation ID in the response header matches the correlation ID in the request header.
+- The `error_code` in the response body is `3` (UNKNOWN_TOPIC_OR_PARTITION).
+- The `throttle_time_ms` field in the response is `0`.
+- Inside the topic response:
+  - The `name` field matches the topic name in the request.
+  - Inside the partition response:
+    - The `index` field matches the partition in the request.
+    - The `base_offset` field is `-1`.
+    - The `log_append_time_ms` field is `-1`.
+    - The `log_start_offset` field is `-1`.
+
+## Notes
+
+- You'll need to parse the `Produce` request in this stage to get the topic name and partition to send in the response.
+- For this stage, simply hardcode the error response without implementing actual validation logic.
+- The official docs for the `Produce` request can be found [here](https://kafka.apache.org/protocol.html#The_Messages_Produce). Make sure to scroll down to the "Produce Response (Version: 11)" section.

stage_descriptions/producing-messages-03-dp1.md

@@ -0,0 +1,57 @@
+In this stage, you'll add support for responding to Produce requests with a valid topic.
+
+## Produce API Response for Valid Topics
+
+When a Kafka broker receives a Produce request, it needs to validate that both the topic and partition exist. If either the topic or partition doesn't exist, it returns an appropriate error code and response.
+
+The broker performs validation in this order:
+1. **Topic validation**: Check if the topic exists by reading the `__cluster_metadata` topic's log file
+2. **Partition validation**: If the topic exists, check if the partition exists within that topic
+
+### Topic Validation
+
+To validate that a topic exists, the broker reads the `__cluster_metadata` topic's log file, located at `/tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000.log`. Inside the log file, the broker finds the topic's metadata, which is a `record` (inside a RecordBatch) with a payload of type `TOPIC_RECORD`. If there exists a `TOPIC_RECORD` with the given topic name and the topic ID, the topic exists.
+
+### Partition Validation
+
+To validate that a partition exists, the broker reads the same `__cluster_metadata` topic's log file and finds the partition's metadata, which is a `record` (inside a RecordBatch) with a payload of type `PARTITION_RECORD`. If there exists a `PARTITION_RECORD` with the given partition index, the UUID of the topic it is associated with, and the UUID of the directory it is associated with, the partition exists.
+
+We've created an interactive protocol inspector for the request & response structures for `Produce`:
+
+- 🔎 [Produce Request (v11)](https://binspec.org/kafka-produce-request-v11)
+- 🔎 [Produce Response (v11)](https://binspec.org/kafka-produce-response-v11)
+
+We've also created an interactive protocol inspector for the `__cluster_metadata` topic's log file:
+- 🔎 [__cluster_metadata log file](https://binspec.org/kafka-topic-log)
+
+In this stage, you'll need to implement the response for a `Produce` request with a valid topic. In later stages, you'll handle successfully producing messages to valid topics and partitions and persist messages to disk using Kafka's RecordBatch format.
+
+## Tests
+
+The tester will execute your program like this:
+
+```bash
+./your_program.sh /tmp/server.properties
+```
+
+It'll then connect to your server on port 9092 and send a `Produce` (v11) request with either an invalid topic name or a valid topic but invalid partition.
+
+The tester will validate that:
+
+- The first 4 bytes of your response (the "message length") are valid.
+- The correlation ID in the response header matches the correlation ID in the request header.
+- The `error_code` in the response body is `3` (UNKNOWN_TOPIC_OR_PARTITION).
+- The `throttle_time_ms` field in the response is `0`.
+- Inside the topic response:
+  - The `name` field matches the topic name in the request.
+  - Inside the partition response:
+    - The `index` field matches the partition in the request.
+    - The `base_offset` field is `-1`.
+    - The `log_append_time_ms` field is `-1`.
+    - The `log_start_offset` field is `-1`.
+
+## Notes
+
+- You'll need to parse the `Produce` request in this stage to get the topic name and partition to send in the response.
+- The official docs for the `Produce` request can be found [here](https://kafka.apache.org/protocol.html#The_Messages_Produce). Make sure to scroll down to the "Produce Response (Version: 11)" section.
+- The official Kafka docs don't cover the structure of records inside the `__cluster_metadata` topic, but you can find the definitions in the Kafka source code [here](https://github.com/apache/kafka/tree/5b3027dfcbcb62d169d4b4421260226e620459af/metadata/src/main/resources/common/metadata).

stage_descriptions/producing-messages-04-ps7.md

@@ -0,0 +1,48 @@
+In this stage, you'll add support for successfully producing a single record.
+
+## Single Record Production
+
+When a Kafka broker receives a Produce request, it needs to validate that the topic and partition exist (using the `__cluster_metadata` topic's log file), store the record in the appropriate log file using Kafka's on-disk format, and return a successful response with the assigned offset.
+
+The record must be persisted to the topic's log file at `<log-dir>/<topic-name>-<partition-index>/00000000000000000000.log` using Kafka's RecordBatch format.
+
+We've created an interactive protocol inspector for the request & response structures for `Produce`:
+
+- 🔎 [Produce Request (v11)](https://binspec.org/kafka-produce-request-v11)
+- 🔎 [Produce Response (v11)](https://binspec.org/kafka-produce-response-v11)
+
+Kafka's on-disk log format uses the same [RecordBatch](https://binspec.org/kafka-record-batches) format that is used in `Produce` and `Fetch` requests.
+
+You can refer to the following interactive protocol inspector for Kafka's log file format:
+- 🔎 [A sample topic's log file](https://binspec.org/kafka-topic-log)
+
+## Tests
+
+The tester will execute your program like this:
+
+```bash
+./your_program.sh /tmp/server.properties
+```
+
+It'll then connect to your server on port 9092 and send multiple successive `Produce` (v11) requests with single records.
+
+The tester will validate that:
+
+- The first 4 bytes of your response (the "message length") are valid.
+- The correlation ID in the response header matches the correlation ID in the request header.
+- The error code in the response body is `0` (NO_ERROR).
+- The `throttle_time_ms` field in the response is `0`.
+- Inside the topic response:
+  - The `name` field matches the topic name in the request.
+  - Inside the partition response:
+    - The `index` field matches the partition in the request.
+    - The `base_offset` field contains the assigned offset for the record. (The `base_offset` is the offset of the record in the partition, not the offset of the batch. So 0 for the first record, 1 for the second record, and so on.)
+    - The `log_append_time_ms` field is `-1` (signifying that the timestamp is the latest).
+    - The `log_start_offset` field is `0`.
+- The record is persisted to the appropriate log file on disk at `<log-dir>/<topic-name>-<partition-index>/00000000000000000000.log`.
+
+## Notes
+
+- On-disk log files must be stored in `RecordBatch` format.
+- The offset assignment should start from `0` for new partitions and increment for each subsequent record.
+- The official docs for the `Produce` request can be found [here](https://kafka.apache.org/protocol.html#The_Messages_Produce). Make sure to scroll down to the "Produce Response (Version: 11)" section.

stage_descriptions/producing-messages-05-sb8.md

@@ -0,0 +1,40 @@
+In this stage, you'll add support for producing multiple records in a single `Produce` request.
+
+## Batch Processing
+
+When a Kafka broker receives a `Produce` request containing a `RecordBatch` with multiple records, it needs to validate that the topic and partition exist, assign sequential offsets to each record within the batch, and store the entire batch atomically to the log file. The `RecordBatch` containing multiple records must be stored as a single unit to the topic's log file at `<log-dir>/<topic-name>-<partition-index>/00000000000000000000.log`.
+
+We've created an interactive protocol inspector for the request & response structures for `Produce`:
+
+- 🔎 [Produce Request (v11)](https://binspec.org/kafka-produce-request-v11)
+- 🔎 [Produce Response (v11)](https://binspec.org/kafka-produce-response-v11)
+
+## Tests
+
+The tester will execute your program like this:
+
+```bash
+./your_program.sh /tmp/server.properties
+```
+
+It'll then connect to your server on port 9092 and send a `Produce` (v11) request containing a RecordBatch with multiple records.
+
+The tester will validate that:
+
+- The first 4 bytes of your response (the "message length") are valid.
+- The correlation ID in the response header matches the correlation ID in the request header.
+- The error code in the response body is `0` (NO_ERROR).
+- The `throttle_time_ms` field in the response is `0`.
+- Inside the topic response:
+  - The `name` field matches the topic name in the request.
+  - Inside the partition response:
+    - The `index` field matches the partition in the request.
+    - The `base_offset` field is 0 (the base offset for the batch).
+    - The `log_append_time_ms` field is `-1` (signifying that the timestamp is the latest).
+    - The `log_start_offset` field is `0`.
+- The records are persisted to the appropriate log file on disk at `<log-dir>/<topic-name>-<partition-index>/00000000000000000000.log` with sequential offsets.
+
+## Notes
+
+- Records within a batch must be assigned sequential offsets (e.g., if the base offset is 5, records get offsets 5, 6, 7).
+- The official docs for the `Produce` request can be found [here](https://kafka.apache.org/protocol.html#The_Messages_Produce). Make sure to scroll down to the "Produce Response (Version: 11)" section.

stage_descriptions/producing-messages-06-mf2.md

@@ -0,0 +1,45 @@
+In this stage, you'll add support for producing to multiple partitions of the same topic in a single request.
+
+## Partition Routing
+
+When a Kafka broker receives a Produce request targeting multiple partitions of the same topic, it needs to validate that the topic and all partitions exist, write records to each partition's log file independently, and return a response containing results for all partitions. Each partition maintains its own offset sequence independently, so partition 0 and partition 1 can both have records starting at offset 0.
+
+We've created an interactive protocol inspector for the request & response structures for `Produce`:
+
+- 🔎 [Produce Request (v11)](https://binspec.org/kafka-produce-request-v11)
+- 🔎 [Produce Response (v11)](https://binspec.org/kafka-produce-response-v11)
+
+## Tests
+
+The tester will execute your program like this:
+
+```bash
+./your_program.sh /tmp/server.properties
+```
+
+It'll then connect to your server on port 9092 and send a `Produce` (v11) request targeting multiple partitions of the same topic. The request will contain multiple RecordBatches, one for each partition. Each RecordBatch will contain a single record. 
+
+The tester will validate that:
+
+- The first 4 bytes of your response (the "message length") are valid.
+- The correlation ID in the response header matches the correlation ID in the request header.
+- The error code in the response body is `0` (NO_ERROR).
+- The `throttle_time_ms` field in the response is `0`.
+- There is a single topic present in the response.
+- Inside the topic response:
+  - The `name` field matches the topic name in the request.
+  - Each partition in the request has a corresponding partition response.
+  - Inside each partition response:
+    - The `index` field matches the partition in the request.
+    - The `base_offset` field contains the assigned offset for that partition.
+    - The error code is `0` (NO_ERROR).
+    - The `log_append_time_ms` field is `-1` (signifying that the timestamp is the latest).
+    - The `log_start_offset` field is `0`.
+- Records are persisted to the correct partition log files on disk.
+- Offset assignment is independent per partition (partition 0 and partition 1 can both have offset 0). 
+
+## Notes
+
+- The response must include entries for all requested partitions.
+- On-disk log files must be stored in `RecordBatch` format with proper CRC validation.
+- The official docs for the `Produce` request can be found [here](https://kafka.apache.org/protocol.html#The_Messages_Produce). Make sure to scroll down to the "Produce Response (Version: 11)" section.