[
{
"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