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

手順

1. Apicurio Registry URL を使用して Java クライアントを作成および設定します。

   Copy to Clipboard Toggle word wrap

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

2. シリアライザーおよびデシリアライザーを設定します。

   Copy to Clipboard Toggle word wrap

   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

   Apicurio Registry が提供する Avro シリアライザー。

   2

   Apicurio Registry が提供する Avro デシリアライザー。

   3

   Avro 形式でデシリアライズ用の Apicurio Registry URL および Avro リーダーを設定します。

3. Kafka Streams クライアントを作成します。

   Copy to Clipboard Toggle word wrap

   KStream<String, LogInput> input = builder.stream(
       INPUT_TOPIC,
       Consumed.with(Serdes.String(), logSerde)
   );