code: def test_produce_and_consume_complex_schema_data(self):
      class Color(Enum):
          red = 1
          green = 2
          blue = 3
  
      class NestedObj1(Record):
          na1 = String()
          nb1 = Double()
  
      class NestedObj2(Record):
          na2 = Integer()
          nb2 = Boolean()
          nc2 = NestedObj1()
  
      class NestedObj3(Record):
          na3 = Integer()
          color = CustomEnum(
              Color,
              required=True,
              required_default=True,
              default=Color.blue,
          )
  
      class NestedObj4(Record):
          na4 = String()
          nb4 = Integer()
  
      class ComplexRecord(Record):
          a = Integer()
          b = Integer()
          color = CustomEnum(Color)
          nested = NestedObj2()
          mapNested = Map(NestedObj3())
          arrayNested = Array(NestedObj4())
  
      client = pulsar.Client(self.serviceUrl)
  
      def produce_consume_test(schema_type):
          topic = "my-complex-schema-topic-" + schema_type
  
          data_schema = AvroSchema(ComplexRecord)
          if schema_type == "json":
              data_schema = JsonSchema(ComplexRecord)
  
          producer = client.create_producer(
              topic=topic, schema=data_schema
          )
  
          consumer = client.subscribe(topic, "test", schema=data_schema)
  
          nested_obj1 = NestedObj1(na1="na1 value", nb1=20.5)
          nested_obj2 = NestedObj2(na2=22, nb2=True, nc2=nested_obj1)
          r = ComplexRecord(
              a=1,
              b=2,
              nested=nested_obj2,
              mapNested={
                  "a": NestedObj3(na3=1, color=Color.red),
                  "b": NestedObj3(na3=2, color=Color.green),
                  "c": NestedObj3(na3=3),
              },
              arrayNested=[
                  NestedObj4(na4="value na4 1", nb4=100),
                  NestedObj4(na4="value na4 2", nb4=200),
              ],
          )
          producer.send(r)
  
          msg = consumer.receive()
          value = msg.value()
          self.assertEqual(value.__class__.__name__, "ComplexRecord")
          self.assertEqual(value, r)
  
          print(
              "Produce and consume complex schema data finish. schema_type",
              schema_type,
          )
  
      produce_consume_test("avro")
      produce_consume_test("json")
  
      client.close()
  
imports: import math
  import logging
  import requests
  from typing import List
  from unittest import TestCase, main
  import fastavro
  import pulsar
  from pulsar.schema import *
  from enum import Enum
  import json
  from fastavro.schema import load_schema
label_desc: Use image node_id 28 for calling the SchemaTest obj's underlying member method code with example usage: obj.test_produce_and_consume_complex_schema_data() without return types