<a id="streams-developer-guide-serdes"></a>

# Kafka Streams Data Types and Serialization for Confluent Platform

A *Serde* is a combined serializer and deserializer that Kafka Streams uses to
convert record keys and values to and from bytes.

Every Kafka Streams application must provide Serdes (Serializer/Deserializer) for
the data types of record keys and record values (for example, `java.lang.String` or
Avro objects) to materialize the data when necessary. Operations that require
such Serdes information include: `stream()`, `table()`, `to()`, `repartition()`,
`groupByKey()`, `groupBy()`.

You can provide Serdes by using either of these methods, but you must use at least one of these methods:

- By setting default Serdes via a `Properties` instance.
- By specifying explicit Serdes when calling the appropriate API methods, which overrides the defaults.

You can configure Java streams applications to deserialize and ingest data in multiple ways, including Kafka console producers, JDBC source connectors, and Java client producers.
For full code examples, see [Pipelining with Kafka Connect and Kafka Streams in Confluent Platform](../connect-streams-pipeline.md#demo-connect-streams-pipeline).

## Configuring Serdes

Serdes specified in the Streams configuration via the `Properties` config are
used as the default in your Kafka Streams application. Because this config’s default
is null, you must either set a default Serde by using this configuration or pass
in Serdes explicitly, as described below.

```java
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;

Properties settings = new Properties();
// Default serde for keys of data records (here: built-in serde for String type)
settings.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
// Default serde for values of data records (here: built-in serde for Long type)
settings.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass().getName());
```

If a `Serde` is specified via `Properties`, the `Serde` class can’t have
generic types, which means that you can’t use a class like
`MySerde<T extends Number> implements Serde<T>`. This implies that you can’t
use any `Serde` that is created via `Serdes.serdeFrom(Serializer<T>, Deserializer<T>)`.
Only fully typed `Serde` classes like `MySerde implements Serde<MyCustomType>` are
supported, due to Java type erasure.

## Overriding default Serdes

You can also specify Serdes explicitly by passing them to the appropriate API methods, which overrides the default serde settings:

```java
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;

final Serde<String> stringSerde = Serdes.String();
final Serde<Long> longSerde = Serdes.Long();

// The stream userCountByRegion has type `String` for record keys (for region)
// and type `Long` for record values (for user counts).
KStream<String, Long> userCountByRegion = ...;
userCountByRegion.to("RegionCountsTopic", Produced.with(stringSerde, longSerde));
```

If you want to override serdes selectively, that is, keep the defaults for some fields, then don’t specify the serde whenever you want to leverage the default settings:

```java
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;

// Use the default serializer for record keys (here: region as String) by not specifying the key serde,
// but override the default serializer for record values (here: userCount as Long).
final Serde<Long> longSerde = Serdes.Long();
KStream<String, Long> userCountByRegion = ...;
userCountByRegion.to("RegionCountsTopic", Produced.valueSerde(Serdes.Long()));
```

If some of your incoming records are corrupted or ill-formatted, they
cause the deserializer class to report an error. An interface named
`org.apache.kafka.streams.errors.DeserializationExceptionHandler`
enables you to customize how to handle such records. Specify the customized
implementation of the interface by using `StreamsConfig`. For more
information, see
[Failure and exception handling FAQ](../faq.md#streams-faq-failure-handling-deserialization-errors).

<a id="streams-developer-guide-serdes-available"></a>

## Available Serdes

### Primitive and basic types

Apache Kafka® includes several built-in serde implementations for Java primitives and basic types such as `byte[]` in
its `kafka-clients` Maven artifact:

```xml
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>8.3.1-ccs</version>
</dependency>
```

This artifact provides the following serde implementations under the package [org.apache.kafka.common.serialization](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/serialization), which you can leverage, for example, when defining default serializers in your Streams configuration.

| Data type   | Serde                                                  |
|-------------|--------------------------------------------------------|
| Boolean     | `Serdes.Boolean()`                                     |
| byte[]      | `Serdes.ByteArray()`, `Serdes.Bytes()` (see tip below) |
| ByteBuffer  | `Serdes.ByteBuffer()`                                  |
| Double      | `Serdes.Double()`                                      |
| Float       | `Serdes.Float()`                                       |
| Integer     | `Serdes.Integer()`                                     |
| List        | `Serdes.ListSerde()`                                   |
| Long        | `Serdes.Long()`                                        |
| Short       | `Serdes.Short()`                                       |
| String      | `Serdes.String()`                                      |
| UUID        | `Serdes.UUID()`                                        |
| Void        | `Serdes.Void()`                                        |

You would use the built-in Serdes as follows, using the example of the String serde:

```java
// When configuring the default Serdes of StreamConfig
Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,   Serdes.String().getClass().getName());
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

// When you want to override Serdes explicitly/selectively
final Serde<String> stringSerde = Serdes.String();
StreamsBuilder builder = new StreamsBuilder();
builder.stream("my-avro-topic", Consumed.with(keyGenericAvroSerde, valueGenericAvroSerde));
```

<a id="streams-data-avro"></a>

### Avro

Confluent provides Schema Registry-compatible
[Avro serdes](https://mvnrepository.com/artifact/io.confluent/kafka-streams-avro-serde?repo=confluent-packages)
for data in generic Avro and in specific Avro format:

```xml
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-streams-avro-serde</artifactId>
    <version>8.3.1</version>
</dependency>
```

Both the generic and the specific Avro serde require you to configure the endpoint of
[Confluent Schema Registry](/platform/current/schema-registry/index.html) via the `schema.registry.url` setting:

* When you define the generic or specific Avro serde as a default serde via `StreamsConfig`, then you must also set
  the Schema Registry endpoint in `StreamsConfig`.
* When you instantiate the generic or specific Avro serde directly (e.g., `new GenericAvroSerde()`), you must
  call `Serde#configure()` on the serde instance to set the Schema Registry endpoint before using the serde instance.
  Additionally, you must tell `Serde#configure()` via a boolean parameter whether the serde instance is used for
  serializing/deserializing record *keys* (`true`) or record *values* (`false`).

#### NOTE
Use `GenericAvroSerde` to enable both forward and backward schema
compatibility if your application requires both forward schema checks of the
producer and backward compatibility for Kafka Streams.

Usage example for Confluent `GenericAvroSerde`:

```java
// Generic Avro serde example
import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;

// When configuring the default serdes of StreamConfig
final Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
streamsConfiguration.put("schema.registry.url", "http://my-schema-registry:8081");

// When you want to override serdes explicitly/selectively
final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url",
                                                                 "http://my-schema-registry:8081");
// `Foo` and `Bar` are Java classes generated from Avro schemas
final Serde<Foo> keyGenericAvroSerde = new GenericAvroSerde();
keyGenericAvroSerde.configure(serdeConfig, true); // `true` for record keys
final Serde<Bar> valueGenericAvroSerde = new GenericAvroSerde();
valueGenericAvroSerde.configure(serdeConfig, false); // `false` for record values

StreamsBuilder builder = new StreamsBuilder();
KStream<Foo, Bar> textLines = builder.stream("my-avro-topic", Consumed.with(keyGenericAvroSerde, valueGenericAvroSerde));
```

Usage example for Confluent `SpecificAvroSerde`:

```java
// Specific Avro serde example
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;

// When configuring the default serdes of StreamConfig
final Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
streamsConfiguration.put("schema.registry.url", "http://my-schema-registry:8081");

// When you want to override serdes explicitly/selectively
final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url",
                                                                 "http://my-schema-registry:8081");
// `Foo` and `Bar` are Java classes generated from Avro schemas
final Serde<Foo> keySpecificAvroSerde = new SpecificAvroSerde<>();
keySpecificAvroSerde.configure(serdeConfig, true); // `true` for record keys
final Serde<Bar> valueSpecificAvroSerde = new SpecificAvroSerde<>();
valueSpecificAvroSerde.configure(serdeConfig, false); // `false` for record values

StreamsBuilder builder = new StreamsBuilder();
KStream<Foo, Bar> textLines = builder.stream("my-avro-topic", Consumed.with(keySpecificAvroSerde, valueSpecificAvroSerde));
```

When you create source streams, you specify input serdes by using the Streams DSL.
When you construct the processor topology by using the lower-level [Processor API](processor-api.md#streams-developer-guide-processor-api),
you can specify the serde class, like the Confluent `GenericAvroSerde`
and `SpecificAvroSerde` classes.

```java
Topology builder = new Topology();
builder.addSource("Source", keyGenericAvroSerde.deserializer(), valueGenericAvroSerde.deserializer(), inputTopic);
```

<a id="streams-data-avro-primitive"></a>

### Avro primitive

Starting with version 5.5.0, Confluent Platform provides a serializer and deserializer
for writing and reading data in “Avro primitive” format.

The [Avro primitive types](https://avro.apache.org/docs/current/spec.html#schema_primitive)
are `null`, `boolean`, `int`, `long`, `float`, `double`, `bytes`,
and `string`. Other types aren’t supported by this serde.

The primary use case for `PrimitiveAvroSerde` is for keys.

This serde’s specific Avro counterpart is `SpecificAvroSerde`, and its
generic Avro counterpart is `GenericAvroSerde`.

This serde reads and writes data according to the Schema Registry [wire format](../../schema-registry/fundamentals/serdes-develop/index.md#messages-wire-format).
It requires access to a Schema Registry endpoint, which you must define in
`PrimitiveAvroSerde#configure(Map, boolean)` by using the
`schema.registry.url` parameter.

Usage example for Confluent `PrimitiveAvroSerde`:

```java
Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, PrimitiveAvroSerde.class);
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, PrimitiveAvroSerde.class);
streamsConfiguration.put(
    AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
    "http://confluent-schema-registry-server:8081/");
```

The following example shows how to explicitly override the application’s
default serdes so that only specific operations, like `KStream#to` use
this serde.

```java
Serde<Long> longAvroSerde = new PrimitiveAvroSerde<Long>();
boolean isKeySerde = true;
longAvroSerde.configure(
    Collections.singletonMap(
        AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
        "http://confluent-schema-registry-server:8081/"),
    isKeySerde);

Serde<GenericRecord> genericAvroSerde = new GenericAvroSerde();
isKeySerde = false;
genericAvroSerde.configure(
    Collections.singletonMap(
        AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
        "http://confluent-schema-registry-server:8081/"),
    isKeySerde);

KStream<Long, GenericRecord> stream = builder.stream("my-input-topic",
    Consumed.with(longAvroSerde, genericAvroSerde));
```

<a id="streams-data-reflection-avro"></a>

### Reflection Avro

Starting with version 5.4.0, Confluent Platform also provides a serializer and deserializer
for writing and reading data in “reflection Avro” format.
This serde’s “generic Avro” counterpart is `GenericAvroSerde`. This serde
reads and writes data according to the wire format defined at
[Formats, Serializers, and Deserializers for Schema Registry](../../schema-registry/fundamentals/serdes-develop/index.md#serializer-and-formatter).
It requires access to a Schema Registry endpoint, which you must define in the
`GenericAvroDeserializer` by using the `schema.registry.url` parameter.

The following code example configures this serde as a Kafka Streams application’s
default serde for both record keys and record values:

```java
Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, ReflectionAvroSerde.class);
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, ReflectionAvroSerde.class);
streamsConfiguration.put(
       AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
       "http://confluent-schema-registry-server:8081/");
```

The following code example explicitly overrides the application’s default serdes,
regardless of how they were configured, so that only specific operations,
like `KStream#to` use this serde:

```java
Serde<MyJavaClassGeneratedFromAvroSchema> reflectionAvroSerde = new ReflectionAvroSerde<>();
   boolean isKeySerde = false;
   reflectionAvroSerde.configure(
       Collections.singletonMap(
           AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
           "http://confluent-schema-registry-server:8081/"),
       isKeySerde);
KStream<String, MyJavaClassGeneratedFromAvroSchema> stream = ...;
stream.to(Serdes.String(), reflectionAvroSerde, "my-output-topic");
```

### JSON example serde

The Kafka Streams code examples also include a basic serde implementation for JSON Schema:

* [PageViewTypedDemo](https://github.com/apache/kafka/blob/trunk/streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/PageViewTypedDemo.java#L83)

As shown in the example file, you can use JSONSerdes inner classes
`Serdes.serdeFrom(<serializerInstance>, <deserializerInstance>)` to construct
JSON compatible serializers and deserializers.

### JSON Schema

Confluent provides a Schema Registry-compatible
[JSON Schema Serde](https://mvnrepository.com/artifact/io.confluent/kafka-streams-json-schema-serde)
for data in JSON format:

```xml
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-streams-json-schema-serde</artifactId>
    <version>8.3.1</version>
</dependency>
```

Similar to how the Avro deserializer can return an instance of a specific Avro
record type or a `GenericRecord`, the JSON Schema deserializer can return an
instance of a specific Java class, or an instance of `JsonNode`. For more
information, see [JSON Schema Serializer and Deserializer for Schema Registry on Confluent Platform](../../schema-registry/fundamentals/serdes-develop/serdes-json.md#serdes-and-formatter-json).

The following code example shows how to use the `KafkaJsonSchemaSerde` class
to serialize and deserialize a JSON record with a schema. For a related code
listing, see
[SerializationTutorial.java](https://github.com/confluentinc/kafka-tutorials/blob/master/_includes/tutorials/serialization/kstreams/code/src/main/java/io/confluent/developer/serialization/SerializationTutorial.java#L82).

```java
private static KafkaJsonSchemaSerde<MovieProtos.Movie> movieProtobufSerde(Properties envProps) {
    final KafkaJsonSchemaSerde<MovieProtos.Movie> jsonSchemaSerde = new KafkaJsonSchemaSerde<>();
    Map<String, Object> serdeConfig = new HashMap<>();
    serdeConfig.put(SCHEMA_REGISTRY_URL_CONFIG, envProps.getProperty("schema.registry.url"));
    jsonSchemaSerde.configure(serdeConfig, false);
    return jsonSchemaSerde;
}
```

### Protobuf

Confluent provides a Schema Registry-compatible
[Protobuf Serde](https://mvnrepository.com/artifact/io.confluent/kafka-streams-protobuf-serde)
for data in Protobuf format:

```xml
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-streams-protobuf-serde</artifactId>
  <version>8.3.1</version>
</dependency>
```

The Protobuf serde provides support for *referenced schemas*, the ability
of a schema to refer to other schemas. Also, a Protobuf schema can register
a referenced schema automatically. For more information, see
[Protobuf Schema Serializer and Deserializer for Schema Registry on Confluent Platform](../../schema-registry/fundamentals/serdes-develop/serdes-protobuf.md#serdes-and-formatter-protobuf).

The following code example shows how to use the `KafkaProtobufSerde` class
to serialize and deserialize a Protobuf record with a schema. For the full code
listing, see
[SerializationTutorial.java](https://github.com/confluentinc/kafka-tutorials/blob/master/_includes/tutorials/serialization/kstreams/code/src/main/java/io/confluent/developer/serialization/SerializationTutorial.java#L82).

```java
protected KafkaProtobufSerde<MovieProtos.Movie> movieProtobufSerde(Properties envProps) {
    final KafkaProtobufSerde<MovieProtos.Movie> protobufSerde = new KafkaProtobufSerde<>();
    Map<String, String> serdeConfig = new HashMap<>();
    serdeConfig.put(SCHEMA_REGISTRY_URL_CONFIG, envProps.getProperty("schema.registry.url"));
    protobufSerde.configure(serdeConfig, false);
    return protobufSerde;
}
```

<a id="streams-developer-guide-window-serdes"></a>

### Window Serdes

Kafka Streams includes serde implementations for windowed types in its
`kafka-streams` Maven artifact:

```xml
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>4.3.0</version>
</dependency>
```

This artifact provides the following windowed serde implementations under the
package org.apache.kafka.streams.kstream:

Serdes
: - `WindowedSerdes.TimeWindowedSerde<T>`
  - `WindowedSerdes.SessionWindowedSerde<T>`

Serializers
: - `TimeWindowedSerializer<T>`
  - `SessionWindowedSerializer<T>`

Deserializers
: - `TimeWindowedDeserializer<T>`
  - `SessionWindowedDeserializer<T>`

#### Usage in code

When using windowed serdes in your application code, create instances by using
constructors or factory methods:

```java
// Time windowed serde - using factory method
Serde<Windowed<String>> timeWindowedSerde =
    WindowedSerdes.timeWindowedSerdeFrom(String.class, 500L);

// Time windowed serde - using constructor
Serde<Windowed<String>> timeWindowedSerde2 =
    new WindowedSerdes.TimeWindowedSerde<>(Serdes.String(), 500L);

// Session windowed serde - using factory method
Serde<Windowed<String>> sessionWindowedSerde =
    WindowedSerdes.sessionWindowedSerdeFrom(String.class);

// Session windowed serde - using constructor
Serde<Windowed<String>> sessionWindowedSerde2 =
    new WindowedSerdes.SessionWindowedSerde<>(Serdes.String());

// Using individual serializers/deserializers
TimeWindowedSerializer<String> serializer = new TimeWindowedSerializer<>(Serdes.String().serializer());
TimeWindowedDeserializer<String> deserializer = new TimeWindowedDeserializer<>(Serdes.String().deserializer(), 500L);
```

#### Usage in command line

When using command-line tools, like `kafka-console-consumer`, you can
configure windowed deserializers by passing the inner class and window size
with configuration properties. The property names use a prefix pattern:

```bash
# Time windowed deserializer configuration
--formatter-property print.key=true \
--formatter-property key.deserializer=org.apache.kafka.streams.kstream.TimeWindowedDeserializer \
--formatter-property key.deserializer.windowed.inner.deserializer.class=org.apache.kafka.common.serialization.StringDeserializer \
--formatter-property key.deserializer.window.size.ms=500

# Session windowed deserializer configuration
--formatter-property print.key=true \
--formatter-property key.deserializer=org.apache.kafka.streams.kstream.SessionWindowedDeserializer \
--formatter-property key.deserializer.windowed.inner.deserializer.class=org.apache.kafka.common.serialization.StringDeserializer
```

#### Deprecated configs

The following `StreamsConfig` parameters are deprecated in favor of passing
parameters directly to serializer/deserializer constructors:

- `StreamsConfig.WINDOWED_INNER_CLASS_SERDE` is deprecated in favor of
  `TimeWindowedSerializer.WINDOWED_INNER_SERIALIZER_CLASS` and
  `TimeWindowedDeserializer.WINDOWED_INNER_DESERIALIZER_CLASS`.
- `StreamsConfig.WINDOW_SIZE_MS_CONFIG` is deprecated in favor of
  `TimeWindowedDeserializer.WINDOW_SIZE_MS_CONFIG`.

<a id="streams-developer-guide-serdes-custom"></a>

## Implementing custom Serdes

If you need to implement custom Serdes, your best starting point is to take a look at the source code references of
existing Serdes (see previous section).  Typically, your workflow is similar to:

1. Write a *serializer* for your data type `T` by implementing
   [org.apache.kafka.common.serialization.Serializer](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java).
2. Write a *deserializer* for `T` by implementing
   [org.apache.kafka.common.serialization.Deserializer](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java).
3. Write a *serde* for `T` by implementing
   [org.apache.kafka.common.serialization.Serde](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/serialization/Serde.java),
   which you either do manually (see existing Serdes in the previous section) or by leveraging helper functions in
   [Serdes](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/serialization/Serdes.java)
   such as `Serdes.serdeFrom(Serializer<T>, Deserializer<T>)`.
   You must implement your own class (that has no generic types) if you want to use your custom serde in the configuration provided to `KafkaStreams`.
   If your serde class has generic types or you use `Serdes.serdeFrom(Serializer<T>, Deserializer<T>)`, you can pass your serde only
   via methods calls (for example `builder.stream("topicName", Consumed.with(...))`.

## Kafka Streams DSL for Scala implicit serdes

When using the Kafka Streams DSL for Scala, you’re not required to configure a
default serde. In fact, it’s not supported. Serdes are instead provided
implicitly by default implementations for common primitive datatypes. For more
information, see [Implicit Serdes](dsl-api.md#streams-developer-guide-dsl-scala-implicit-serdes) and
[User-Defined Serdes](dsl-api.md#streams-developer-guide-dsl-scala-user-serdes).

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).