Specific Record cannot be deserializer if class name changes

[simplesteph]

The schema registry accepts a schema change from:

{
     "type": "record",
     "namespace": "com.example",
     "name": "CustomerV1",
     "fields": [
       { "name": "first_name", "type": "string", "doc": "First Name of Customer" }
     ]
}

to

{
     "type": "record",
     "namespace": "com.example",
     "name": "CustomerV2",
     "fields": [
       { "name": "first_name", "type": "string", "doc": "First Name of Customer" },
       { "name": "last_name", "type": "string", "default": "NA" }
     ]
}

(emphasis on CustomerV1 to CustomerV2)

The deserializer fails with:

Exception in thread "main" java.lang.ClassCastException: com.example.CustomerV2 cannot be cast to com.example.CustomerV1

OR 

Exception in thread "main" org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition customer-avro-0 at offset 4. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 12
Caused by: org.apache.kafka.common.errors.SerializationException: Could not find class com.example.CustomerV2 specified in writer's schema whilst finding reader's schema for a SpecificRecord.

The use case was to show V1 -> V2 and schema evolution within one project I think this could be solved in two ways

1. The schema registry fully rejects that change. After all, I'm not even sure changing a record name is a compatible change
2. We should be able to provide a schema to the deserializer, making it a specific deserializer linked to a SpecificRecord class. This way protected Object deserialize(byte[] payload, Schema readerSchema) could be called. Currently that method only is called in the tests... a bit unexpected.

[finleysg]

The inability to specify a reader schema in order to support schema evolution is surprising. I have spent a lot of time pouring through the code thinking that I was simply missing something, but afaik the capability is just not there.

The primary use case for us in using Avro and the registry is to support FORWARD compatibility so a system relying on an older schema does not have to change. At least not immediately.

Ended up writing my own Avro serde code, which I'm sure is not as good as this library.

[Tito3nity]

How to solve it? Already tried but always failed. Below the result : Error deserializing key/value for partition customer-avro-0 at offset xxxx. If needed, please seek past the record to continue consumption.

[frankgrimes97]

Hi, we also would like to have the reader schema properly derived from the SpecificRecord class provided by the client application. (derived from the schema of the compiled SpecificRecord's schema on its classpath, I suppose)

We'd be willing to work on a PR for this... but before we undertake that work we'd like to have some indication that the feature would be accepted upstream in this project.

Thoughts?