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

この手順では、Service Registry からの Apache Avro スキーマを使用するように Java で書かれた Kafka Streams クライアントを設定する方法について説明します。

前提条件

  • Service Registry がインストールされている必要があります。
  • スキーマが Service Registry に登録されている必要があります。

手順

  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)
);