[GitHub] [pulsar-client-python] giacomo-porro opened a new issue, #34: Python client does not handle default schema values

[GitBox <gi...@apache.org>]

giacomo-porro opened a new issue, #34:
URL: https://github.com/apache/pulsar-client-python/issues/34

   **Describe the bug**
   When trying to update a schema on a topic ( having as a compatibility check strategy set to BACKWARD_TRANSITIVE or BACKWARD in my case ) 
   pulsar returns the error "IncompatibleSchema" even if all the guidelines have been followed. 
   
   **To Reproduce**
   Steps to reproduce the behavior:
   
   1. Create the producer:
   ```import pulsar
   from pulsar.schema import *
   
   class Example(Record):
   a = String()
   b = Integer()
   c = Integer()
   
   client = pulsar.Client('pulsar://localhost:6650')
   producer = client.create_producer(
   topic='tenant/namespace/my-topic',
   schema=JsonSchema(Example) )
   
   producer.send(Example(a='Hello', b=1))
   ```
   
   2. Create the consumer:
   ```import pulsar
   from pulsar.schema import *
   
   class Example(Record):
   a = String()
   b = Integer()
   c = Integer()
   
   client = pulsar.Client('pulsar://localhost:6650')
   consumer = client.subscribe(
   topic='tenant/namespace/my-topic',
   subscription_name='my-subscription',
   schema=JsonSchema(Example))
   
   while True:
   msg = consumer.receive()
   ex = msg.value()
   try:
   print("Received message a={} b={} c={}".format(ex.a, ex.b, ex.c))
   # Acknowledge successful processing of the message
   consumer.acknowledge(msg)
   except:
   # Message failed to be processed
   consumer.negative_acknowledge(msg)
   ```
   
   3. Run consumer first and then producer, the schema should be created on the topic correctly and the consumer reads the message correctly
   
   4. Stop the consumer
   
   5. Make a change to the consumer, add a float field to the Example class but with a default value:
   ```class Example(Record):
   a = String()
   b = Integer()
   c = Integer()
   d = Float(default=1.0)
   ```
   
   6. Run the consumer again, it throws the error IncompatibleSchema 
   
   **Expected behavior**
   Since the schema compatibility check is set to BACKWARD_TRANSITIVE or BACKWARD, the policy is to upgrade consumer first and an optional property has been added to the schema,
   I expect the schema to be registered correctly.
   
   **Screenshots**
   If applicable, add screenshots to help explain your problem.
   
   **Desktop (please complete the following information):**
    - OS: MacOS Mojave
   
   **Additional context**
   Using pulsar docker container v2.6.0, pulsar-client v2.6.0 for Python
   
   Looking at the logs inside the container I noticed that when the schema gets sent to pulsar from the python client it does not include the default value of the fields that have it, here's an example:
   ```
   Error during schema compatibility check: Unable to read schema: 
   {
     "type" : "record",
     "name" : "Example",
     "fields" : [ {
       "name" : "a",
       "type" : [ "null", "string" ]
     }, {
       "name" : "b",
       "type" : [ "null", "int" ]
     }, {
       "name" : "c",
       "type" : [ "null", "boolean" ]
     }]
   }
   using schema:
   {
     "type" : "record",
     "name" : "Example",
     "fields" : [ {
       "name" : "a",
       "type" : [ "null", "string" ]
     }, {
       "name" : "b",
       "type" : [ "null", "int" ]
     }, {
       "name" : "c",
       "type" : [ "null", "boolean" ]
     }, {
       "name" : "d",
       "type" : [ "null", "float" ]
     }]
   }
   ```
   
   As you can see the second schema is missing the "default": "1.0" key-value pair for the property "d". This makes the new schema incompatible with the old one.
   
   I dug I little deeper in the python client's code and I noticed that in the class Record ( which is extended from my Example class ), and specifically in the schema() method, the definition of the default property on the schema is completely missing.
   
   ```
   @classmethod
   def schema(cls):
       schema = {
           'name': str(cls.__name__),
           'type': 'record',
           'fields': []
       }
   
       for name in sorted(cls._fields.keys()):
           field = cls._fields[name]
           field_type = field.schema() if field._required else ['null', field.schema()]
           schema['fields'].append({
               'name': name,
               'type': field_type
           })
       return schema
   ```
   I think that here, this piece of code
   
   ```
   schema['fields'].append({
               'name': name,
               'type': field_type
           })
   ```
   Should also handle the default value of the field itself e.g.: 
   
   ```
   schema['fields'].append({
               'name': name,
               'type': field_type,
               'default': field.default()
           })
   ```


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org