Serializers, Deserializers and Built-in Serdes

## The problem serializers solve A Kafka broker is intentionally dumb about your data: it stores each message as an opaque sequence of bytes (a `byte[]`) and never interprets it. But your application code works with typed objects — a `String` username, a `Long` id, an `Order` POJO. Something has to translate between the two worlds. On the way out (producer), that translator is a **Serializer**; on the way in (consumer), it's a **Deserializer**. ## The Serializer interface It lives in `org.apache.kafka.common.serialization.Serializer<T>`. The method that matters is: ``` byte[] serialize(String topic, T data) ``` Given the topic name and a typed object, it returns the raw bytes. (There is also an overload that receives the record `Headers`, plus optional `configure(...)` and `close()` lifecycle methods.) The broker stores exactly those bytes. ## How the producer chooses one — key.serializer / value.serializer A `KafkaProducer` is configured with two **mandatory** properties: - `key.serializer` — class used to serialize the record key - `value.serializer` — class used to serialize the record value Each is the **fully-qualified class name** of a `Serializer` implementation, e.g. `org.apache.kafka.common.serialization.StringSerializer`. The producer loads the class by reflection at construction time. These are **separate** settings because a record's key and value are independent and frequently have different types — a `String` key with a JSON value is common. ## Why typed generics matter `KafkaProducer<K, V>` is generic. If you declare `KafkaProducer<String, byte[]>` but configure a `LongSerializer`, the mismatch is only caught at runtime when `serialize()` casts and fails, throwing a `SerializationException`. The generic type and the configured serializer class must agree. ## What happens at send time For each `ProducerRecord`, the producer calls the key serializer on the key and the value serializer on the value, then appends the resulting byte arrays to a batch destined for a partition. Serialization happens **before** partitioning (the default partitioner can hash the serialized key bytes). ## Edge cases - Missing `key.serializer`/`value.serializer` → `ConfigException` when the producer is created. - A `null` key or value is passed through as `null` bytes, not run through the serializer for the actual content in most built-in serializers (they return `null` for `null` input). - Serialization is synchronous and on the calling thread; an expensive serializer adds latency to `send()`.