Repository: hamba/avro
Pull request #502 (2025-02-10 12:57:38)

fix: panic with concurrent schema parsing
Issue:  https://github.com/hamba/avro/issues/481

This pull request introduces a new test for concurrent schema parsing, adds a method to merge schema caches, and updates the schema parsing logic to utilize an internal cache. The most important changes are as follows:

### Testing Enhancements:

* [`concurrent_parse_test.go`](diffhunk://#diff-16c19ec3114bc919dbb65c16e0e0ed30fe915f69d0c8148bb5a999ea7384c295R1-R148): Added a new test `TestConcurrentParse` to validate concurrent schema parsing by running multiple goroutines that parse a test schema. This ensures the thread-safety and correctness of the parsing function.

### Schema Cache Enhancements:

* [`schema.go`](diffhunk://#diff-c3b0d951288715c695f51ce2727c87a056e89ccbb4f3003be71baa8b405383e9R134-R144): Added a new method `AddAll` to the `SchemaCache` struct, which allows merging all schemas from one cache into another. This is useful for combining schema caches without duplicating entries.

### Schema Parsing Logic Enhancements:

* [`schema_parse.go`](diffhunk://#diff-778f05933a6d581331989dcf1793cfca10bd80b5e0e3693ce8c146b26497c8efR77-R87): Updated the `ParseBytesWithCache` function to use an internal `SchemaCache` for intermediate parsing steps. This change helps in isolating the parsing process and then merging the results back into the original cache, improving the modularity and maintainability of the code.


Patch:
diff --git a/schema.go b/schema.go
index 7cba4c8..84a9fd2 100644
--- a/schema.go
+++ b/schema.go
@@ -131,6 +131,17 @@ func (c *SchemaCache) Get(name string) Schema {
 	return nil
 }
 
+// AddAll adds all schemas from the given cache to the current cache.
+func (c *SchemaCache) AddAll(cache *SchemaCache) {
+	if cache == nil {
+		return
+	}
+	cache.cache.Range(func(key, value interface{}) bool {
+		c.cache.Store(key, value)
+		return true
+	})
+}
+
 // Schemas is a slice of Schemas.
 type Schemas []Schema
 
diff --git a/schema_parse.go b/schema_parse.go
index 6903c02..630f202 100644
--- a/schema_parse.go
+++ b/schema_parse.go
@@ -74,11 +74,17 @@ func ParseBytesWithCache(schema []byte, namespace string, cache *SchemaCache) (S
 		json = string(schema)
 	}
 
+	internalCache := &SchemaCache{}
+	internalCache.AddAll(cache)
+
 	seen := seenCache{}
-	s, err := parseType(namespace, json, seen, cache)
+	s, err := parseType(namespace, json, seen, internalCache)
 	if err != nil {
 		return nil, err
 	}
+
+	cache.AddAll(internalCache)
+
 	return derefSchema(s), nil
 }
 


Test patch:
diff --git a/schema_test.go b/schema_test.go
index c5f70a9..a46c314 100644
--- a/schema_test.go
+++ b/schema_test.go
@@ -3,6 +3,7 @@ package avro_test
 import (
 	"encoding/json"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/hamba/avro/v2"
@@ -2244,3 +2245,18 @@ func TestNewSchema_IgnoresInvalidProperties(t *testing.T) {
 		}, rec.Props())
 	})
 }
+
+func TestConcurrentParse(t *testing.T) {
+	var wg sync.WaitGroup
+
+	for i := 0; i < 10000; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, err := avro.ParseFiles("testdata/concurrent-schema.avsc")
+			require.NoError(t, err)
+		}()
+	}
+
+	wg.Wait()
+}
diff --git a/testdata/concurrent-schema.avsc b/testdata/concurrent-schema.avsc
new file mode 100644
index 0000000..1172b1e
--- /dev/null
+++ b/testdata/concurrent-schema.avsc
@@ -0,0 +1,121 @@
+{
+  "type": "record",
+  "name": "FootballUpdateEvent",
+  "namespace": "com.example.avro",
+  "fields": [
+    {
+      "name": "event_metadata",
+      "type": {
+        "type": "record",
+        "name": "EventMetadata",
+        "fields": [
+          { "name": "name", "type": "string", "default": "" },
+          { "name": "trace", "type": "string", "default": "" },
+          { "name": "stamp", "type": "long", "default": 0 },
+          { "name": "destination", "type": "string", "default": "" }
+        ]
+      },
+      "default": {
+        "name": "",
+        "trace": "",
+        "stamp": 0,
+        "destination": ""
+      }
+    },
+    {
+      "name": "player",
+      "type": {
+        "type": "record",
+        "name": "Player",
+        "fields": [
+          { "name": "team_id", "type": "string", "default": "" },
+          { "name": "name", "type": "string", "default": "" },
+          { "name": "team", "type": "string", "default": "" },
+          { "name": "contract", "type": "long", "default": 0 },
+          { "name": "xg", "type": "double", "default": 0.0 },
+          { "name": "xgp", "type": "double", "default": 0.0 },
+          { "name": "xgp99", "type": "double", "default": 0.0 },
+          { "name": "xg90", "type": "double", "default": 0.0 },
+          {
+            "name": "xg90p",
+            "type": {
+              "type": "record",
+              "name": "MatchXG",
+              "fields": [
+                { "name": "matchXG", "type": "double", "default": 0.0 },
+                { "name": "matchXGP", "type": "string", "default": "" }
+              ]
+            },
+            "default": {
+              "matchXG": 0.0,
+              "matchXGP": ""
+            }
+          },
+          {
+            "name": "ttd",
+            "type": "MatchXG",
+            "default": {
+              "matchXG": 0.0,
+              "matchXGP": ""
+            }
+          },
+          {
+            "name": "leagueXG",
+            "type": {
+              "type": "record",
+              "name": "LeagueXG",
+              "fields": [
+                { "name": "top_assist", "type": "string", "default": "" },
+                { "name": "top_score", "type": "string", "default": "" },
+                { "name": "top_xg", "type": "string", "default": "" },
+                { "name": "top_creation", "type": "string", "default": "" }
+              ]
+            },
+            "default": {
+              "top_assist": "",
+              "top_score": "",
+              "top_xg": "",
+              "top_creation": ""
+            }
+          },
+          {
+            "name": "player_numbers",
+            "type": {
+              "type": "array",
+              "items": {
+                "type": "record",
+                "name": "PlayerNumber",
+                "fields": [
+                  { "name": "player_number", "type": "long", "default": 0 }
+                ]
+              }
+            },
+            "default": []
+          },
+          {
+            "name": "contact_renewal",
+            "type": [
+              "null",
+              {
+                "type": "record",
+                "name": "ContactRenewal",
+                "fields": [
+                  { "name": "stamp", "type": "long", "default": 0 },
+                  { "name": "type", "type": "string", "default": "" },
+                  { "name": "xg", "type": "MatchXG", "default": {
+                    "matchXG": 0.0,
+                    "matchXGP": ""
+                  } }
+                ]
+              }
+            ],
+            "default": null
+          },
+          { "name": "defence", "type": "string", "default": "" },
+          { "name": "offence", "type": "string", "default": "" }
+        ]
+      },
+      "default": {}
+    }
+  ]
+}