How do you send Java objects (POJOs) as JSON with KafkaTemplate, and what should you know about Spring's JsonSerializer?

**The problem.** Kafka moves `byte[]`. To send a domain object like `OrderCreated` you need a *serializer* that turns it into bytes, and the consumer needs a matching *deserializer*. For JSON, Spring for Apache Kafka provides `org.springframework.kafka.support.serializer.JsonSerializer<T>` (and `JsonDeserializer<T>`), backed by **Jackson**. **Wiring the producer.** ```yaml spring: kafka: producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer ``` Now declare `KafkaTemplate<String, OrderCreated>` and call `send("orders", order.getId(), order)`. JsonSerializer calls its ObjectMapper's `writeValueAsBytes(order)`. **Type headers (`__TypeId__`).** By default JsonSerializer sets `addTypeInfo=true`, writing the fully-qualified class name into a Kafka **header** named `__TypeId__`. The paired `JsonDeserializer` reads that header to know which class to instantiate — convenient for Java-to-Java flows because one deserializer can handle multiple event types. Turn it off with `JsonSerializer.ADD_TYPE_INFO_HEADERS=false` (or `serializer.setAddTypeInfo(false)`) when the consumer is a non-JVM app, or when you don't want to leak/couple concrete class names. Then the consumer must be told the target type another way (default type, or type-mapping headers). **Customizing the ObjectMapper.** The default mapper may not handle Java 8 dates or unknown fields the way you want. Construct the serializer with your own mapper: ```java ObjectMapper mapper = JsonMapper.builder() .addModule(new JavaTimeModule()) .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false) .build(); ProducerFactory<String, OrderCreated> pf = new DefaultKafkaProducerFactory<>( configProps, new StringSerializer(), new JsonSerializer<>(mapper)); ``` Passing serializer instances to `DefaultKafkaProducerFactory` (instead of just class names in the config map) is the clean way to inject a configured mapper. **Consumer-side safety — trusted packages.** `JsonDeserializer` using `__TypeId__` will try to instantiate whatever class name the header says. Deserializing arbitrary types from untrusted input is a security risk. Restrict it with `spring.kafka.consumer.properties.spring.json.trusted.packages=com.myco.events` (or `*` only when you trust producers). This is a real interview gotcha. **Type mismatches.** If the producer writes class `a.b.OrderCreated` in the header but the consumer app has it under a different package, deserialization fails unless you configure **type mappings** (`spring.json.type.mapping`) that alias logical names to local classes — decoupling wire identity from Java package structure. This is the robust pattern for evolving schemas across services. **Where errors surface.** Serialization happens on the *calling* thread inside `send()`, before buffering. A non-serializable object or Jackson failure throws `SerializationException` **synchronously** from `send()`, not via the future — so wrap send in try/catch if inputs can be bad. **Alternatives.** JsonSerializer is easy and human-readable but verbose and schema-less. For strong schema/evolution guarantees teams often use Avro or Protobuf with a Schema Registry (`KafkaAvroSerializer`). Those are out of scope here but worth naming. **When to use.** JSON via JsonSerializer is a great default for internal Java-to-Java event flows and quick starts; reach for Avro/Protobuf + registry when you need enforced schema evolution or cross-language contracts at scale.