Ensure schemaId initialization is thread-safe (#2540)

* Ensure schemaId initialization is thread-safe

* Minor cleanup

Author: rayokota

src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs

@@ -237,7 +237,7 @@ public override async Task<byte[]> SerializeAsync(T value, SerializationContext
 
                     if (latestSchema != null)
                     {
-                        schemaId = new SchemaId(SchemaType.Protobuf, latestSchema.Id, latestSchema.Guid);
+                        schemaId = new SchemaId(SchemaType.Protobuf, latestSchema.Id, latestSchema.Guid, indexArray);
                     }
                     else if (!subjectsRegistered.Contains(subject))
                     {
@@ -259,7 +259,7 @@ await RegisterOrGetReferences(value.Descriptor.File, context, autoRegisterSchema
                         // note: different values for schemaId should never be seen here.
                         // TODO: but fail fast may be better here.
 
-                        schemaId = new SchemaId(SchemaType.Protobuf, outputSchema.Id, outputSchema.Guid);
+                        schemaId = new SchemaId(SchemaType.Protobuf, outputSchema.Id, outputSchema.Guid, indexArray);
                         subjectsRegistered.Add(subject);
                     }
                 }
@@ -282,8 +282,6 @@ await RegisterOrGetReferences(value.Descriptor.File, context, autoRegisterSchema
                         .ConfigureAwait(continueOnCapturedContext: false);
                 }
 
-                schemaId.MessageIndexes = indexArray;
-
                 var buffer = new byte[value.CalculateSize()];
                 value.WriteTo(buffer);
                 buffer = await ExecuteRules(context.Component == MessageComponentType.Key,

src/Confluent.SchemaRegistry/SchemaId.cs

@@ -80,6 +80,14 @@ public SchemaId(SchemaType schemaType, int id, string guid)
         MessageIndexes = null;
     }
 
+    public SchemaId(SchemaType schemaType, int id, string guid, IReadOnlyList<int> messageIndexes)
+    {
+        SchemaType = schemaType;
+        Id = id;
+        Guid = guid != null ? System.Guid.Parse(guid) : null;
+        MessageIndexes = messageIndexes;
+    }
+
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
     public ReadOnlyMemory<byte> FromBytes(ReadOnlyMemory<byte> payload)
     {