AVRO-2254: [java] fix unresolved schema name (#2365)

* AVRO-2254: fix unresolved shema name

Author: clesaec

lang/java/compiler/src/main/java/org/apache/avro/compiler/idl/SchemaResolver.java

@@ -23,6 +23,7 @@
 import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 
 import org.apache.avro.Protocol;
@@ -44,6 +45,8 @@ private SchemaResolver() {
 
   private static final String UR_SCHEMA_NS = "org.apache.avro.compiler";
 
+  private static final AtomicInteger COUNTER = new AtomicInteger();
+
   /**
    * Create a schema to represent a "unresolved" schema. (used to represent a
    * schema where the definition is not known at the time) This concept might be
@@ -53,8 +56,8 @@ private SchemaResolver() {
    * @return
    */
   static Schema unresolvedSchema(final String name) {
-    Schema schema = Schema.createRecord(UR_SCHEMA_NAME, "unresolved schema", UR_SCHEMA_NS, false,
-        Collections.EMPTY_LIST);
+    Schema schema = Schema.createRecord(UR_SCHEMA_NAME + '_' + COUNTER.getAndIncrement(), "unresolved schema",
+        UR_SCHEMA_NS, false, Collections.EMPTY_LIST);
     schema.addProp(UR_SCHEMA_ATTR, name);
     return schema;
   }
@@ -66,8 +69,8 @@ static Schema unresolvedSchema(final String name) {
    * @return
    */
   static boolean isUnresolvedSchema(final Schema schema) {
-    return (schema.getType() == Schema.Type.RECORD && schema.getProp(UR_SCHEMA_ATTR) != null
-        && UR_SCHEMA_NAME.equals(schema.getName()) && UR_SCHEMA_NS.equals(schema.getNamespace()));
+    return (schema.getType() == Schema.Type.RECORD && schema.getProp(UR_SCHEMA_ATTR) != null && schema.getName() != null
+        && schema.getName().startsWith(UR_SCHEMA_NAME) && UR_SCHEMA_NS.equals(schema.getNamespace()));
   }
 
   /**

lang/java/compiler/src/test/idl/input/union.avdl

@@ -0,0 +1,16 @@
+@namespace("org.apache.avro.gen")
+protocol UnionFwd {
+
+	record TestRecord {
+		union {SR1, SR2} unionField;
+	}
+
+	record SR1 {
+		string field;
+	}
+
+	record SR2 {
+		string field;
+	}
+
+}

lang/java/compiler/src/test/idl/output/union.avpr

@@ -0,0 +1,38 @@
+{
+  "protocol": "UnionFwd",
+  "namespace": "org.apache.avro.gen",
+  "types": [
+    {
+      "type": "record",
+      "name": "TestRecord",
+      "fields": [
+        {
+          "name": "unionField",
+          "type": [
+            {
+              "type": "record",
+              "name": "SR1",
+              "fields": [
+                {
+                  "name": "field",
+                  "type": "string"
+                }
+              ]
+            },
+            {
+              "type": "record",
+              "name": "SR2",
+              "fields": [
+                {
+                  "name": "field",
+                  "type": "string"
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }
+  ],
+  "messages": {}
+}

lang/java/idl/src/main/java/org/apache/avro/idl/SchemaResolver.java

@@ -23,6 +23,7 @@
 
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -41,6 +42,8 @@ private SchemaResolver() {
 
   private static final String UR_SCHEMA_NS = "org.apache.avro.compiler";
 
+  private static final AtomicInteger COUNTER = new AtomicInteger();
+
   /**
    * Create a schema to represent an "unresolved" schema. (used to represent a
    * schema whose definition does not exist, yet).
@@ -49,8 +52,8 @@ private SchemaResolver() {
    * @return an unresolved schema for the given name
    */
   static Schema unresolvedSchema(final String name) {
-    Schema schema = Schema.createRecord(UR_SCHEMA_NAME, "unresolved schema", UR_SCHEMA_NS, false,
-        Collections.emptyList());
+    Schema schema = Schema.createRecord(UR_SCHEMA_NAME + '_' + COUNTER.getAndIncrement(), "unresolved schema",
+        UR_SCHEMA_NS, false, Collections.emptyList());
     schema.addProp(UR_SCHEMA_ATTR, name);
     return schema;
   }
@@ -62,8 +65,8 @@ static Schema unresolvedSchema(final String name) {
    * @return whether the schema is an unresolved schema
    */
   static boolean isUnresolvedSchema(final Schema schema) {
-    return (schema.getType() == Schema.Type.RECORD && schema.getProp(UR_SCHEMA_ATTR) != null
-        && UR_SCHEMA_NAME.equals(schema.getName()) && UR_SCHEMA_NS.equals(schema.getNamespace()));
+    return (schema.getType() == Schema.Type.RECORD && schema.getProp(UR_SCHEMA_ATTR) != null && schema.getName() != null
+        && schema.getName().startsWith(UR_SCHEMA_NAME) && UR_SCHEMA_NS.equals(schema.getNamespace()));
   }
 
   /**

lang/java/idl/src/test/idl/input/union.avdl

@@ -0,0 +1,16 @@
+@namespace("org.apache.avro.gen")
+protocol UnionFwd {
+
+	record TestRecord {
+		union {SR1, SR2} unionField;
+	}
+
+	record SR1 {
+		string field;
+	}
+
+	record SR2 {
+		string field;
+	}
+
+}

lang/java/idl/src/test/idl/output/union.avpr

@@ -0,0 +1,38 @@
+{
+  "protocol": "UnionFwd",
+  "namespace": "org.apache.avro.gen",
+  "types": [
+    {
+      "type": "record",
+      "name": "TestRecord",
+      "fields": [
+        {
+          "name": "unionField",
+          "type": [
+            {
+              "type": "record",
+              "name": "SR1",
+              "fields": [
+                {
+                  "name": "field",
+                  "type": "string"
+                }
+              ]
+            },
+            {
+              "type": "record",
+              "name": "SR2",
+              "fields": [
+                {
+                  "name": "field",
+                  "type": "string"
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }
+  ],
+  "messages": {}
+}