{"record":{"id":"2632d41e26eb2de7","repo":"apache/pulsar","slug":"incorrect-custom-schema-inputs-topic-s","errorCode":null,"errorMessage":"Incorrect custom schema inputs,Topic %s ","messagePattern":"Incorrect custom schema inputs,Topic (.+?) ","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java","lineNumber":143,"sourceCode":"                        .setSerdeClassName(serdeClassName)\n                        .setIsRegexPattern(false);\n            });\n        }\n        if (functionConfig.getCustomSchemaInputs() != null) {\n            functionConfig.getCustomSchemaInputs().forEach((topicName, conf) -> {\n                try {\n                    ConsumerConfig consumerConfig = OBJECT_MAPPER.readValue(conf, ConsumerConfig.class);\n                    ConsumerSpec cs = sourceSpec.putInputSpecs(topicName)\n                            .setSchemaType(consumerConfig.getSchemaType())\n                            .setIsRegexPattern(false);\n                    if (consumerConfig.getSchemaProperties() != null) {\n                        consumerConfig.getSchemaProperties().forEach(cs::putSchemaProperties);\n                    }\n                    if (consumerConfig.getConsumerProperties() != null) {\n                        consumerConfig.getConsumerProperties().forEach(cs::putConsumerProperties);\n                    }\n                } catch (JsonProcessingException e) {\n                    throw new IllegalArgumentException(\n                            String.format(\"Incorrect custom schema inputs,Topic %s \", topicName));\n                }\n            });\n        }\n        if (functionConfig.getInputSpecs() != null) {\n            functionConfig.getInputSpecs().forEach((topicName, consumerConf) -> {\n                ConsumerSpec bldr = sourceSpec.putInputSpecs(topicName)\n                        .setIsRegexPattern(consumerConf.isRegexPattern());\n                if (isNotBlank(consumerConf.getSchemaType())) {\n                    bldr.setSchemaType(consumerConf.getSchemaType());\n                } else if (isNotBlank(consumerConf.getSerdeClassName())) {\n                    bldr.setSerdeClassName(consumerConf.getSerdeClassName());\n                }\n                if (consumerConf.getReceiverQueueSize() != null) {\n                    bldr.setReceiverQueueSize().setValue(consumerConf.getReceiverQueueSize());\n                }\n                if (consumerConf.getSchemaProperties() != null) {\n                    consumerConf.getSchemaProperties().forEach(bldr::putSchemaProperties);","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java#L125-L161","documentation":"When converting a FunctionConfig, custom per-topic schema info is serialized to JSON to build ConsumerSpec objects. If that JSON processing fails (JsonProcessingException) for an input topic's custom schema type/properties, the exception is rethrown as this IllegalArgumentException naming the topic.","triggerScenarios":"Setting an inputSpec with an invalid schemaType/schemaProperties/consumerProperties that cannot be JSON-serialized — typically a non-serializable value placed in schema type configuration, or a malformed custom-schema input string that fails JSON parsing in the inputSpecs conversion path for the given topicName.","commonSituations":"Supplying schema configs via CLI as raw JSON strings with syntax errors; putting non-String/complex objects in schema properties when only JSON-primitive maps are expected; copy-pasted schema definitions with unescaped characters; mixing Avro/Protobuf schema descriptors incorrectly.","solutions":["Validate that each inputSpec's schema type string is a valid schema type and its schema/consumer properties are simple string maps","Fix malformed JSON supplied in the schema configuration (quote keys, escape strings, remove comments/trailing commas)","Use supported schema types (e.g. JSON, AVRO, STRING, INT64) or fully qualified class names for AutoConsume/Custom schemas","Test the schema config in isolation (serialize it with the same ObjectMapper) before submitting the function"],"exampleFix":"// before\ninputSpecs.put(\"topic-1\", new ConsumerConfig().setSchemaType(\"AVRO; bad json {\")); // fails JSON processing\n// after\nConsumerConfig cc = new ConsumerConfig();\ncc.setSchemaType(\"AVRO\");\ncc.setSchemaProperties(Collections.singletonMap(\"avroSchemaLocation\", \"/schemas/t1.avsc\"));\ninputSpecs.put(\"topic-1\", cc);","handlingStrategy":"try-catch","validationCode":"// Pre-validate input custom schema config is JSON-friendly\nConsumerConfig cc = inputSpecs.get(topic);\nif (cc != null && cc.getSchemaType() != null) {\n  try {\n    new ObjectMapper().writeValueAsString(cc);\n  } catch (JsonProcessingException e) {\n    throw new IllegalArgumentException(\"Input schema config for topic \" + topic + \" is not JSON-serializable\");\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  FunctionDetails d = FunctionConfigUtils.convert(cfg, pkg);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage() != null && e.getMessage().startsWith(\"Incorrect custom schema inputs\")) {\n    String topic = e.getMessage().replace(\"Incorrect custom schema inputs,Topic \", \"\").trim();\n    log.error(\"Fix custom schema input config for topic {}\", topic);\n  } else {\n    throw e;\n  }\n}","preventionTips":["Keep schemaType a known type string and schema/consumer properties as Map<String,String>","Validate JSON schema configs with a parser before submission","Avoid embedding raw complex objects in schema properties","Test function conversion locally (call convert()) before deploying to a cluster"],"tags":["pulsar-functions","schema","json","configuration"],"backgroundTag":"schema-validation-failed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}