Неизвестный магический байт с кафкой-авро-консоль-потребителем

Question

Я пытался подключиться с kafka-avro-console-consumer от Confluent к нашему устаревшему кластеру Kafka, который был развернут без реестра Confluent Schema.Я предоставил схему явно, используя такие свойства, как:

kafka-console-consumer --bootstrap-server kafka02.internal:9092 \
    --topic test \
    --from-beginning \
    --property key.schema='{"type":"long"}' \
    --property value.schema='{"type":"long"}'

, но я получаю «Неизвестный магический байт!»ошибка с org.apache.kafka.common.errors.SerializationException

Возможно ли использовать сообщения Avro от Kafka с помощью Confluent kafka-avro-console-consumer, которые не были сериализованы с AvroSerializer из Confluent и с реестром схем?

Answer 1

kafka-console-consumer не знает о key.schema или value.schema, знает только производитель Avro . Исходный код здесь

Обычный потребитель консоли не заботится о формате данных - он просто печатает байты в кодировке UTF8

Свойство, которое kafka-<b>avro</b>-console-consumerпринимает только schema.registry.url.Итак, чтобы ответить на вопрос, да, его необходимо сериализовать с помощью Confluent сериализаторов.

Answer 2

Сериализатор / десериализатор реестра Confluent Schema использует формат wire , который включает информацию об идентификаторе схемы и т. Д. В начальные байты сообщения.

Если ваше сообщение не было сериализовано с помощью сериализатора реестра Schema, вы не сможете десериализовать его с ним и получите ошибку Unknown magic byte!.

Итак, вы 'Вам нужно будет написать получателя, который извлекает сообщения, выполняет десериализацию с использованием ваших схем Avro avsc, а затем, предполагая, что вы хотите сохранить данные, повторно сериализуете их с помощью сериализатора реестра схем

Редактировать: Я недавно написал статью, которая объясняет все это более подробно: https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained