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

手順

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

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

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

   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

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

   2

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

   3

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

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

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