Isolate use or orjson to JSON serdes (#2025)

Author: rayokota

requirements/requirements-examples.txt

@@ -16,6 +16,7 @@ avro>=1.11.1,<2
 
 pyrsistent
 jsonschema
+orjson >= 3.10
 
 googleapis-common-protos
 protobuf

requirements/requirements-json.txt

@@ -1,2 +1,3 @@
 pyrsistent
-jsonschema
+jsonschema
+orjson >= 3.10

requirements/requirements-schemaregistry.txt

@@ -2,4 +2,3 @@ attrs>=21.2.0
 cachetools>=5.5.0
 httpx>=0.26
 authlib>=1.0.0
-orjson >= 3.10

src/confluent_kafka/avro/serializer/message_serializer.py

@@ -20,7 +20,7 @@
 # derived from https://github.com/verisign/python-confluent-schemaregistry.git
 #
 import io
-import orjson
+import json
 import logging
 import struct
 import sys
@@ -80,7 +80,7 @@ def __init__(self, registry_client, reader_key_schema=None, reader_value_schema=
     # Encoder support
     def _get_encoder_func(self, writer_schema):
         if HAS_FAST:
-            schema = orjson.loads(str(writer_schema))
+            schema = json.loads(str(writer_schema))
             parsed_schema = parse_schema(schema)
             return lambda record, fp: schemaless_writer(fp, parsed_schema, record)
         writer = avro.io.DatumWriter(writer_schema)
@@ -176,9 +176,9 @@ def _get_decoder_func(self, schema_id, payload, is_key=False):
         if HAS_FAST:
             # try to use fast avro
             try:
-                fast_avro_writer_schema = parse_schema(orjson.loads(str(writer_schema_obj)))
+                fast_avro_writer_schema = parse_schema(json.loads(str(writer_schema_obj)))
                 if reader_schema_obj is not None:
-                    fast_avro_reader_schema = parse_schema(orjson.loads(str(reader_schema_obj)))
+                    fast_avro_reader_schema = parse_schema(json.loads(str(reader_schema_obj)))
                 else:
                     fast_avro_reader_schema = None
                 schemaless_reader(payload, fast_avro_writer_schema)

src/confluent_kafka/kafkatest/verifiable_client.py

@@ -14,7 +14,7 @@
 
 
 import datetime
-import orjson
+import json
 import os
 import re
 import signal
@@ -61,8 +61,8 @@ def err(self, s, term=False):
     def send(self, d):
         """ Send dict as JSON to stdout for consumtion by kafkatest handler """
         d['_time'] = str(datetime.datetime.now())
-        self.dbg('SEND: %s' % orjson.dumps(d).decode("utf-8"))
-        sys.stdout.write('%s\n' % orjson.dumps(d).decode("utf-8"))
+        self.dbg('SEND: %s' % json.dumps(d))
+        sys.stdout.write('%s\n' % json.dumps(d))
         sys.stdout.flush()
 
     @staticmethod

src/confluent_kafka/schema_registry/_async/avro.py

@@ -15,7 +15,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import io
-import orjson
+import json
 from typing import Dict, Union, Optional, Callable
 
 from fastavro import schemaless_reader, schemaless_writer
@@ -270,7 +270,7 @@ async def __init_impl(
                 # i.e. {"type": "string"} has a name of string.
                 # This function does not comply.
                 # https://github.com/fastavro/fastavro/issues/415
-                schema_dict = orjson.loads(schema.schema_str)
+                schema_dict = json.loads(schema.schema_str)
                 schema_name = parsed_schema.get("name", schema_dict.get("type"))
         else:
             schema_name = None

src/confluent_kafka/schema_registry/_async/schema_registry_client.py

@@ -16,7 +16,7 @@
 # limitations under the License.
 #
 import asyncio
-import orjson
+import json
 import logging
 import time
 import urllib
@@ -416,7 +416,7 @@ async def send_request(
                              " application/json"}
 
         if body is not None:
-            body = orjson.dumps(body).decode('utf-8')
+            body = json.dumps(body)
             headers = {'Content-Length': str(len(body)),
                        'Content-Type': "application/vnd.schemaregistry.v1+json"}
 

src/confluent_kafka/schema_registry/_sync/avro.py

@@ -15,7 +15,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import io
-import orjson
+import json
 from typing import Dict, Union, Optional, Callable
 
 from fastavro import schemaless_reader, schemaless_writer
@@ -270,7 +270,7 @@ def __init_impl(
                 # i.e. {"type": "string"} has a name of string.
                 # This function does not comply.
                 # https://github.com/fastavro/fastavro/issues/415
-                schema_dict = orjson.loads(schema.schema_str)
+                schema_dict = json.loads(schema.schema_str)
                 schema_name = parsed_schema.get("name", schema_dict.get("type"))
         else:
             schema_name = None

src/confluent_kafka/schema_registry/_sync/schema_registry_client.py

@@ -16,7 +16,7 @@
 # limitations under the License.
 #
 
-import orjson
+import json
 import logging
 import time
 import urllib
@@ -416,7 +416,7 @@ def send_request(
                              " application/json"}
 
         if body is not None:
-            body = orjson.dumps(body).decode('utf-8')
+            body = json.dumps(body)
             headers = {'Content-Length': str(len(body)),
                        'Content-Type': "application/vnd.schemaregistry.v1+json"}
 

src/confluent_kafka/schema_registry/common/avro.py

@@ -3,7 +3,7 @@
 from collections import defaultdict
 from copy import deepcopy
 from io import BytesIO
-import orjson
+import json
 from typing import Dict, Union, Optional, Set
 
 from fastavro import repository, validate
@@ -91,7 +91,7 @@ def load(self, subject):
 
 def parse_schema_with_repo(schema_str: str, named_schemas: Dict[str, AvroSchema]) -> AvroSchema:
     copy = deepcopy(named_schemas)
-    copy["$root"] = orjson.loads(schema_str)
+    copy["$root"] = json.loads(schema_str)
     repo = LocalSchemaRepository(copy)
     return load_schema("$root", repo=repo)