Fix some cases not handled by schema name patching (#541)

Co-authored-by: Karthik Ramgopal <kramgopa@linkedin.com>

parser/src/main/java/com/linkedin/avroutil1/writer/avsc/AvscSchemaWriter.java

@@ -68,8 +68,9 @@ protected Path pathForSchema(AvroSchema maybeNamed) {
     }
     AvroNamedSchema namedSchema = (AvroNamedSchema) maybeNamed;
     AvroName name = namedSchema.getName();
+    String namespace = getNamespace(name);
 
-    if (!name.hasNamespace()) {
+    if (namespace == null || namespace.isEmpty()) {
       return Paths.get(getSimpleName(name) + "." + AvscFile.SUFFIX);
     }
 
@@ -226,17 +227,24 @@ protected void writeNamedSchema(AvroNamedSchema schema, AvscWriterContext contex
    */
   protected void writeSchemaRef(AvroNamedSchema schema, AvscWriterContext context, AvscWriterConfig config,
       JsonGenerator generator) throws IOException {
+    AvroName schemaName = schema.getName();
 
     // Emit fullname always if configured to do so.
     if (config.isAlwaysEmitNamespace()) {
-      generator.writeString(getFullName(schema.getName(), false));
+      generator.writeString(getFullName(schemaName, false));
+      return;
     }
 
-    // Figure out what the context namespace is
+    // Figure out what the context namespace is. If it is the same as the namespace, then write the simple name, else
+    // write the full name.
     String contextNamespace =
         config.isUsePreAvro702Logic() ? context.getAvro702ContextNamespace() : context.getCorrectContextNamespace();
-    String qualified = schema.getName().qualified(contextNamespace);
-    generator.writeString(qualified);
+    String namespace = getNamespace(schemaName);
+    if (namespace.equals(contextNamespace)) {
+      generator.writeString(getSimpleName(schemaName));
+    } else {
+      generator.writeString(getFullName(schemaName, false));
+    }
   }
 
   protected AvroName emitSchemaName(AvroNamedSchema schema, AvscWriterContext context, AvscWriterConfig config,
@@ -295,7 +303,7 @@ protected AvroName emitSchemaName(AvroNamedSchema schema, AvscWriterContext cont
       generator.writeStringField("name", getSimpleName(schemaName));
     }
 
-    context.pushNamingContext(schema, contextNamespaceAfter, contextNamespaceAfter702);
+    context.pushNamingContext(contextNamespaceAfter, contextNamespaceAfter702);
 
     return extraAlias;
   }

parser/src/main/java/com/linkedin/avroutil1/writer/avsc/AvscWriterContext.java

@@ -6,7 +6,6 @@
 
 package com.linkedin.avroutil1.writer.avsc;
 
-import com.linkedin.avroutil1.model.AvroName;
 import com.linkedin.avroutil1.model.AvroNamedSchema;
 import com.linkedin.avroutil1.model.AvroSchema;
 import java.util.ArrayDeque;
@@ -33,16 +32,16 @@ public class AvscWriterContext {
   private final Map<String, AvroNamedSchema> known = new HashMap<>(); //by their fullname
 
   public AvscWriterContext() {
-    //context starts at the "root"  namespace
-    contextStack.push(new NamingContext(null, "", ""));
+    // Context starts at the "root" namespace ie. empty.
+    contextStack.push(new NamingContext("", ""));
   }
 
   /**
    * @return the correct namespace at this current inner-most context
    */
   public String getCorrectContextNamespace() {
     assert contextStack.peek() != null;
-    return contextStack.peek().correcttNamespace;
+    return contextStack.peek().correctNamespace;
   }
 
   /**
@@ -72,32 +71,27 @@ public boolean schemaEncountered(AvroNamedSchema schema) {
     return true;
   }
 
-  public void pushNamingContext(AvroNamedSchema root, String correctNamespace, String pre702ParsedNamespace) {
-    contextStack.push(new NamingContext(root.getName(), correctNamespace, pre702ParsedNamespace));
+  public void pushNamingContext(String correctNamespace, String pre702ParsedNamespace) {
+    contextStack.push(new NamingContext(correctNamespace, pre702ParsedNamespace));
   }
 
   public void popNamingContext() {
     contextStack.pop(); //will throw if empty
   }
 
   private static class NamingContext {
-    /**
-     * the (always correct) named type at the root of this context (if any)
-     */
-    private final AvroName root;
     /**
      * the correct namespace of this context, as defined by the avro specification
      */
-    private final String correcttNamespace;
+    private final String correctNamespace;
     /**
      * what would be the context namespace here on parsing if the schema was
      * written under pre-702 (incorrect) logic
      */
     private final String pre702ParsedNamespace;
 
-    public NamingContext(AvroName root, String correctNamespace, String pre702ParsedNamespace) {
-      this.root = root;
-      this.correcttNamespace = correctNamespace;
+    public NamingContext(String correctNamespace, String pre702ParsedNamespace) {
+      this.correctNamespace = correctNamespace;
       this.pre702ParsedNamespace = pre702ParsedNamespace;
     }
   }