@Override
  public Object deserialize(String s, byte[] bytes) {
    return this.deserialize(bytes, readerSchema);
  }
@Test
  public void testDeserialize() {
    // Test deserializing evolved schema.
    // Verify that records from older schemas are deserialized to the latest schema
    String schemaRegistryUrl = "mock://my-scope-name";
    String subject = "mytopic";
    SchemaRegistryClient mockRegistryClient = mockSchemaRegistryClient(schemaRegistryUrl, subject);

    Map<String, Object> map = new HashMap<>();
    map.put(AbstractKafkaAvroSerDeConfig.AUTO_REGISTER_SCHEMAS, true);
    map.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
    Serializer<Object> serializer = new KafkaAvroSerializer(mockRegistryClient);
    serializer.configure(map, true);

    byte[] bytes =
        serializer.serialize(
            subject,
            new GenericRecordBuilder(AVRO_SCHEMA_V1)
                .set("name", "KeyName")
                .set("age", 1)
                .set("favorite_number", 2)
                .set("favorite_color", "color3")
                .build());

    Object deserialized =
        mockDeserializerProvider(schemaRegistryUrl, subject, null)
            .getDeserializer(new HashMap<>(), true)
            .deserialize(subject, bytes);

    GenericRecord expected =
        new GenericRecordBuilder(AVRO_SCHEMA)
            .set("name", "KeyName")
            .set("favorite_number", 2)
            .set("favorite_color", "color3")
            .build();

    assertEquals(expected, deserialized);
  }