{"record":{"id":"4628920b673dbafb","repo":"apache/pulsar","slug":"invalid-schema-class-s","errorCode":null,"errorMessage":"Invalid schema class %s","messagePattern":"Invalid schema class (.+?)","errorType":"exception","errorClass":"Exception","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/python/python_instance.py","lineNumber":574,"sourceCode":"    else:  # load custom schema\n      record_kclass = self.get_record_class(type_class_name)\n      schema_kclass = util.import_class(os.path.dirname(self.user_code), schema_type)\n      args_count = 0\n      try:\n        args_count = len(inspect.signature(schema_kclass.__init__).parameters)\n      except:  # for compatibility with python 2\n        args_count = len(inspect.getargspec(schema_kclass.__init__).args)\n      if args_count == 1:  # doesn't take any arguments\n        schema = schema_kclass()\n      elif args_count == 2:  # take one argument, it can be either schema properties or record class\n        try:\n          schema = schema_kclass(record_kclass)\n        except TypeError:\n          schema = schema_kclass(schema_properties)\n      elif args_count >= 3:  # take two or more arguments\n        schema = schema_kclass(record_kclass, schema_properties)\n      else:\n        raise Exception(\"Invalid schema class %s\" % schema_type)\n    return schema\n\n  def get_record_class(self, class_name):\n      record_kclass = None\n      if class_name != None and len(class_name) > 0:\n        try:\n          record_kclass = util.import_class(os.path.dirname(self.user_code), class_name)\n        except:\n          pass\n      return record_kclass\n  def get_negative_ack_args(self):\n    \"\"\"Build the negative-ack redelivery delay argument for Client.subscribe().\n\n    Returns a dict to splat into the subscribe() call: either empty, or carrying\n    negative_ack_redelivery_delay_ms.\n\n    SourceSpec.negativeAckRedeliveryDelayMs is a proto3 scalar with no presence, so an unset field\n    reads as 0. Only a positive value is forwarded, leaving the client default (60s) in place","sourceCodeStart":556,"sourceCodeEnd":592,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/python/python_instance.py#L556-L592","documentation":"python_instance.get_schema builds a schema object from the configured schema type and record class. If the resolved schema class accepts fewer than the required constructor arguments (args_count < 3 fallback path and no matching branch), Pulsar raises Exception('Invalid schema class %s') — the configured schema class is not a supported Avro/JSON-style schema that can be instantiated with the record class and schema properties.","triggerScenarios":"Configuring a custom or built-in schema class whose __init__ accepts zero or one argument such that args_count branches all fail; passing schemaType that maps to a class incompatible with the resolved record class; get_schema called from run or setup_producer when output schema data is misconfigured.","commonSituations":"Custom schema class with a nonstandard constructor signature; using a schema type (e.g. 'string' or 'void') in a context requiring a record class; record class resolution failing so record_kclass/schema_properties combination matches no branch.","solutions":["Use a supported schema type (avro/json/protobuf) with a valid record class, or omit schema config for builtin types.","Give the custom schema class a constructor accepting (record_class, schema_properties) or at least one argument that matches the branch conditions.","Verify get_record_class resolves the configured recordClassName to a real Python class.","Log args_count and schema_type to see which instantiation branch was expected."],"exampleFix":"# before\nclass MySchema:\n    def __init__(self):  # no args — unsupported\n        ...\n# after\nclass MySchema:\n    def __init__(self, record_class, schema_properties=None):\n        ...","handlingStrategy":"validation","validationCode":"import inspect\nsig = inspect.signature(MySchema.__init__)\nassert len(sig.parameters) >= 2 or any(\n    p.default is not inspect.Parameter.empty for p in list(sig.parameters.values())[1:]), \\\n    'schema class constructor incompatible'","typeGuard":null,"tryCatchPattern":"try:\n    schema = InstanceConfig.get_schema(schema_type)\nexcept Exception as e:\n    logging.error('schema class %s not instantiable: %s', schema_type, e)","preventionTips":["Prefer builtin schema types (avro/json) when possible.","Give custom schema classes constructors accepting (record_class, schema_properties).","Verify recordClassName resolves to a real class.","Test get_schema in unit tests with the exact schema config."],"tags":["python","pulsar-functions","schema","configuration"],"backgroundTag":"invalid-schema-class","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}