Блог

Я написал небольшой класс Java для проверки использования темы Kafka, закодированной в Avro.

     Properties appProps = new Properties();

    appProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "http://***kfk14bro1.lc:9092");
    appProps.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://***kfk14str1.lc:8081");
    appProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "consumer");
    appProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    appProps.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,LogAndContinueExceptionHandler.class);


    StreamsBuilder streamsBuilder = new StreamsBuilder();

    streamsBuilder.stream(
                  "coordinates", Consumed.with(Serdes.String(), new GenericAvroSerde()))
              .peek((key, value) -> System.out.println("key="   key   ", value="   value));

    new KafkaStreams(streamsBuilder.build(), appProps).start();
  

Когда я запускаю этот класс, SerdeConfigs регистрируются нормально, что можно увидеть в приведенном ниже журнале:

 [consumer-56b0e0ca-d336-45cc-b388-46a68dbfab8b-StreamThread-1] INFO io.confluent.kafka.serializers.KafkaAvroSerializerConfig - KafkaAvroSerializerConfig values: 
    schema.registry.url = [http://***kfk14str1.lc:8081]
    basic.auth.user.info = [hidden]
    auto.register.schemas = true
    max.schemas.per.subject = 1000
    basic.auth.credentials.source = URL
    schema.registry.basic.auth.user.info = [hidden]
    value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
    key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy

[normal-consumer-56b0e0ca-d336-45cc-b388-46a68dbfab8b-StreamThread-1] INFO io.confluent.kafka.serializers.KafkaAvroDeserializerConfig - KafkaAvroDeserializerConfig values: 
    schema.registry.url = [http://***kfk14str1.lc:8081]
    basic.auth.user.info = [hidden]
    auto.register.schemas = true
    max.schemas.per.subject = 1000
    basic.auth.credentials.source = URL
    schema.registry.basic.auth.user.info = [hidden]
    specific.avro.reader = false
    value.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
    key.subject.name.strategy = class io.confluent.kafka.serializers.subject.TopicNameStrategy
  

но сообщения не используются и генерирует приведенный ниже журнал для каждого сообщения:

 [normal-consumer-56b0e0ca-d336-45cc-b388-46a68dbfab8b-StreamThread-1] WARN org.apache.kafka.streams.errors.LogAndContinueExceptionHandler - Exception caught during Deserialization, taskId: 0_0, topic: coordinates, partition: 0, offset: 782205986
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 83
Caused by: java.lang.NullPointerException
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:116)
at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:88)
at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:55)
at io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer.deserialize(GenericAvroDeserializer.java:63)
at io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer.deserialize(GenericAvroDeserializer.java:39)
at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:58)
at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:60)
  

Но я могу просто отлично читать с консоли пользователя avro, поэтому я знаю, что с данными, записанными в тему, все в порядке. Приведенная ниже команда выводит журналы в порядке:

 ~/kafka/confluent-5.1.2/bin/kafka-avro-console-consumer --bootstrap-server http://***kfk14bro1.lc:9092 --topic coordinates --property schema.registry.url=http://***kfk14str1.lc:8081 --property auto.offset.reset=latest
  

Когда вы создаете экземпляр Avro Serde самостоятельно, он не настраивается автоматически с URL-адресом схемы-реестра.

Итак, либо вы должны настроить это самостоятельно, либо определить serdes по умолчанию, добавив:

 appProps.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
appProps.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class.getName());
  

И путем удаления

 Consumed.with(Serdes.String(), new GenericAvroSerde())
  

Для настройки Serde используйте следующий код (адаптируйте его к вашей ситуации):

 GenericAvroSerde genericAvroSerde = new GenericAvroSerde();
boolean isKeySerde = false;
genericAvroSerde.configure(
     Collections.singletonMap(
         AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
         "http://confluent-schema-registry-server:8081/"),
     isKeySerde);