[Python Schema] Support setting namespace for python schema

pulsar-client-cpp/python/pulsar/schema/definition.py

@@ -57,6 +57,9 @@ def _get_fields(cls, dct):
 
 class Record(with_metaclass(RecordMeta, object)):
 
+    # This field is used to set namespace for Avro Record schema.
+    _avro_namespace = None
+
     def __init__(self, default=None, required_default=False, required=False, *args, **kwargs):
         self._required_default = required_default
         self._default = default
@@ -101,15 +104,22 @@ def schema(cls):
 
     @classmethod
     def schema_info(cls, defined_names):
-        if cls.__name__ in defined_names:
-            return cls.__name__
-
-        defined_names.add(cls.__name__)
-        schema = {
-            'name': str(cls.__name__),
-            'type': 'record',
-            'fields': []
-        }
+        namespace_prefix = ''
+        if cls._avro_namespace is not None:
+            namespace_prefix = cls._avro_namespace + '.'
+        namespace_name = namespace_prefix + cls.__name__
+
+        if namespace_name in defined_names:
+            return namespace_name
+
+        defined_names.add(namespace_name)
+
+        schema = {'name': str(cls.__name__)}
+        if cls._avro_namespace is not None:
+            schema['namespace'] = cls._avro_namespace
+        schema['type'] = 'record'
+        schema['fields'] = []
+
         for name in sorted(cls._fields.keys()):
             field = cls._fields[name]
             field_type = field.schema_info(defined_names) \

pulsar-client-cpp/python/schema_test.py

@@ -891,6 +891,7 @@ class NestedObj3(Record):
             na3 = Integer()
 
         class NestedObj4(Record):
+            _avro_namespace = 'xxx4'
             na4 = String()
             nb4 = Integer()
 
@@ -900,6 +901,7 @@ class Color(Enum):
             blue = 3
 
         class ComplexRecord(Record):
+            _avro_namespace = 'xxx.xxx'
             a = Integer()
             b = Integer()
             color = Color
@@ -914,16 +916,17 @@ class ComplexRecord(Record):
         print('complex schema: ', ComplexRecord.schema())
         self.assertEqual(ComplexRecord.schema(), {
             "name": "ComplexRecord",
+            "namespace": "xxx.xxx",
             "type": "record",
             "fields": [
                 {"name": "a", "type": ["null", "int"]},
                 {'name': 'arrayNested', 'type': ['null', {'type': 'array', 'items':
-                    {'name': 'NestedObj4', 'type': 'record', 'fields': [
+                    {'name': 'NestedObj4', 'namespace': 'xxx4', 'type': 'record', 'fields': [
                         {'name': 'na4', 'type': ['null', 'string']},
                         {'name': 'nb4', 'type': ['null', 'int']}
                     ]}}
                 ]},
-                {'name': 'arrayNested2', 'type': ['null', {'type': 'array', 'items': 'NestedObj4'}]},
+                {'name': 'arrayNested2', 'type': ['null', {'type': 'array', 'items': 'xxx4.NestedObj4'}]},
                 {"name": "b", "type": ["null", "int"]},
                 {'name': 'color', 'type': ['null', {'type': 'enum', 'name': 'Color', 'symbols': [
                     'red', 'green', 'blue']}]},
@@ -1104,5 +1107,12 @@ def produce_consume_test(schema_type):
 
         client.close()
 
+    def test(self):
+        class NamespaceDemo(Record):
+            _namespace = 'xxx.xxx.xxx'
+            x = String()
+            y = Integer()
+        print('schema: ', NamespaceDemo.schema())
+
 if __name__ == '__main__':
     main()

site2/docs/client-libraries-python.md

@@ -307,6 +307,26 @@ class Example(Record):
     sub = MySubRecord()
 ```
 
+##### Set namespace for Avro schema
+
+Set the namespace for Avro Record schema using the special field `_avro_namespace`.
+```python
+class NamespaceDemo(Record):
+   _avro_namespace = 'xxx.xxx.xxx'
+   x = String()
+   y = Integer()
+```
+
+The schema definition is like this.
+```
+{
+  'name': 'NamespaceDemo', 'namespace': 'xxx.xxx.xxx', 'type': 'record', 'fields': [
+    {'name': 'x', 'type': ['null', 'string']}, 
+    {'name': 'y', 'type': ['null', 'int']}
+  ]
+}
+```
+
 ## End-to-end encryption
 
 [End-to-end encryption](https://pulsar.apache.org/docs/en/next/cookbooks-encryption/#docsNav) allows applications to encrypt messages at producers and decrypt messages at consumers.