Ensure schemaId initialization is thread-safe (#2120)

* Ensure schemaId initialization is thread-safe

* Minor fix

Author: rayokota

src/confluent_kafka/schema_registry/_async/protobuf.py

@@ -395,7 +395,8 @@ async def __serialize(self, message: Message, ctx: Optional[SerializationContext
             latest_schema = await self._get_reader_schema(subject, fmt='serialized')
 
         if latest_schema is not None:
-            self._schema_id = SchemaId(PROTOBUF_TYPE, latest_schema.schema_id, latest_schema.guid)
+            self._schema_id = SchemaId(PROTOBUF_TYPE, latest_schema.schema_id,
+                                       latest_schema.guid, self._index_array)
 
         elif subject not in self._known_subjects and ctx is not None:
             references = await self._resolve_dependencies(ctx, message.DESCRIPTOR.file)
@@ -408,11 +409,13 @@ async def __serialize(self, message: Message, ctx: Optional[SerializationContext
             if self._auto_register:
                 registered_schema = await self._registry.register_schema_full_response(
                     subject, self._schema, normalize_schemas=self._normalize_schemas)
-                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id, registered_schema.guid)
+                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id,
+                                           registered_schema.guid, self._index_array)
             else:
                 registered_schema = await self._registry.lookup_schema(
                     subject, self._schema, normalize_schemas=self._normalize_schemas)
-                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id, registered_schema.guid)
+                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id,
+                                           registered_schema.guid, self._index_array)
 
             self._known_subjects.add(subject)
 
@@ -428,7 +431,6 @@ def field_transformer(rule_ctx, field_transform, msg): return (  # noqa: E731
 
         with _ContextStringIO() as fo:
             fo.write(message.SerializeToString())
-            self._schema_id.message_indexes = self._index_array
             buffer = fo.getvalue()
 
             if latest_schema is not None:

src/confluent_kafka/schema_registry/_sync/protobuf.py

@@ -395,7 +395,8 @@ def __serialize(self, message: Message, ctx: Optional[SerializationContext] = No
             latest_schema = self._get_reader_schema(subject, fmt='serialized')
 
         if latest_schema is not None:
-            self._schema_id = SchemaId(PROTOBUF_TYPE, latest_schema.schema_id, latest_schema.guid)
+            self._schema_id = SchemaId(PROTOBUF_TYPE, latest_schema.schema_id,
+                                       latest_schema.guid, self._index_array)
 
         elif subject not in self._known_subjects and ctx is not None:
             references = self._resolve_dependencies(ctx, message.DESCRIPTOR.file)
@@ -408,11 +409,13 @@ def __serialize(self, message: Message, ctx: Optional[SerializationContext] = No
             if self._auto_register:
                 registered_schema = self._registry.register_schema_full_response(
                     subject, self._schema, normalize_schemas=self._normalize_schemas)
-                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id, registered_schema.guid)
+                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id,
+                                           registered_schema.guid, self._index_array)
             else:
                 registered_schema = self._registry.lookup_schema(
                     subject, self._schema, normalize_schemas=self._normalize_schemas)
-                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id, registered_schema.guid)
+                self._schema_id = SchemaId(PROTOBUF_TYPE, registered_schema.schema_id,
+                                           registered_schema.guid, self._index_array)
 
             self._known_subjects.add(subject)
 
@@ -428,7 +431,6 @@ def field_transformer(rule_ctx, field_transform, msg): return (  # noqa: E731
 
         with _ContextStringIO() as fo:
             fo.write(message.SerializeToString())
-            self._schema_id.message_indexes = self._index_array
             buffer = fo.getvalue()
 
             if latest_schema is not None: