Kafka Streams - SerializationException: неизвестный магический байт - PullRequest
Новая
       27

Kafka Streams - SerializationException: неизвестный магический байт

0 голосов
/ 18 декабря 2018

Я пытаюсь создать приложение Kafka Streams, которое обрабатывает записи Avro, но я получаю следующую ошибку: 

Exception in thread "streams-application-c8031218-8de9-4d55-a5d0-81c30051a829-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately.
at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:74)
at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:91)
at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:567)
at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:900)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:801)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:749)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:719)
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

Я не уверен, что является причиной этой ошибки.Я просто пытаюсь сначала получить записи Avro в приложение, где они затем будут обработаны, а затем выведены в другую тему, но, похоже, это не работает.Я включил код из приложения ниже.Кто-нибудь может понять, почему он не работает? 

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-application");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    Serde<String> stringSerde = Serdes.String();
    Serde<trackingReport> specificAvroTrackingReportSerde = new SpecificAvroSerde<trackingReport>();

    specificAvroTrackingReportSerde.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"), false);


    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, trackingReport> inputreports = builder.stream("intesttopic", Consumed.with(stringSerde, specificAvroTrackingReportSerde));


    KStream<String, trackingReport> outputreports = inputreports;

    String outputTopic = "outtesttopic";
    outputreports.to(outputTopic, Produced.with(stringSerde, specificAvroTrackingReportSerde));

    Topology topology = builder.build();

    KafkaStreams streams = new KafkaStreams(topology, props);
    streams.start();

1 Ответ

0 голосов
/ 18 декабря 2018

Неизвестный магический байт!

Означает, что ваши данные не соответствуют формату связи, который ожидается для реестра схем.

Или, другими словами, данные, которые вы пытаетесь прочитать, это не Avro.Или, если это так, и схема отправляется как часть сообщения (нужно будет увидеть код вашего производителя), то вам не следует использовать десериализаторы Confluent Avro, которые ожидают определенный формат байтов в сообщении.

Вы можете ожидать ту же ошибку, запустив kafka-avro-console-consumer

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

DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
...