docs(codec): replace examples with Schema Registry codec

Author: ce1ebrimbor

docs/batch_codec_testing_gap.md

@@ -0,0 +1,81 @@
+# Batch Codec Testing Gap
+
+## Problem
+
+`BatchCodecTestcase` tests the codec in isolation (protocol conformance, `isinstance` dispatch, fallback). It cannot test the full publish→subscribe round trip like `CodecTestcase` does for single messages.
+
+The FakeProducer's `publish_batch` path fails when a custom codec is set because the subscriber pipeline calls `decode_message()` on a `StreamMessage` with `body=list[bytes]`, and `decode_message()` expects `bytes`, not a list.
+
+## Root Cause
+
+Three things are wrong in the batch path:
+
+### 1. `_get_parser_and_decoder()` is not batch-aware
+
+When `codec=` is set, the subscriber wiring always uses `codec.decode` as the decoder — even for batch subscribers. Batch subscribers need `codec.decode_batch` (if the codec implements `BatchCodecProto`).
+
+```python
+# Current (usecase.py _get_parser_and_decoder):
+if codec:
+    async_decoder = codec.decode  # always single-message decoder
+
+# Needed:
+if codec and is_batch_subscriber and isinstance(codec, BatchCodecProto):
+    async_decoder = codec.decode_batch
+elif codec:
+    async_decoder = codec.decode
+```
+
+This requires `_get_parser_and_decoder` to know whether the current subscriber is a batch subscriber. Currently it doesn't — it only sees parser/decoder/codec options, not the subscriber type.
+
+### 2. FakeProducer `publish_batch` builds StreamMessages eagerly
+
+```python
+# Current (kafka/testing.py FakeProducer.publish_batch):
+messages = [await build_message(m, ...) for m in cmd.batch_bodies]
+await self._execute_handler(list(messages), topic, handler)
+```
+
+It builds individual `ConsumerRecord`s via `build_message()`, then passes the list to the handler. The batch parser (`AioKafkaBatchParser.parse_batch`) expects raw `ConsumerRecord`s, not pre-parsed messages. The FakeProducer should pass raw records and let the batch parser handle the batching.
+
+### 3. `call_item.is_suitable()` always sets single decoder
+
+```python
+# Current (call_item.py):
+message.set_decoder(decoder)  # always the single-message decoder
+```
+
+For batch subscribers, this should set the batch decoder when available.
+
+## Fix (3 touch points)
+
+### Touch 1: `_get_parser_and_decoder()` batch detection
+
+Add batch subscriber detection. Options:
+- Pass a `is_batch: bool` flag from `_build_fastdepends_model`
+- Check if the parser is a batch parser type
+- Add a flag to `_CallOptions`
+
+Then dispatch:
+```python
+if codec and is_batch and isinstance(codec, BatchCodecProto):
+    async_decoder = codec.decode_batch
+elif codec:
+    async_decoder = codec.decode
+```
+
+### Touch 2: FakeProducer batch path
+
+Pass raw `ConsumerRecord` tuples to the subscriber pipeline instead of pre-built `StreamMessage`s. The batch parser will handle batching + decoding correctly.
+
+### Touch 3: `call_item.is_suitable()` or `set_decoder`
+
+No change needed if Touch 1 correctly sets the batch decoder — `set_decoder` already stores whatever callable it receives.
+
+## Effort
+
+Small-medium. Three files, ~20 lines of logic. The main complexity is plumbing `is_batch` through to `_get_parser_and_decoder` without polluting the interface.
+
+## When
+
+Separate PR after the current codec encode wiring merges. Not blocking — `BatchCodecTestcase` unit tests verify the protocol dispatch correctly, and the full round-trip for single messages is already covered by `CodecTestcase`.

docs/docs/en/getting-started/serialization/codec.md

@@ -26,45 +26,25 @@ class CodecProto(Protocol):
     ) -> tuple[bytes, str | None]: ...
 ```
 
-- **`decode`** — receives a `StreamMessage` with raw bytes in `msg.body` and returns the decoded Python value. You can mutate `msg.body` before delegating to `decode_message`.
-- **`encode`** — receives a `PublishCommand` with the outgoing message body and metadata, and an optional serializer. Returns a `(bytes, content_type)` tuple.
-- **`cmd.destination`** — the target topic, subject, or queue name. Useful for codecs that need destination-specific behavior (e.g. Schema Registry topic-to-schema resolution).
+- **`decode`** — receives a `StreamMessage` with raw bytes in `msg.body` and returns the decoded Python value.
+- **`encode`** — receives a `PublishCommand` containing the message body, destination, and headers. Returns a `(bytes, content_type)` tuple. Access the payload via `cmd.body` and the target topic/subject/queue via `cmd.destination`.
 
-If no codec is set, `DefaultCodec` is used automatically. It handles JSON objects, plain text, and raw bytes without any configuration.
+If no codec is set, `DefaultCodec` is used automatically. It handles JSON objects, plain text, and raw bytes.
 
-## Compression Example
+## Example: Schema Registry
 
-A Gzip codec that compresses outgoing messages and decompresses incoming ones:
+A Confluent Avro codec that encodes and decodes messages using the [Confluent wire format](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format){target="_blank"} (magic byte + schema ID + Avro payload). Requires `fastavro` and `confluent-kafka`:
 
-=== "AIOKafka"
-    ```python linenums="1" hl_lines="15-27 30"
-    {!> docs_src/getting_started/serialization/codec_gzip_kafka.py !}
-    ```
-
-=== "Confluent"
-    ```python linenums="1" hl_lines="15-27 30"
-    {!> docs_src/getting_started/serialization/codec_gzip_confluent.py !}
-    ```
-
-=== "RabbitMQ"
-    ```python linenums="1" hl_lines="15-27 30"
-    {!> docs_src/getting_started/serialization/codec_gzip_rabbit.py !}
-    ```
-
-=== "NATS"
-    ```python linenums="1" hl_lines="15-27 30"
-    {!> docs_src/getting_started/serialization/codec_gzip_nats.py !}
-    ```
+```bash
+pip install fastavro confluent-kafka
+```
 
-=== "Redis"
-    ```python linenums="1" hl_lines="15-27 30"
-    {!> docs_src/getting_started/serialization/codec_gzip_redis.py !}
-    ```
+```python linenums="1" hl_lines="22-66 68-76"
+{!> docs_src/getting_started/serialization/codec_schema_registry_kafka.py !}
+```
 
-=== "MQTT"
-    ```python linenums="1" hl_lines="15-27 30"
-    {!> docs_src/getting_started/serialization/codec_gzip_mqtt.py !}
-    ```
+!!! note
+    The codec fetches and caches schemas from the registry at startup and on first encounter. The `subject` follows Confluent's naming convention: `{topic}-value`.
 
 ## Priority