apache / apache/pulsar-client-python

Python client does not handle default schema values

Abierto
#34 9 comentarios 2 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Python
Estrellas
75
Forks
53
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

**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()
})
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.