apache / apache/pulsar-client-python

Python client does not handle default schema values

Đang mở
#34 9 bình luận 2 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
75
Fork
53
Chỉ số merge pull request
Không có pull request nào được merge trong 30 ngày

Mô tả

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

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.