Add optional field to avro schema for schema registry compatibility mode FULL,

Viewed 40
[
  {
    "type": "record",
    "namespace":"kafka.avro.personal.schema.nested",
    "name": "Address",
    "fields": [
      {
        "name": "streetaddress",
        "type": "string"
      },
      {
        "name": "city",
        "type": "string"
      }
    ]
  },
  {
    "type": "record",
    "name": "Personal",
    "namespace":"kafka.avro.personal.schema.nested",
    "fields": [
      {
        "name": "firstname",
        "type": "string"
      },
      {
        "name": "lastname",
        "type": "string"
      },
      {"name":"status",
      "default" :"SALARY", "type": { "type": "enum", "name": "Status",
            "symbols" : ["RETIRED", "SALARY", "HOURLY", "PART_TIME"]}
      },
      {
        "name": "address",
        "type": ["null","Address"]
      }
       
    ]
  }
]

Ran the producer which auto registered schema in schema registry. Then added new optional field. Tried to publish again and ran into incompatibility issue, even though Compatibility FULL allows addition and deletion of optional fields.

 {
   "name": "optionalT",
   "type": ["null","string"],
   "default": null
 }

Stack Trace:

Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error registering Avro schema: {"type":"record","name":"Personal","namespace":"kafka.avro.personal.schema.nested","fields":[{"name":"firstname","type":"string"},{"name":"lastname","type":"string"},{"name":"status","type":{"type":"enum","name":"Status","symbols":["RETIRED","SALARY","HOURLY","PART_TIME"]},"default":"SALARY"},{"name":"address","type":["null",{"type":"record","name":"Address","fields":[{"name":"streetaddress","type":"string"},{"name":"city","type":"string"}]}]},{"name":"optionalT","type":["null","string"],"default":null}]}
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:103)
    at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53)
    at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:952)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:912)
    at evolution.compatibility.DataProducerPerson.main(DataProducerPerson.java:51)
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema being registered is incompatible with an earlier schema; error code: 409
    at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:236)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:265)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:365)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:357)
    at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:343)
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:168)
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:222)
    at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:198)
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:70)
    ... 5 more
0 Answers
Related