WIP

confluent_kafka/avro/cached_schema_registry_client.py

@@ -213,7 +213,7 @@ def register(self, subject, avro_schema):
         url = '/'.join([self.url, 'subjects', subject, 'versions'])
         # body is { schema : json_string }
 
-        body = {'schema': json.dumps(avro_schema.to_json())}
+        body = {'schema': json.dumps(str(avro_schema))}
         result, code = self._send_request(url, method='POST', body=body)
         if (code == 401 or code == 403):
             raise ClientError("Unauthorized access. Error code:" + str(code))
@@ -253,7 +253,7 @@ def check_registration(self, subject, avro_schema):
         url = '/'.join([self.url, 'subjects', subject])
         # body is { schema : json_string }
 
-        body = {'schema': json.dumps(avro_schema.to_json())}
+        body = {'schema': json.dumps(str(avro_schema))}
         result, code = self._send_request(url, method='POST', body=body)
         if code == 401 or code == 403:
             raise ClientError("Unauthorized access. Error code:" + str(code))
@@ -374,7 +374,7 @@ def get_version(self, subject, avro_schema):
             return version
 
         url = '/'.join([self.url, 'subjects', subject])
-        body = {'schema': json.dumps(avro_schema.to_json())}
+        body = {'schema': json.dumps(str(avro_schema))}
 
         result, code = self._send_request(url, method='POST', body=body)
         if code == 404:
@@ -402,7 +402,7 @@ def test_compatibility(self, subject, avro_schema, version='latest'):
         """
         url = '/'.join([self.url, 'compatibility', 'subjects', subject,
                         'versions', str(version)])
-        body = {'schema': json.dumps(avro_schema.to_json())}
+        body = {'schema': json.dumps(str(avro_schema))}
         try:
             result, code = self._send_request(url, method='POST', body=body)
             if code == 404:

confluent_kafka/avro/serializer/message_serializer.py

@@ -79,7 +79,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 = writer_schema.to_json()
+            schema = str(writer_schema)
             parsed_schema = parse_schema(schema)
             return lambda record, fp: schemaless_writer(fp, parsed_schema, record)
         writer = avro.io.DatumWriter(writer_schema)
@@ -174,8 +174,8 @@ 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(writer_schema_obj.to_json())
-                fast_avro_reader_schema = parse_schema(reader_schema_obj.to_json())
+                fast_avro_writer_schema = parse_schema(str(writer_schema_obj))
+                fast_avro_reader_schema = parse_schema(str(reader_schema_obj))
                 schemaless_reader(payload, fast_avro_writer_schema)
 
                 # If we reach this point, this means we have fastavro and it can

setup.py

@@ -68,7 +68,7 @@ def get_install_requirements(path):
 setup(name='confluent-kafka',
       # Make sure to bump CFL_VERSION* in confluent_kafka/src/confluent_kafka.h
       # and version and release in docs/conf.py.
-      version='1.4.2',
+      version='1.4.2-kaluza',
       description='Confluent\'s Python client for Apache Kafka',
       author='Confluent Inc',
       author_email='[email protected]',