apache / apache/pulsar-client-python
Python client does not handle default schema values
- 主要语言
- Python
- 星标
- 75
- 派生
- 53
- PR 合并指标
- 30 天内没有已合并 PR
描述
**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()
})
```
贡献指南
评估
这个 Issue 还没有评估数据。