AVRO-3940: [java] Allow schema redefinition when equal (#3304)

ParseContext.put accept known schemas when strictly identical.
Call to this method was made before schema was fully parsed avoiding
exact schema redefinition in files.

Call ParseContext.put when schema is fully parsed.

This has an inpact on the schemas ordering returned by the parser.

Author: RustedBones

lang/java/avro/src/main/java/org/apache/avro/ParseContext.java

@@ -277,7 +277,7 @@ public void rollback() {
    * references, even if parsed from different files. Note: the context must be
    * committed for this method to work.
    *
-   * @return all parsed schemas, in the order they were parsed
+   * @return all parsed schemas
    * @throws AvroTypeException if a schema reference cannot be resolved
    */
   public List<Schema> resolveAllSchemas() {

lang/java/avro/src/main/java/org/apache/avro/Schema.java

@@ -1863,7 +1863,6 @@ private static Schema parseRecord(JsonNode schema, ParseContext context, String
     Name name = parseName(schema, currentNameSpace);
     String doc = parseDoc(schema);
     Schema result = new RecordSchema(name, doc, isTypeError);
-    context.put(result);
 
     JsonNode fieldsNode = schema.get("fields");
     if (fieldsNode == null || !fieldsNode.isArray())
@@ -1880,6 +1879,7 @@ private static Schema parseRecord(JsonNode schema, ParseContext context, String
     result.setFields(fields);
     parsePropertiesAndLogicalType(schema, result, SCHEMA_RESERVED);
     parseAliases(schema, result);
+    context.put(result);
     return result;
   }
 
@@ -1925,9 +1925,9 @@ private static Schema parseEnum(JsonNode schema, ParseContext context, String cu
     }
 
     Schema result = new EnumSchema(name, doc, symbols, defaultSymbol);
-    context.put(result);
     parsePropertiesAndLogicalType(schema, result, ENUM_RESERVED);
     parseAliases(schema, result);
+    context.put(result);
     return result;
   }
 
@@ -1960,9 +1960,9 @@ private static Schema parseFixed(JsonNode schema, ParseContext context, String c
       throw new SchemaParseException("Invalid or no size: " + schema);
 
     Schema result = new FixedSchema(name, doc, sizeNode.intValue());
-    context.put(result);
     parsePropertiesAndLogicalType(schema, result, SCHEMA_RESERVED);
     parseAliases(schema, result);
+    context.put(result);
     return result;
   }
 

lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectData.java

@@ -723,7 +723,6 @@ protected Schema createSchema(Type type, Map<String, Schema> names) {
           boolean error = Throwable.class.isAssignableFrom(c);
           schema = Schema.createRecord(name, doc, space, error);
           consumeAvroAliasAnnotation(c, schema);
-          names.put(c.getName(), schema);
           for (Field field : getCachedFields(c))
             if ((field.getModifiers() & (Modifier.TRANSIENT | Modifier.STATIC)) == 0
                 && !field.isAnnotationPresent(AvroIgnore.class)) {
@@ -769,6 +768,7 @@ protected Schema createSchema(Type type, Map<String, Schema> names) {
         }
         names.put(fullName, schema);
       }
+      names.put(c.getName(), schema);
       return schema;
     }
     return super.createSchema(type, names);
@@ -907,11 +907,7 @@ public Protocol getProtocol(Class iface) {
       }
     }
 
-    // reverse types, since they were defined in reference order
-    List<Schema> types = new ArrayList<>(names.values());
-    Collections.reverse(types);
-    protocol.setTypes(types);
-
+    protocol.setTypes(new ArrayList<>(names.values()));
     return protocol;
   }
 

lang/java/avro/src/test/java/org/apache/avro/TestSchema.java

@@ -32,7 +32,10 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
@@ -606,13 +609,14 @@ void testParseMultipleFile() throws IOException {
     parser.parse(f1);
     parser.parse(f2);
     parser.parse(f3);
-    final List<Schema> schemas = parser.getParsedNamedSchemas();
+    final Map<String, Schema> schemas = parser.getParsedNamedSchemas().stream()
+        .collect(Collectors.toMap(Schema::getName, Function.identity()));
     Assertions.assertEquals(3, schemas.size());
-    Schema schemaAppEvent = schemas.get(0);
-    Schema schemaDocInfo = schemas.get(1);
-    Schema schemaResponse = schemas.get(2);
+    Schema schemaAppEvent = schemas.get("ApplicationEvent");
+    Schema schemaDocInfo = schemas.get("DocumentInfo");
+    Schema schemaResponse = schemas.get("MyResponse");
     Assertions.assertNotNull(schemaAppEvent);
-    Assertions.assertEquals(3, schemaAppEvent.getFields().size());
+    Assertions.assertEquals(4, schemaAppEvent.getFields().size());
     Field documents = schemaAppEvent.getField("documents");
     Schema docSchema = documents.schema().getTypes().get(1).getElementType();
     Assertions.assertEquals(docSchema, schemaDocInfo);

lang/java/avro/src/test/resources/multipleFile/ApplicationEvent.avsc

@@ -22,6 +22,22 @@
       }],
       "doc": "",
       "default": null
+    },
+    {
+      "name": "response",
+      "type": {
+        "namespace": "model",
+        "type": "record",
+        "doc": "",
+        "name": "MyResponse",
+        "fields": [
+          {
+            "name": "isSuccessful",
+            "type": "boolean",
+            "doc": "Indicator for successful or unsuccessful call"
+          }
+        ]
+      }
     }
   ]
 

lang/java/maven-plugin/src/test/avro/multipleSchemas/ApplicationEvent.avsc

@@ -22,6 +22,22 @@
       }],
       "doc": "",
       "default": null
+    },
+    {
+      "name": "response",
+      "type": {
+        "namespace": "model",
+        "type": "record",
+        "doc": "",
+        "name": "MyResponse",
+        "fields": [
+          {
+            "name": "isSuccessful",
+            "type": "boolean",
+            "doc": "Indicator for successful or unsuccessful call"
+          }
+        ]
+      }
     }
   ]