Avro backward schema evolution throws ClassCastException

I am getting a ClassCastException while trying to test Avro schema evolution with a simple Java program.

Avro version: 1.10.0

customer-v1.avsc

{
"type": "record",
"namespace": "com.practice.kafka",
"name": "Customer",
"doc": "Avro schema for Customer",
"fields": [
{"name":  "first_name", "type":  "string", "doc": "Customer first name"},
{"name":  "last_name", "type":  "string", "doc": "Customer last name"},
{"name":  "automated_email", "type":  "boolean", "default": true, "doc": "Receive marketing emails or not"}
]
}

customer-v2.avsc

{
"type": "record",
"namespace": "com.practice.kafka",
"name": "CustomerV2",
"doc": "Avro schema for Customer",
"fields": [
{"name":  "first_name", "type":  "string", "doc": "Customer first name"},
{"name":  "last_name", "type":  "string", "doc": "Customer last name"},
{"name":  "phone_number", "type":  ["null","boolean"], "default": null, "doc": "Optional phone number"},
{"name":  "email", "type":  "string", "default":  "missing@example.com", "doc":  "Optional email address"}
]
}

Program to serialize v1 and deserialize v2

package com.practice.kafka;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import java.io.File;
import java.io.IOException;
public class BackwardSchemaEvolutionSample {
public static void main(String[] args) {
// Step 1 - Create specific record
Customer customer = Customer.newBuilder().setFirstName("John").setLastName("Doe").setAutomatedEmail(false).build();
// Step 2 - Write specific record to a file
final DatumWriter<Customer> datumWriter = new SpecificDatumWriter<>();
try (DataFileWriter<Customer> dataFileWriter = new DataFileWriter<>(datumWriter)) {
dataFileWriter.create(customer.getSchema(), new File("customer-v1.avro"));
dataFileWriter.append(customer);
} catch (IOException e) {
e.printStackTrace();
}
// Step 3 - Read specific record from a file
final File file = new File("customer-v1.avro");
final DatumReader<CustomerV2> datumReader = new SpecificDatumReader<>();
CustomerV2 customerRecord;
try (DataFileReader<CustomerV2> dataFileReader = new DataFileReader<>(file, datumReader)) {
customerRecord = dataFileReader.next();
System.out.println(customerRecord.toString());
} catch (IOException e) {
e.printStackTrace();
}
}
}

Result

Exception in thread "main" java.lang.ClassCastException: class com.practice.kafka.Customer cannot be cast to class com.practice.kafka.CustomerV2 (com.practice.kafka.Customer and com.practice.kafka.CustomerV2 are in unnamed module of loader 'app')
at com.practice.kafka.SchemaEvolutionSample.main(SchemaEvolutionSample.java:34)

Can you please let me know how to fix this error?

You defined 2 data types Customer and Customer2 and you cannot have any casting since they are not in inherit relation.
So Java can't do casting and you are getting ClassCastException.
In your code only solution is to catch ClassCastException and in catch block convert Customer to Customer2.

I assume that you are emulating changes of your schema in Kafka environment.
In this scenario you will extend existing avro schema by adding new fields, or removing old fields.

As far as the name of class stays the same, avro schema changes will work.