Почему kafka-avro-console-продюсер не соблюдает значение по умолчанию для поля?

Question

Хотя для поля определено значение по умолчанию, 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, но, как я уже сказал, это не имеет значения.

Answer 1

Как определено в Avro spec

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

Следовательно, производитель все еще должен предоставить это поле.

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

Answer 2

Ошибка указывает, что сообщение несовместимо с определенной вами схемой 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