13.8. Kafka Streams アプリケーションからのスキーマの使用

手順

1. Service Registry で REST クライアントを作成および設定します。以下に例を示します。

   String registryUrl = "https://registry.example.com/api";
   RegistryService client = RegistryClient.cached(registryUrl);

2. シリアライザー、デシリアライザーの設定、および Kafka Streams クライアントを作成します。以下に例を示します。

   Serializer<LogInput> serializer = new AvroKafkaSerializer<>( 

   1

   
       client,
       new DefaultAvroDatumProvider<LogInput>().setUseSpecificAvroReader(true)
   );
   Deserializer<LogInput> deserializer = new AvroKafkaDeserializer <> ( 

   2

   
       client,
       new DefaultAvroDatumProvider<LogInput>().setUseSpecificAvroReader(true)
   );
   Serde<LogInput> logSerde = Serdes.serdeFrom( 

   3

   
       serializer,
       deserializer
   );
   KStream<String, LogInput> input = builder.stream( 

   4

   
       INPUT_TOPIC,
       Consumed.with(Serdes.String(), logSerde)
   );

   1

   Service Registry によって提供されるシリアライザー。

   2

   Service Registry によって提供されるデシリアライザー。

   3

   デシリアライズは Apache Avro 形式です。

   4

   Kafka Streams クライアントアプリケーション。