7.6. Kafka Streams 애플리케이션의 스키마 사용

프로세스

1. 서비스 레지스트리 URL을 사용하여 Java 클라이언트를 생성하고 구성합니다.

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

2. serializer 및 deserializer를 구성합니다.

   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

   서비스 레지스트리에서 제공하는 Avro serializer입니다.

   2

   서비스 레지스트리에서 제공하는 Avro deserializer입니다.

   3

   Avro 형식으로 역직렬화를 위해 서비스 레지스트리 URL 및 Avro 리더를 구성합니다.

3. Kafka Streams 클라이언트를 생성합니다.

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