Repository: ImFlog/schema-registry-plugin
Pull request #132 (2023-04-26 20:01:06)

reproduce bug regarding duplicate type definitions from local references
See Issue #130

ImFlog: I spent some time on this tonight and I have questions for you.
Following this [stackoverflow post](https://stackoverflow.com/questions/48100575/avro-multiple-record-of-same-type-in-single-schema) I think we can reference the schema with the name = "${namespace}.${typeRef}".

Based on the test case you provided, it would give something like this :

```json
{
      "name": "A",
      "namespace": "com.mycompany",
      "type": "record",
      "fields":
        [
          {
            "name": "nested",
            "type":
              {
                "name": "B",
                "namespace": "com.mycompany",
                "type": "enum",
                "symbols": ["X1", "X2"]
              }
          },
          {
            "name": "nested1",
            "type":
              {
                "name": "com.mycompany.B",
                "type": "enum"
              }
          }
        ]
    }
```

WDYT ?

I still have some challenge to tackle around the namespace retrieval (retrieve the namespace of the localReference and if not available use the global namespace). Plus the fact that it's a recursive code that is not that easy to understand 😅 


Also CC @dejank1986 as it would solve #126 too.
chriskn: The version 1.8.0 would not use the namespace for referencing type B. Because the B has the same Namespace then A. The schema generated with 1.8.0 would look like this:

```
{
  "name": "A",
  "namespace": "com.mycompany",
  "type": "record",
  "fields": [
    {
      "name": "nested",
      "type": {
        "name": "B",
        "type": "enum",
        "symbols": ["X1", "X2"]
      }
    },
    {
      "name": "nested1",
      "type": "B"
    }
  ]
}
```

Is it possible to compare the namespace of a reference with the namespace of the subject and if they are the same, skip the namespace when referencing? 


ImFlog: Thanks for the input, I understand better what It should look like.

If the namespace is not specified we can indeed not specify it, but in case it's defined in the above local reference I guess I need to specify It in the `type` field.

I will update the test you provided to reflect that, I still need to add a bit of manual parsing of the content to verify if the namespace is available or not but it's clear in my head (I think).

Will have a bit of time early next week I think.
ImFlog: Hello @chriskn, the tests are green and I managed the cases around namespace management as requested.
I am honestly not very fan of the code, it's a bit hard to follow (like most of the recursive ones I guess ...), but at least it works.
If you want to give it a spin locally or do a review that would be great, I want to merge this soon as you are not the only one impacted.
Thanks
chriskn: Hello @ImFlog, thank you for the fix! I will test it as soon as possible, latest tomorrow. 
chriskn: I am currently testing the new version and getting the following exception: 

```
Caused by: org.json.JSONException: JSONObject["name"] not found.
	at org.json.JSONObject.get(JSONObject.java:587)
	at org.json.JSONObject.getString(JSONObject.java:874)
	at com.github.imflog.schema.registry.parser.AvroSchemaParser.replaceType(AvroSchemaParser.kt:64)
	at com.github.imflog.schema.registry.parser.AvroSchemaParser.replaceType(AvroSchemaParser.kt:71)
	at com.github.imflog.schema.registry.parser.AvroSchemaParser.replaceType(AvroSchemaParser.kt:58)
	at com.github.imflog.schema.registry.parser.AvroSchemaParser.replaceType(AvroSchemaParser.kt:71)
	at com.github.imflog.schema.registry.parser.AvroSchemaParser.resolveLocalReferences(AvroSchemaParser.kt:42)
	at com.github.imflog.schema.registry.parser.SchemaParser.parseSchemaFromFile(SchemaParser.kt:39)
	at com.github.imflog.schema.registry.tasks.compatibility.CompatibilityTaskAction.run(CompatibilityTaskAction.kt:29)
	at com.github.imflog.schema.registry.tasks.compatibility.CompatibilityTask.testCompatibility(CompatibilityTask.kt:44)
```

I will look into it 
chriskn: @ImFlog, I ran into multiple issues with the current implementation 

Array type references are not supported
Types with additional, java specific information like  
```
     "items": {
        "type": "string",
        "avro.java.string": "String"
      }
```
are not supported. 

I will try to fix these issues 
chriskn: I invested some hours today and came to the conclusion that it does not make sense to reimplement the schema parsing. Instead, the `Schema.Parser` provided by `org.apache.avro` should be used. 

In the current state, this plugin is completely broken regarding schema parsing.
Therefore, I reverted the `AvroSchemaParser` to version 1.8.0.

One test is failing afterward. The reason is that the `Street` record is referenced by a local schema, but not included when parsing the schema. This issue did not occur in the latest version because of other issues with the schema parsing.

Since I don't see much benefit from referencing remote schemas (then can be downloaded before using the download task) I did not fix this issue. 

 It's up to you how to go further from here. 
ImFlog: @chriskn we are not reimplementing the Avro parsing, only replacing local references before doing the normal parsing. The goal is to be able to handle local + remote dependencies for Avro schemas.
There are some edge-cases but I will try to fix them (I will revert your last change back).

In the meantime you can stick to 1.8.0 and hopefully I will be able to fix this during the weekend.

Thank you for spending time on this PR 🙇‍♂️ 
ImFlog: @chriskn Can you explain a bit the issue with the custom java types ?
ImFlog: After thinking a bit about what you say and pocking things, I think that the easier is to fork the avro parser in order to silently ignore the dependencies it doesn't know.
I've done a draft locally and it seems to work for all the use cases. As the project is not moving very fast I should be able to keep up with the updates easily.

I just need to think about how I can setup the link between the two repo properly and without too much hassle.
ImFlog: It finally works @chriskn 🎉 
Can you make a sign if you get some time to test this new version on your real project ?
I have one week after the publication of a new release on the gradle plugin portal before it's boring to remove so Ideally I first get a sign from you and then I push the new version.
If Ok, I will delete the version and push this one without the SNAPSHOT suffix.
Thanks
ImFlog: Just published version `1.11.1`. 
Keep me posted 🙏
chriskn: Hi @ImFlog, since you hosted the dependency `avro-1.12.0-ImFlog` on GitHub (username and password required to retrieve the dependency), I can't build and publish a local release to test this.  
ImFlog: Hi @chriskn, I've spent the night fighting with Maven and jitpack to finally be able to make the branch compile without the need for custom authentication on github registry.

If you add the repository `maven("https://jitpack.io")` in your buildScript you should be able to use the new version (still on a temporary branch on my avro fork repository).

Not a big fan of how I had to make it work as it will be hard to keep up with avro updates with it but I am first trying to assess if everything works well and fixing the current issues before thinking about how I could improve it.

Happy to get feedback, I would like to put this behind me before my holidays :)
chriskn: When testing this branch, I am getting the following exception

```
Caused by: com.github.imflog.avro.SchemaParseException: Undefined name: "<MyReferencedSchema>"
	at com.github.imflog.avro.Schema.parse(Schema.java:1825)
	at com.github.imflog.avro.Schema.parse(Schema.java:1915)
	at com.github.imflog.avro.Schema.parse(Schema.java:1872)
	at com.github.imflog.avro.Schema$Parser.parse(Schema.java:1610)
	at com.github.imflog.avro.Schema$Parser.parse(Schema.java:1596)
	at com.github.imflog.schema.registry.parser.AvroSchemaParser.resolveLocalReferences(AvroSchemaParser.kt:27)
	at com.github.imflog.schema.registry.parser.SchemaParser.parseSchemaFromFile(SchemaParser.kt:39)
	at com.github.imflog.schema.registry.tasks.compatibility.CompatibilityTaskAction.run(CompatibilityTaskAction.kt:29)
	at com.github.imflog.schema.registry.tasks.compatibility.CompatibilityTask.testCompatibility(CompatibilityTask.kt:44)
```

Looks like referenced schemas are not correctly resolved when used in an array

chriskn: Hi @ImFlog, I added to failing test demonstrating the issue. It's not limited to arrays
ImFlog: Sorry for the late response @chriskn, this is finally working. While refactoring I forgot to parse the local dependency in "reverse" as stated in the documentation.
Both of your tests work correctly now.
If you have no other failing case, I will clean up my custom avro version and update the PR accordingly.
After that I will be able to merge this and finally move on to other things 😅 

Patch:
diff --git a/.github/workflows/main_build.yml b/.github/workflows/main_build.yml
index 2d7f4c3..02643d2 100644
--- a/.github/workflows/main_build.yml
+++ b/.github/workflows/main_build.yml
@@ -27,4 +27,4 @@ jobs:
       - name: "Build with Gradle"
         env:
           KAFKA_VERSION: ${{ matrix.kafka_version }}
-        run: "./gradlew integrationTest"
+        run: "./gradlew integrationTest"
\ No newline at end of file
diff --git a/.github/workflows/pr_build.yml b/.github/workflows/pr_build.yml
index 48d751d..d0281a1 100644
--- a/.github/workflows/pr_build.yml
+++ b/.github/workflows/pr_build.yml
@@ -15,4 +15,4 @@ jobs:
           java-version: "8"
           cache: "gradle"
       - name: "Build with Gradle"
-        run: "./gradlew integrationTest --stacktrace"
+        run: "./gradlew integrationTest --stacktrace"
\ No newline at end of file
diff --git a/README.md b/README.md
index 6cee4fe..736577a 100644
--- a/README.md
+++ b/README.md
@@ -2,11 +2,15 @@
 ![Github Actions](https://github.com/ImFlog/schema-registry-plugin/workflows/Master/badge.svg)
 
 # Schema-registry-plugin
-The aim of this plugin is to adapt the [Confluent schema registry maven plugin](https://docs.confluent.io/current/schema-registry/docs/maven-plugin.html) for Gradle builds.
+
+The aim of this plugin is to adapt
+the [Confluent schema registry maven plugin](https://docs.confluent.io/current/schema-registry/docs/maven-plugin.html)
+for Gradle builds.
 
 ## Usage
 
-As the plugin relies on packages developed by confluent you need to add the `https://packages.confluent.io/maven/` repository
+As the plugin relies on packages developed by confluent you need to add the `https://packages.confluent.io/maven/`
+repository
 to your `buildscript`:
 
 <div class="tabbed-code-block">
@@ -23,7 +27,7 @@ buildscript {
     }
 }
 plugins {
-  id "com.github.imflog.kafka-schema-registry-gradle-plugin" version "X.X.X"
+    id "com.github.imflog.kafka-schema-registry-gradle-plugin" version "X.X.X"
 }
 ```
 
@@ -33,25 +37,29 @@ plugins {
 
 ```kotlin
 buildscript {
-  repositories {
-    gradlePluginPortal()
-    maven {
-        url = uri("https://packages.confluent.io/maven/")
+    repositories {
+        gradlePluginPortal()
+        maven {
+            url = uri("https://packages.confluent.io/maven/")
+        }
     }
-  }
 }
 plugins {
-  id("com.github.imflog.kafka-schema-registry-gradle-plugin") version "X.X.X"
+    id("com.github.imflog.kafka-schema-registry-gradle-plugin") version "X.X.X"
 }
 ```
 
   </details>
 </div>
 
-Where "X.X.X" is the current version, see [gradle plugin portal](https://plugins.gradle.org/plugin/com.github.imflog.kafka-schema-registry-gradle-plugin) for details.
+Where "X.X.X" is the current version,
+see [gradle plugin portal](https://plugins.gradle.org/plugin/com.github.imflog.kafka-schema-registry-gradle-plugin) for
+details.
 
 ## Tasks
+
 When you install the plugin, four tasks are added under `registry` group:
+
 * downloadSchemasTask
 * testSchemasTask
 * registerSchemasTask
@@ -60,6 +68,7 @@ When you install the plugin, four tasks are added under `registry` group:
 What these tasks do and how to configure them is described in the following sections.
 
 ### Global configuration
+
 ```groovy
 schemaRegistry {
     url = 'http://registry-url:8081/'
@@ -67,6 +76,7 @@ schemaRegistry {
     outputDirectory = "/home/kafka/results"
 }
 ```
+
 * `url` is where the script can reach the Schema Registry.
 * `quiet` is whether you want to disable "INFO" level logs.
   This can be useful if you test the compatibility of a lot of schema.
@@ -75,9 +85,11 @@ schemaRegistry {
   This is an optional parameter.
 
 ### Download schemas
+
 Like the name of the task imply, this task is responsible for retrieving schemas from a schema registry.
 
 A DSL is available to configure the task:
+
 ```groovy
 
 // Optional
@@ -88,18 +100,20 @@ schemaRegistry {
     download {
         // Optional
         metadata = new MetadataExtension(true, "path/to/metadata/")
-      
+
         // extension of the output file depends on the the schema type
         subject('avroSubject', '/absolutPath/src/main/avro')
         subject('protoSubject', 'src/main/proto')
         subject('jsonSubject', 'src/main/json')
-        
+
         // You can use a regex to download multiple schemas at once
         subjectPattern('avro.*', 'src/main/avro')
     }
 }
 ```
+
 Here is the list of all the signatures for the `subject` extension:
+
 * `subject(inputSubject: String, outputPath: String)`
 * `subject(inputSubject: String, outputPath: String, outputFileName: String)`
 * `subject(inputSubject: String, outputPath: String, version: Int)`
@@ -108,17 +122,20 @@ Here is the list of all the signatures for the `subject` extension:
 
 You can configure the metadata extension in order to download the schemas metadata in json files.
 It will be saved in files named like the schema file but suffixed by `-metadata.json` in the outputPath you specify
-and defaults to the same output directory as your schemas.  
+and defaults to the same output directory as your schemas.
 
 NB:
+
 * If not provided, the outputFileName is equal to the inputSubject.
 * It's not possible to specify the outputFileName for subject pattern as it would override the
   file for each downloaded schema.
 
 ### Test schemas compatibility
+
 This task test compatibility between local schemas and schemas stored in the Schema Registry.
 
 A DSL is available to specify what to test:
+
 ```groovy
 schemaRegistry {
     url = 'http://registry-url:8081'
@@ -134,7 +151,8 @@ schemaRegistry {
     }
 }
 ```
-You have to list all the (subject, avsc file path) pairs that you want to test. 
+
+You have to list all the (subject, avsc file path) pairs that you want to test.
 
 If you have references with other schemas stored in the registry that are required before the compatibility check,
 you can call the `addReference("name", "subject", version)`,
@@ -150,6 +168,7 @@ this will add a reference from a local file and inline it in the schema registry
 The addLocalReference calls can be chained.
 
 Notes:
+
 * If you want to reuse Subjects with the register task you can define a `Subject` object like so:
   ```groovy
   import com.github.imflog.schema.registry.Subject
@@ -167,23 +186,29 @@ Notes:
   ```
 
 #### Avro
+
 Mixing local and remote references is perfectly fine for Avro without specific configurations.
 
 #### Json
+
 Mixing local and remote references is perfectly fine for JSON.
 
-If you need to add reference to local schema to a JSON schema, make sure that the local reference contains a `$id` attribute.
+If you need to add reference to local schema to a JSON schema, make sure that the local reference contains a `$id`
+attribute.
 This id is the value that need to be put on the `$ref` part.
 For more concrete example, take a look at the [json example](examples/json/build.gradle).
 
 #### Protobuf
+
 :warning: Local references is not yet supported for PROTOBUF.
 
 ### Register schemas
+
 Once again the name speaks for itself.
 This task register schemas from a local path to a Schema Registry.
 
 A DSL is available to specify what to register:
+
 ```groovy
 schemaRegistry {
     url = 'http://registry-url:8081'
@@ -199,6 +224,7 @@ schemaRegistry {
     }
 }
 ```
+
 If you have references to other schemas required before the register,
 you can call the `addReference("name", "subject", version)`,
 this will add a reference to use from the registry.
@@ -213,8 +239,9 @@ this will add a reference from a local file and inline it in the schema registry
 The addLocalReference calls can be chained.
 
 Notes:
-* A registered.csv file will be created with the following format `subject, path, id` 
-if you need information about the registered id.
+
+* A registered.csv file will be created with the following format `subject, path, id`
+  if you need information about the registered id.
 * If you want to reuse Subjects with the compatibility task you can define a `Subject` object like so:
   ```groovy
   import com.github.imflog.schema.registry.Subject
@@ -232,22 +259,28 @@ if you need information about the registered id.
   ```
 
 #### Avro
+
 Mixing local and remote references is perfectly fine for Avro without specific configurations.
 
 #### Json
+
 Mixing local and remote references is perfectly fine for JSON.
 
-If you need to add reference to local schema to a JSON schema, make sure that the local reference contains a `$id` attribute.
+If you need to add reference to local schema to a JSON schema, make sure that the local reference contains a `$id`
+attribute.
 This id is the value that need to be put on the `$ref` part.
 For more concrete example, take a look at the [json example](examples/json/build.gradle).
 
 #### Protobuf
+
 :warning: Local references is not yet supported for PROTOBUF.
 
 ### Configure subjects
+
 This task sets the schema compatibility level for registered subjects.
 
 A DSL is available to specify which subjects to configure:
+
 ```groovy
 schemaRegistry {
     url = 'http://registry-url:8081'
@@ -257,18 +290,22 @@ schemaRegistry {
     }
 }
 ```
+
 See the Confluent
 [Schema Registry documentation](https://docs.confluent.io/current/schema-registry/avro.html#compatibility-types)
 for more information on valid compatibility levels.
 
-You have to list the (subject, compatibility-level) 
+You have to list the (subject, compatibility-level)
 
 ## Security
+
 According to how your schema registry instance security configuration,
 you can configure the plugin to access it securely.
 
 ### Basic authentication
+
 An extension allow you to specify the basic authentication like so:
+
 ```groovy
 schemaRegistry {
     url = 'http://registry-url:8081'
@@ -280,30 +317,38 @@ schemaRegistry {
 ```
 
 ### Encryption (SSL)
+
 If you want to encrypt the traffic in transit (using SSL), use the following extension:
+
 ```groovy
 schemaRegistry {
     url = 'https://registry-url:8081'
     ssl {
         configs = [
-            "ssl.truststore.location": "/path/to/registry.truststore.jks",
-            "ssl.truststore.password": "truststorePassword",
-            "ssl.keystore.location": "/path/to/registry.keystore.jks",
-            "ssl.keystore.password": "keystorePassword"
+                "ssl.truststore.location": "/path/to/registry.truststore.jks",
+                "ssl.truststore.password": "truststorePassword",
+                "ssl.keystore.location"  : "/path/to/registry.keystore.jks",
+                "ssl.keystore.password"  : "keystorePassword"
         ]
     }
 }
 ```
-Valid key values are listed here: [org.apache.kafka.common.config.SslConfigs](https://github.com/confluentinc/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java)
+
+Valid key values are listed
+here: [org.apache.kafka.common.config.SslConfigs](https://github.com/confluentinc/kafka/blob/master/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java)
 
 ### Examples
+
 Detailed examples can be found in the [examples directory](examples).
 
 ## Version compatibility
+
 When using the plugin, a default version of the confluent Schema registry is use.
 The 5.5.X version of the schema-registry introduced changes that made the older version of the plugin obsolete.
 
-It was easier to introduce all the changes in one shot instead of supporting both version. Here is what it implies for users:
+It was easier to introduce all the changes in one shot instead of supporting both version. Here is what it implies for
+users:
+
 * plugin versions above 1.X.X support the confluent version > 5.5.X (Avro / Json / Protobuf)
 * plugin versions should support anything below 5.4.X
 
@@ -313,28 +358,34 @@ take a look at [examples/override-confluent-version](examples/override-confluent
 ## Developing
 
 ### Running tests
+
 In order to customize the Kafka version to run in integration tests,
 you can specify the ENV VAR KAFKA_VERSION with the version that you want to test upon.
 The library is tested with the following versions:
+
 * 6.2.6
 * 7.2.2
 * 7.3.1 (by default if no env_var is passed)
 
 PS: If you are running an ARM computer (like apple M1),
 you can add the `.arm64` suffix to the version to run ARM container and speed up tests.
+
 ```bash
 KAFKA_VERSION=7.2.0.arm64 ./gradlew integrationTest
 ````
 
 ### Publishing locally
+
 In order to build the plugin locally, you can run the following commands:
+
 ```bash
 ./gradlew build # To compile and test the code
 ./gradlew publishToMavenLocal # To push the plugin to your mavenLocal
 ```
 
-Once the plugin is pushed into your mavenLocal, you can use it by 
+Once the plugin is pushed into your mavenLocal, you can use it by
 adding the `mavenLocal` to the buildscript repositories like so:
+
 ```groovy
 buildscript {
     repositories {
@@ -343,7 +394,7 @@ buildscript {
         maven {
             url "http://packages.confluent.io/maven/"
         }
-  }
+    }
     dependencies {
         classpath "com.github.imflog:kafka-schema-registry-gradle-plugin:X.X.X-SNAPSHOT"
     }
diff --git a/build.gradle.kts b/build.gradle.kts
index cb0f37f..523921a 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -14,11 +14,13 @@ plugins {
 repositories {
     mavenCentral()
     maven("https://packages.confluent.io/maven/")
+    maven("https://jitpack.io")
 }
 
 // Dependencies versions
 val kotlinVersion = "1.8.20"
 val confluentVersion = "7.3.3"
+val avroVersion = "1.11.2"
 dependencies {
     implementation(gradleApi())
     implementation(kotlin("stdlib", kotlinVersion))
@@ -29,6 +31,8 @@ dependencies {
     }
     implementation("io.confluent", "kafka-json-schema-provider")
     implementation("io.confluent", "kafka-protobuf-provider")
+    // Our custom avro version. See https://github.com/ImFlog/avro
+    implementation("com.github.ImFlog.avro", "avro", avroVersion)
 }
 
 tasks.withType<KotlinCompile>().configureEach {
diff --git a/examples/gradle/wrapper/gradle-wrapper.properties b/examples/gradle/wrapper/gradle-wrapper.properties
index ffed3a2..070cb70 100644
--- a/examples/gradle/wrapper/gradle-wrapper.properties
+++ b/examples/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.6-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParser.kt b/src/main/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParser.kt
index 0731a5b..078e20f 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParser.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParser.kt
@@ -3,18 +3,9 @@ package com.github.imflog.schema.registry.parser
 import com.github.imflog.schema.registry.LocalReference
 import com.github.imflog.schema.registry.SchemaType
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
-import org.json.JSONArray
-import org.json.JSONObject
+import com.github.imflog.avro.Schema.Parser
 import java.io.File
 
-private const val TYPE = "type"
-
-private const val ITEMS = "items"
-
-private const val VALUES = "values"
-
-private const val FIELDS = "fields"
-
 class AvroSchemaParser(
     client: SchemaRegistryClient,
     rootDir: File
@@ -22,33 +13,19 @@ class AvroSchemaParser(
 
     override val schemaType: SchemaType = SchemaType.AVRO
 
-    private val keysToUpdate: Collection<String> = listOf(TYPE, ITEMS, VALUES, FIELDS)
-
     override fun resolveLocalReferences(
         subject: String,
         schemaContent: String,
         localReferences: List<LocalReference>
     ): String {
-        return replaceType(JSONObject(schemaContent), localReferences).toString()
-    }
-
-    private fun replaceType(type: Any, localReferences: List<LocalReference>): Any {
-        return when (type) {
-            is String -> localReferences
-                .filter { it.name == type }
-                .map { JSONObject(it.content(rootDir)) }
-                .map { replaceType(it, localReferences) }
-                .firstOrNull() ?: type
-
-            is JSONArray -> JSONArray(type.map { replaceType(it, localReferences) })
-            is JSONObject -> {
-                type.keySet()
-                    .filter { keysToUpdate.contains(it) }
-                    .forEach { type.put(it, replaceType(type.get(it), localReferences)) }
-                return type
-            }
-
-            else -> type
-        }
+        val parser = Parser()
+            .setValidate(false)
+            .setValidateDefaults(false)
+            .setValidateUnknownTypes(false)
+        localReferences
+            .reversed()
+            .map { reference -> reference.content(rootDir) }
+            .forEach { parser.parse(it) }
+        return parser.parse(schemaContent).toString()
     }
 }
\ No newline at end of file


Test patch:
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParserTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParserTest.kt
index 664abb4..aea5cb5 100644
--- a/src/test/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParserTest.kt
+++ b/src/test/kotlin/com/github/imflog/schema/registry/parser/AvroSchemaParserTest.kt
@@ -3,16 +3,22 @@ package com.github.imflog.schema.registry.parser
 import com.github.imflog.schema.registry.LocalReference
 import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient
 import org.assertj.core.api.Assertions
+import org.intellij.lang.annotations.Language
 import org.json.JSONObject
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
+import org.junit.jupiter.api.assertDoesNotThrow
 import org.junit.jupiter.api.io.TempDir
+import java.io.File
 import java.nio.file.Path
+import java.nio.file.Paths
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class AvroSchemaParserTest {
 
     private val schemaRegistryClient = MockSchemaRegistryClient()
+    private val testFilesPath = "${Paths.get("").toAbsolutePath()}/src/test/resources/"
+
     @TempDir
     lateinit var folderRule: Path
 
@@ -34,6 +40,7 @@ class AvroSchemaParserTest {
                 {"name": "address", "type": "$ADDRESS_REFERENCE_NAME"}
              ]
         }"""
+
     }
 
     @Test
@@ -70,4 +77,170 @@ class AvroSchemaParserTest {
             .usingRecursiveComparison() // To check the content of the JSON Objects instead of the equal method
             .isEqualTo(JSONObject(ADDRESS_SCHEMA))
     }
+
+    @Test
+    fun `Should resolve duplicated references with same namespace correctly`() {
+        // Given
+        val parser = AvroSchemaParser(schemaRegistryClient, File(testFilesPath))
+        val reference = LocalReference("B", "${testFilesPath}testType.avsc")
+        val schema = File("${testFilesPath}testSubject.avsc").readText()
+        // When
+        val resolvedSchema = parser.resolveLocalReferences(
+            "test",
+            schema,
+            listOf(reference)
+        )
+        // Then
+        val resolved = JSONObject(resolvedSchema).toString()
+
+        @Language("JSON")
+        val expected = """{
+          "name": "A",
+          "namespace": "com.mycompany",
+          "type": "record",
+          "fields": [
+            {
+              "name": "nested",
+              "type": {
+                "name":"B",
+                "type": "enum",
+                "symbols": ["X1", "X2"]
+              }
+            },
+            {
+              "name": "nested1",
+              "type": "B"
+            }
+          ]
+        }"""
+
+        Assertions.assertThat(resolved).isEqualTo(
+            JSONObject(expected).toString()
+        )
+    }
+
+    @Test
+    fun `Should resolve complex nested array example correctly`() {
+        // Given
+        val parser = AvroSchemaParser(schemaRegistryClient, File(testFilesPath))
+
+        val schema = File("${testFilesPath}ParentArraySubject.avsc")
+            .readText()
+
+        assertDoesNotThrow {
+            parser.resolveLocalReferences(
+                "test",
+                schema,
+                listOf(
+                    LocalReference("NestedArrayType", "${testFilesPath}NestedArrayType.avsc"),
+                    LocalReference("NestedNestedType", "${testFilesPath}NestedNestedType.avsc"),
+                )
+            )
+        }
+    }
+
+    @Test
+    fun `Should resolve complex nested record example correctly`() {
+        // Given
+        val parser = AvroSchemaParser(schemaRegistryClient, File(testFilesPath))
+
+        val schema = File("${testFilesPath}ParentSubject.avsc")
+            .readText()
+
+        assertDoesNotThrow {
+            parser.resolveLocalReferences(
+                "test",
+                schema,
+                listOf(
+                    LocalReference("NestedType", "${testFilesPath}NestedType.avsc"),
+                    LocalReference("NestedNestedType", "${testFilesPath}NestedNestedType.avsc"),
+                )
+            )
+        }
+    }
+
+    @Test
+    fun `Should resolve duplicated array references correctly`() {
+        // Given
+        val parser = AvroSchemaParser(schemaRegistryClient, File(testFilesPath))
+        val reference = LocalReference("B", "${testFilesPath}testType.avsc")
+        val schema = File("${testFilesPath}testSubjectWithArrayReference.avsc")
+            .readText()
+        // When
+        val resolvedSchema = parser.resolveLocalReferences(
+            "test",
+            schema,
+            listOf(reference)
+        )
+        // Then
+        val resolved = JSONObject(resolvedSchema).toString()
+
+        @Language("JSON")
+        val expected = """{
+           "name": "A",
+           "namespace": "com.mycompany",
+           "type": "record",
+           "fields": [
+             {
+               "name": "nested",
+               "type": {
+                "type": "array",
+                "items": {
+                        "name":"B",
+                        "type":"enum",
+                        "symbols": ["X1","X2"]
+                    }
+                }
+            },
+            {
+                "name":"nested1",
+                "type":{
+                    "type":"array",
+                    "items":"B"
+                }
+            }
+          ]
+        }"""
+
+        Assertions.assertThat(resolved).isEqualTo(
+            JSONObject(expected).toString()
+        )
+    }
+
+    @Test
+    fun `Should parse with unknown remote references correctly`() {
+        // Given
+        val parser = AvroSchemaParser(schemaRegistryClient, File(testFilesPath))
+        val schema = File("${testFilesPath}testSubject.avsc")
+            .readText()
+        // When
+        val resolvedSchema = parser.resolveLocalReferences(
+            "test",
+            schema,
+            listOf()
+        )
+        // Then
+        val resolved = JSONObject(resolvedSchema).toString()
+
+        @Language("JSON")
+        val expected = """{
+          "type": "record",
+          "name": "A",
+          "namespace": "com.mycompany",
+          "fields": [
+            {
+              "name": "nested",
+              "type": "B"
+            },
+            {
+              "name": "nested1",
+              "type": "B"
+            }
+          ]
+        }"""
+
+        Assertions.assertThat(resolved).isEqualTo(
+            JSONObject(expected).toString()
+        )
+    }
 }
\ No newline at end of file
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskActionTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskActionTest.kt
index ac3a1e0..fdafa24 100644
--- a/src/test/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskActionTest.kt
+++ b/src/test/kotlin/com/github/imflog/schema/registry/tasks/compatibility/CompatibilityTaskActionTest.kt
@@ -326,9 +326,10 @@ class CompatibilityTaskActionTest {
                 "src/main/avro/local/test.avsc",
                 "AVRO"
             )
-                .addReference("Street", "Street", 1)
-                .addLocalReference("Country", "src/main/avro/local/country.avsc")
                 .addLocalReference("Address", "src/main/avro/local/address.avsc")
+                .addLocalReference("Country", "src/main/avro/local/country.avsc")
+                .addReference("Street", "Street", 1)
+
         )
 
         // when
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt
index 762e631..9d7f060 100644
--- a/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt
+++ b/src/test/kotlin/com/github/imflog/schema/registry/tasks/register/RegisterTaskActionTest.kt
@@ -14,6 +14,7 @@ import org.junit.jupiter.api.io.TempDir
 import java.io.File
 import java.nio.file.Files
 import java.nio.file.Path
+import java.nio.file.Paths
 
 class RegisterTaskActionTest {
     @TempDir
@@ -325,4 +326,28 @@ class RegisterTaskActionTest {
                |""".trimMargin()
             )
     }
+
+    @Test
+    fun `Should register schema with duplicated local references`() {
+        // Given
+        val registryClient =
+            MockSchemaRegistryClient(listOf(AvroSchemaProvider(), JsonSchemaProvider(), ProtobufSchemaProvider()))
+        val projectDirAbsolutePath = Paths.get("").toAbsolutePath().toString()
+        val schemaPath = "$projectDirAbsolutePath/src/test/resources/"
+        val subjects = listOf(
+            Subject("test", "${schemaPath}testSubject.avsc", "AVRO")
+                .addLocalReference("B", "${schemaPath}testType.avsc")
+        )
+
+        // When
+        val errorCount = RegisterTaskAction(
+            registryClient,
+            folderRule.root.toFile(),
+            subjects,
+            null
+        ).run()
+
+        // Then
+        Assertions.assertThat(errorCount).isEqualTo(0)
+    }
 }
diff --git a/src/test/resources/NestedArrayType.avsc b/src/test/resources/NestedArrayType.avsc
new file mode 100644
index 0000000..0eb671d
--- /dev/null
+++ b/src/test/resources/NestedArrayType.avsc
@@ -0,0 +1,14 @@
+{
+  "type": "record",
+  "name": "NestedArrayType",
+  "namespace": "com.test.company",
+  "fields": [
+    {
+      "name": "someNestedArray",
+      "type": {
+        "type": "array",
+        "items": "NestedNestedType"
+      }
+    }
+  ]
+}
diff --git a/src/test/resources/NestedNestedType.avsc b/src/test/resources/NestedNestedType.avsc
new file mode 100644
index 0000000..63888b3
--- /dev/null
+++ b/src/test/resources/NestedNestedType.avsc
@@ -0,0 +1,18 @@
+{
+  "type": "record",
+  "name": "NestedNestedType",
+  "namespace": "com.test.company",
+  "fields": [
+    {
+      "name": "name",
+      "type": {
+        "type": "string",
+        "avro.java.string": "String"
+      }
+    },
+    {
+      "name": "number",
+      "type": "int"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/NestedType.avsc b/src/test/resources/NestedType.avsc
new file mode 100644
index 0000000..634c5c2
--- /dev/null
+++ b/src/test/resources/NestedType.avsc
@@ -0,0 +1,11 @@
+{
+  "type": "record",
+  "name": "NestedType",
+  "namespace": "com.test.company",
+  "fields": [
+    {
+      "name": "someNestedRecord",
+      "type": "NestedNestedType"
+    }
+  ]
+}
diff --git a/src/test/resources/ParentArraySubject.avsc b/src/test/resources/ParentArraySubject.avsc
new file mode 100644
index 0000000..55d70d4
--- /dev/null
+++ b/src/test/resources/ParentArraySubject.avsc
@@ -0,0 +1,11 @@
+{
+  "type": "record",
+  "name": "ParentSubjectArray",
+  "namespace": "com.test.company",
+  "fields": [
+    {
+      "name": "element",
+      "type": "NestedArrayType"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/ParentSubject.avsc b/src/test/resources/ParentSubject.avsc
new file mode 100644
index 0000000..cd219e0
--- /dev/null
+++ b/src/test/resources/ParentSubject.avsc
@@ -0,0 +1,11 @@
+{
+  "type": "record",
+  "name": "ParentSubject",
+  "namespace": "com.test.company",
+  "fields": [
+    {
+      "name": "element",
+      "type": "NestedType"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/testSubject.avsc b/src/test/resources/testSubject.avsc
new file mode 100644
index 0000000..0551b62
--- /dev/null
+++ b/src/test/resources/testSubject.avsc
@@ -0,0 +1,15 @@
+{
+  "type": "record",
+  "name": "A",
+  "namespace": "com.mycompany",
+  "fields": [
+    {
+      "name": "nested",
+      "type": "B"
+    },
+    {
+      "name": "nested1",
+      "type": "B"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/testSubjectComplex.avsc b/src/test/resources/testSubjectComplex.avsc
new file mode 100644
index 0000000..3661023
--- /dev/null
+++ b/src/test/resources/testSubjectComplex.avsc
@@ -0,0 +1,14 @@
+{
+  "type": "record",
+  "name": "A",
+  "namespace": "com.mycompany",
+  "fields": [
+    {
+      "name": "element",
+      "type": [
+        "B",
+        "C"
+      ]
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/testSubjectWithArrayReference.avsc b/src/test/resources/testSubjectWithArrayReference.avsc
new file mode 100644
index 0000000..cc2447b
--- /dev/null
+++ b/src/test/resources/testSubjectWithArrayReference.avsc
@@ -0,0 +1,21 @@
+{
+  "type": "record",
+  "name": "A",
+  "namespace": "com.mycompany",
+  "fields": [
+    {
+      "name": "nested",
+      "type": {
+        "type": "array",
+        "items": "com.mycompany.B"
+      }
+    },
+    {
+      "name": "nested1",
+      "type": {
+        "type": "array",
+        "items": "com.mycompany.B"
+      }
+    }
+  ]
+}
\ No newline at end of file
diff --git a/src/test/resources/testType.avsc b/src/test/resources/testType.avsc
new file mode 100644
index 0000000..81c4082
--- /dev/null
+++ b/src/test/resources/testType.avsc
@@ -0,0 +1,6 @@
+{
+  "type": "enum",
+  "name": "B",
+  "namespace": "com.mycompany",
+  "symbols" : ["X1", "X2"]
+}
\ No newline at end of file