// Module included in the following assemblies:
//  assembly-using-kafka-client-serdes

[id='registry-serdes-config-stream-{context}']
= Using a schema from a Kafka Streams application

[role="_abstract"]
This procedure describes how to configure a Kafka Streams client written in Java to use an Apache Avro schema from {registry}.

.Prerequisites

* {registry} is installed
* The schema is registered with {registry}

.Procedure

. Create and configure a Java client with the {registry} URL:
+
[source,shell,subs="+quotes,attributes"]
----
String registryUrl = "https://registry.example.com/apis/registry/v2";

RegistryService client = RegistryClient.cached(registryUrl);
----

. Configure the serializer and deserializer:
+
[source,java,subs="+quotes,attributes"]
----
Serializer<LogInput> serializer = new AvroKafkaSerializer<LogInput>(); <1>

Deserializer<LogInput> deserializer = new AvroKafkaDeserializer <LogInput>(); <2>

Serde<LogInput> logSerde = Serdes.serdeFrom(
    serializer,
    deserializer
); 

Map<String, Object> config = new HashMap<>();
config.put(SerdeConfig.REGISTRY_URL, registryUrl);
config.put(AvroKafkaSerdeConfig.USE_SPECIFIC_AVRO_READER, true);
logSerde.configure(config, false); <3>

----
<1> The Avro serializer provided by {registry}.
<2> The Avro deserializer provided by {registry}.
<3> Configures the {registry} URL and the Avro reader for deserialization in Avro format.

. Create the Kafka Streams client:
+
[source,java,subs="+quotes,attributes"]
----
KStream<String, LogInput> input = builder.stream(
    INPUT_TOPIC,
    Consumed.with(Serdes.String(), logSerde)
);
----
