Improve lock utilization on the consumer hot path (#2370)

Author: Havret

src/Confluent.SchemaRegistry/AsyncSerde.cs

@@ -18,6 +18,7 @@
 #pragma warning disable CS0618
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
@@ -41,7 +42,7 @@ public abstract class AsyncSerde<TParsedSchema>
 
         protected SemaphoreSlim serdeMutex = new SemaphoreSlim(1);
 
-        private readonly IDictionary<Schema, TParsedSchema> parsedSchemaCache = new Dictionary<Schema, TParsedSchema>();
+        private readonly IDictionary<Schema, TParsedSchema> parsedSchemaCache = new ConcurrentDictionary<Schema, TParsedSchema>();
         private SemaphoreSlim parsedSchemaMutex = new SemaphoreSlim(1);
 
         protected AsyncSerde(ISchemaRegistryClient schemaRegistryClient, SerdeConfig config, RuleRegistry ruleRegistry = null)
@@ -98,10 +99,15 @@ protected string GetSubjectName(string topic, bool isKey, string recordType)
 
         protected async Task<TParsedSchema> GetParsedSchema(Schema schema)
         {
+            if (parsedSchemaCache.TryGetValue(schema, out TParsedSchema parsedSchema))
+            {
+                return parsedSchema;
+            }
+            
             await parsedSchemaMutex.WaitAsync().ConfigureAwait(continueOnCapturedContext: false);
             try
             {
-                if (!parsedSchemaCache.TryGetValue(schema, out TParsedSchema parsedSchema))
+                if (!parsedSchemaCache.TryGetValue(schema, out parsedSchema))
                 {
                     if (parsedSchemaCache.Count > schemaRegistryClient.MaxCachedSchemas)
                     {

src/Confluent.SchemaRegistry/CachedSchemaRegistryClient.cs

@@ -23,6 +23,7 @@
 using System.Threading.Tasks;
 using System.Linq;
 using System;
+using System.Collections.Concurrent;
 using System.Net;
 using System.Threading;
 using System.Security.Cryptography.X509Certificates;
@@ -65,7 +66,7 @@ public class CachedSchemaRegistryClient : ISchemaRegistryClient
         private IRestService restService;
         private int identityMapCapacity;
         private int latestCacheTtlSecs;
-        private readonly Dictionary<int, Schema> schemaById = new Dictionary<int, Schema>();
+        private readonly ConcurrentDictionary<int, Schema> schemaById = new ConcurrentDictionary<int, Schema>();
 
         private readonly Dictionary<string /*subject*/, Dictionary<Schema, int>> idBySchemaBySubject =
             new Dictionary<string, Dictionary<Schema, int>>();
@@ -657,11 +658,15 @@ public async Task<Schema> GetSchemaAsync(int id, string format = null)
         /// <inheritdoc/>
         public async Task<Schema> GetSchemaBySubjectAndIdAsync(string subject, int id, string format = null)
         {
+            if (this.schemaById.TryGetValue(id, out Schema schema) && checkSchemaMatchesFormat(format, schema.SchemaString))
+            {
+                return schema;
+            }
+            
             await cacheMutex.WaitAsync().ConfigureAwait(continueOnCapturedContext: false);
             try
             {
-                if (!this.schemaById.TryGetValue(id, out Schema schema) ||
-                    !checkSchemaMatchesFormat(format, schema.SchemaString))
+                if (!this.schemaById.TryGetValue(id, out schema) || !checkSchemaMatchesFormat(format, schema.SchemaString))
                 {
                     CleanCacheIfFull();
                     schema = (await restService.GetSchemaBySubjectAndIdAsync(subject, id, format)