A Serde packages the serializer and deserializer for one Java type. Kafka Streams uses it at topology boundaries because Kafka brokers store and forward record keys and values as byte arrays, while stream operators work with concrete objects. section 6.4.4 section 6.3.1
Use the Serdes utility class to obtain built-in Serde implementations. Its factory methods cover String, ByteArray, Bytes, Long, Short, Integer, Double, Float, ByteBuffer, UUID, and Void. Keeping one Serde reference avoids repeatedly supplying separate key and value serializers and deserializers. section 6.3.3
Configure a string stream boundary
Create a reusable string Serde, then pass it to Consumed for reading and Produced for writing. Kafka Streams deserializes source records before mapValues converts values to uppercase, then serializes results for the output topic. section 6.3.1 section 6.3.3
Listing 6.6 Hello World: The Yelling app
public class KafkaStreamsYellingApp extends BaseStreamsApplication {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaStreamsYellingApp.class);
@Override
public Topology topology(Properties streamProperties) {
Serde<String> stringSerde = Serdes.String(); #1
StreamsBuilder builder = new StreamsBuilder(); #2
KStream<String, String> simpleFirstStream = builder.stream("src-topic",
Consumed.with(stringSerde, stringSerde)); #3
KStream<String, String> upperCasedStream =
simpleFirstStream.mapValues(value-> value.toUpperCase()); #4
upperCasedStream.to("out-topic",
Produced.with(stringSerde, stringSerde)); #5
return builder.build(streamProperties);
}
public static void main(String[] args) throws Exception {
Properties streamProperties = new Properties();
streamProperties.put(StreamsConfig.APPLICATION_ID_CONFIG,
"yelling_app_id");
streamProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9092");
KafkaStreamsYellingApp yellingApp = new KafkaStreamsYellingApp();
Topology topology = yellingApp.topology(streamProperties);
try(KafkaStreams kafkaStreams =
new KafkaStreams(topology, streamProperties)) {
LOG.info("Hello World Yelling App Started");
kafkaStreams.start(); #6
LOG.info("Shutting down the Yelling APP now");
}
}
}
Follow the complete Yelling app from Serde creation through source configuration, value transformation, and sink serialization. section 6.3.3
Choose a Serde strategy by value type:
- Use a built-in Serdes factory for supported primitive or standard types, such as String. section 6.3.3
- Build a custom Serde<T> with a Serializer<T> and Deserializer<T> for arbitrary domain objects, then combine them with Serdes.serdeFrom. section 6.4.4
- Use Schema Registry-aware Serdes for Avro, Protobuf, or JSON-based schemas when schema compatibility and shared serialization tooling are needed. section 3.7 section 6.4.5
Serdes also cross topology boundaries into state: when a KTable materializes a state store, its key and value Serdes let that byte-array-backed store serialize and deserialize entries. Serdes supplied through Consumed can be passed to the store. section 8.2
