AVRO-4076: [java] Parse Fields before adding to the context.

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

@@ -402,9 +402,12 @@ public boolean equals(Object o) {
     if (!(o instanceof Protocol))
       return false;
     Protocol that = (Protocol) o;
+
+    List<Schema> resolvedSchemas = this.context.resolveAllSchemas();
+    List<Schema> thatResolvedSchemas = that.context.resolveAllSchemas();
     return Objects.equals(this.name, that.name) && Objects.equals(this.namespace, that.namespace)
-        && Objects.equals(this.context.resolveAllSchemas(), that.context.resolveAllSchemas())
-        && Objects.equals(this.messages, that.messages) && this.propsEqual(that);
+        && Objects.equals(this.messages, that.messages) && this.propsEqual(that)
+        && resolvedSchemas.size() == thatResolvedSchemas.size() && resolvedSchemas.containsAll(thatResolvedSchemas);
   }
 
   @Override

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

@@ -1862,7 +1862,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())
@@ -1877,6 +1876,7 @@ private static Schema parseRecord(JsonNode schema, ParseContext context, String
             name, f.name(), getOptionalText(field, "logicalType"));
     }
     result.setFields(fields);
+    context.put(result);
     parsePropertiesAndLogicalType(schema, result, SCHEMA_RESERVED);
     parseAliases(schema, result);
     return result;

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

@@ -599,25 +599,30 @@ void testParseMultipleFile() throws IOException {
     File f1 = new File(directory.getPath(), "ApplicationEvent.avsc");
     File f2 = new File(directory.getPath(), "DocumentInfo.avsc");
     File f3 = new File(directory.getPath(), "MyResponse.avsc");
+    File f4 = new File(directory.getPath(), "EmbeddedRecord.avsc");
     Assertions.assertTrue(f1.exists(), "File not exist for test " + f1.getPath());
     Assertions.assertTrue(f2.exists(), "File not exist for test " + f2.getPath());
     Assertions.assertTrue(f3.exists(), "File not exist for test " + f3.getPath());
+    Assertions.assertTrue(f4.exists(), "File not exist for test " + f4.getPath());
     SchemaParser parser = new SchemaParser();
     parser.parse(f1);
     parser.parse(f2);
     parser.parse(f3);
+    parser.parse(f4);
     final List<Schema> schemas = parser.getParsedNamedSchemas();
-    Assertions.assertEquals(3, schemas.size());
-    Schema schemaAppEvent = schemas.get(0);
-    Schema schemaDocInfo = schemas.get(1);
-    Schema schemaResponse = schemas.get(2);
+    Assertions.assertEquals(4, schemas.size());
+    Schema eventHeaderEvent = schemas.get(0);
+    Schema schemaAppEvent = schemas.get(1);
+    Schema schemaDocInfo = schemas.get(2);
+    Schema schemaResponse = schemas.get(3);
     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);
     Assertions.assertNotNull(schemaDocInfo);
     Assertions.assertNotNull(schemaResponse);
+    Assertions.assertNotNull(eventHeaderEvent);
   }
 
   @Test

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

@@ -4,6 +4,20 @@
   "doc": "",
   "name": "ApplicationEvent",
   "fields": [
+    {
+      "name": "embedded_record",
+      "type": {
+        "type": "record",
+        "name": "EmbeddedRecord",
+        "namespace": "model",
+        "fields": [
+          {
+            "name": "type",
+            "type": "string"
+          }
+        ]
+      }
+    },
     {
       "name": "applicationId",
       "type": "string",

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

@@ -0,0 +1,11 @@
+{
+  "namespace": "model",
+  "type" : "record",
+  "name" : "EmbeddedRecord",
+  "fields" : [
+    {
+      "name" : "type",
+      "type" : "string"
+    }
+  ]
+}