Почему кафка-авро-консоль-производитель не соблюдает значение поля по умолчанию?

Хотя значение по умолчанию определено для поля, kafka-avro-console-producer полностью игнорирует его:

$ kafka-avro-console-producer --broker-list localhost:9092 --topic test-avro \
--property schema.registry.url=http://localhost:8081 --property \
value.schema='{"type":"record","name":"myrecord1","fields": \
[{"name":"f1","type":"string"},{"name": "f2", "type": "int", "default": 0}]}'

{"f1": "value1"}

org.apache.kafka.common.errors.SerializationException: Error 
deserializing json {"f1": "value1"} to Avro of schema 
{"type":"record","name":"myrecord1","fields": 
[{"name":"f1","type":"string"},{"name":"f2","type":"int","default":0}]}
Caused by: org.apache.avro.AvroTypeException: Expected int. Got END_OBJECT
    at org.apache.avro.io.JsonDecoder.error(JsonDecoder.java:698)
    at org.apache.avro.io.JsonDecoder.readInt(JsonDecoder.java:172)
    at org.apache.avro.io.ValidatingDecoder.readInt(ValidatingDecoder.java:83)
    at org.apache.avro.generic.GenericDatumReader.readInt(GenericDatumReader.java:511)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:182)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:240)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:230)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:174)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)
    at io.confluent.kafka.formatter.AvroMessageReader.jsonToAvro(AvroMessageReader.java:213)
    at io.confluent.kafka.formatter.AvroMessageReader.readMessage(AvroMessageReader.java:180)
    at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:54)
    at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)

Как тогда использовать его, чтобы принять значение по умолчанию? Конфигурация верхнего уровня установлена ​​на "BACKWARD" Проверка уровня совместимости, хотя я не думаю, что это имеет какое-либо отношение к вопросу. Это схема версии 2, а версия 1 была определена только с полем f1, но, как я уже сказал, я не думаю, что это имеет значение.


apache-kafka avro confluent-schema-registry kafka-producer-api
person hdjur_jcv    schedule 02.04.2019    source источник
comment
AFAIK, значение default применимо только к читателю / потребителю. Поле по-прежнему должно быть определено автором / производителем.   -  person OneCricketeer    schedule 03.04.2019
comment
Я уже проголосовал за это как за полезный комментарий, и если вы продвинете его в ответ, я приму его как тот, который решает проблему и является наиболее полезным.   -  person hdjur_jcv    schedule 03.04.2019
comment
@hdjur_jcv Я считаю, что мой ответ объясняет, как делать то, что описал cricket_007.   -  person Giorgos Myrianthous    schedule 03.04.2019
comment
Привет, Giorgos Myrianthous, я также проголосовал за ваш ответ как за полезный, и я благодарю вас за это, но я думаю, что ответ cricket_007 правильный, потому что он раскрывает мое заблуждение. Включение нулевого входного значения на стороне производителя не означает использование нулевого значения по умолчанию на стороне потребителя (что должно быть единственной моей целью), как видно из вашего примера.   -  person hdjur_jcv    schedule 03.04.2019

Ответы (2)


arrow_upward
2
arrow_downward

Как определено в спецификации Avro

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

Таким образом, производитель все еще должен предоставить это поле.

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

person OneCricketeer    schedule 03.04.2019

arrow_upward
2
arrow_downward

Ошибка указывает на то, что сообщение несовместимо с определенной вами схемой Avro. Насколько я понимаю, вы хотите разрешить null значения для поля f2. Для этого вам нужно изменить свой value.schema на (обратите внимание на определение "type"):

value.schema='{"type":"record","name":"myrecord1","fields": [{"name":"f1","type":"string"},{"name": "f2", "type": ["null", "int"], "default": 0}]}'

но вам все равно нужно будет определить ключ f2 с нулевым значением. Следующее должно помочь вам:

kafka-avro-console-producer --broker-list localhost:9092 --topic test-avro \ 
    --property schema.registry.url=http://localhost:8081 \ 
    --property value.schema='{"type":"record","name":"myrecord1","fields": [{"name":"f1","type":"string"},{"name": "f2", "type": ["null", "int"], "default": 0}]}'

{"f1":"value1","f2":null}

И вы можете подтвердить, что это сработало, используя kafka-avro-console-consumer:

kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic test-avro --from-beginning
{"f1":"value1","f2":null}
^CProcessed a total of 1 messages
person Giorgos Myrianthous    schedule 02.04.2019