Fixes deserialization when reader/writer schemas are different. (Azure#26592)

* Add test case for portal.

* Adding code to consider both reader and writer schemas.

* Separating fields because there is ambiguity about generated class schemas.

* Updating test cases to consider different reader/writer schemas.

* Adding testcase for forwards compatibility checks.

* Adding person2 java and avro file.

* Adding recording file.

* Specify avro files.

Author: conniey

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/pom.xml

@@ -130,7 +130,8 @@
               <testSourceDirectory>src/samples/resources/avro/</testSourceDirectory>
               <testOutputDirectory>${project.basedir}/src/samples/java/</testOutputDirectory>
               <testIncludes>
-                <testInclude>**/*.avro</testInclude>
+                <testInclude>Person.avro</testInclude>
+                <testInclude>HandOfCards.avro</testInclude>
               </testIncludes>
               <stringType>String</stringType>
             </configuration>

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/src/main/java/com/azure/data/schemaregistry/apacheavro/AvroSerializer.java

@@ -23,12 +23,16 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 
 /**
  * Class containing implementation of Apache Avro serializer
@@ -156,8 +160,8 @@ <T> T decode(byte[] bytes, byte[] schemaBytes, TypeReference<T> typeReference) {
         Objects.requireNonNull(bytes, "'bytes' must not be null.");
         Objects.requireNonNull(schemaBytes, "'schemaBytes' must not be null.");
 
-        String schemaString = new String(schemaBytes, StandardCharsets.UTF_8);
-        Schema schemaObject = parseSchemaString(schemaString);
+        final String schemaString = new String(schemaBytes, StandardCharsets.UTF_8);
+        final Schema schemaObject = parseSchemaString(schemaString);
 
         if (isSingleObjectEncoded(bytes)) {
             final BinaryMessageDecoder<T> messageDecoder = new BinaryMessageDecoder<>(SpecificData.get(), schemaObject);
@@ -229,6 +233,47 @@ static boolean isSingleObjectEncoded(byte[] schemaBytes) {
         return V1_HEADER[0] == schemaBytes[0] && V1_HEADER[1] == schemaBytes[1];
     }
 
+    /**
+     * Gets the type's schema if there is one.
+     *
+     * @param clazz Class to get schema.
+     * @param <T> The type of object.
+     *
+     * @return The {@link Schema} or {@code null} if it was not a GenericContainer, could not instantiate the type, or
+     *     there was no default constructor.
+     */
+    <T> Schema getSchemaFromTypeReference(Class<T> clazz) {
+        if (!GenericContainer.class.isAssignableFrom(clazz)) {
+            return null;
+        }
+
+        final Optional<Constructor<?>> defaultConstructor;
+        try {
+            defaultConstructor = Arrays.stream(clazz.getDeclaredConstructors())
+                .filter(constructor -> constructor.getParameterCount() == 0)
+                .findFirst();
+        } catch (SecurityException e) {
+            logger.info("Could not get declaring constructors for deserializing T ({}). Using writer schema.",
+                clazz, e);
+            return null;
+        }
+
+        if (!defaultConstructor.isPresent()) {
+            return null;
+        }
+
+        Object instance = null;
+        try {
+            instance = defaultConstructor.get().newInstance();
+        } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
+            logger.info("Could not create new instance for deserializing T ({}). Using writer schema.", clazz, e);
+        }
+
+        return instance instanceof GenericContainer
+            ? ((GenericContainer) instance).getSchema()
+            : null;
+    }
+
     /**
      * Gets a schema for the given class if it is an Avro primitive type.
      *
@@ -256,10 +301,8 @@ private static Schema getPrimitiveSchema(Class<?> clazz) {
      *
      * @return correct Avro DatumReader object given encoder configuration
      */
-    @SuppressWarnings("unchecked")
     private <T> DatumReader<T> getDatumReader(Schema writerSchema, TypeReference<T> typeReference) {
-        // Suppressing this warning because we know that the Type is a representation of the Class<T>
-        final Class<T> clazz = (Class<T>) typeReference.getJavaType();
+        final Class<T> clazz = typeReference.getJavaClass();
         final Schema primitiveSchema = getPrimitiveSchema(clazz);
 
         if (primitiveSchema != null) {
@@ -270,6 +313,14 @@ private <T> DatumReader<T> getDatumReader(Schema writerSchema, TypeReference<T>
             }
         }
 
+        final Schema readerSchema = getSchemaFromTypeReference(clazz);
+        if (readerSchema != null && !readerSchema.equals(writerSchema)) {
+            logger.verbose("The writer schema is different than reader schema. Using reader schema. "
+                + "Writer: '{}'. Reader: '{}'", writerSchema, readerSchema);
+
+            return new SpecificDatumReader<>(writerSchema, readerSchema);
+        }
+
         if (SpecificRecord.class.isAssignableFrom(clazz)) {
             return new SpecificDatumReader<>(writerSchema);
         } else {

sdk/schemaregistry/azure-data-schemaregistry-apacheavro/src/main/java/com/azure/data/schemaregistry/apacheavro/SchemaRegistryApacheAvroSerializer.java

@@ -115,7 +115,7 @@ public <T> Mono<T> deserializeAsync(InputStream inputStream, TypeReference<T> ty
                         new IllegalStateException("Illegal format: unsupported record format indicator in payload"));
                 }
 
-                String schemaId = getSchemaIdFromPayload(buffer);
+                final String schemaId = getSchemaIdFromPayload(buffer);
 
                 return this.schemaRegistryClient.getSchema(schemaId)
                     .handle((registryObject, sink) -> {
@@ -171,7 +171,10 @@ public Mono<Void> serializeAsync(OutputStream outputStream, Object object) {
             return monoError(logger, exception);
         }
 
-        return this.maybeRegisterSchema(serializerOptions.getSchemaGroup(), schema.getFullName(), schema.toString())
+        final String schemaFullName = schema.getFullName();
+        final String schemaString = schema.toString();
+
+        return this.maybeRegisterSchema(serializerOptions.getSchemaGroup(), schemaFullName, schemaString)
             .handle((id, sink) -> {
                 ByteBuffer recordFormatIndicatorBuffer = ByteBuffer
                     .allocate(RECORD_FORMAT_INDICATOR_SIZE)