{"record":{"id":"f855d34da6e4963e","repo":"apache/pulsar","slug":"topic-s-has-an-incorrect-schema-info","errorCode":null,"errorMessage":"Topic %s has an incorrect schema Info","messagePattern":"Topic (.+?) has an incorrect schema Info","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java","lineNumber":679,"sourceCode":"\n        // Check if the Input serialization/deserialization class exists in jar or already loaded and that it\n        // implements SerDe class\n        if (functionConfig.getCustomSerdeInputs() != null) {\n            functionConfig.getCustomSerdeInputs().forEach((topicName, inputSerializer) -> {\n                ValidatorUtils.validateSerde(inputSerializer, typeArgs[0], validatableFunctionPackage.getTypePool(),\n                        true);\n            });\n        }\n\n        // Check if the Input serialization/deserialization class exists in jar or already loaded and that it\n        // implements SerDe class\n        if (functionConfig.getCustomSchemaInputs() != null) {\n            functionConfig.getCustomSchemaInputs().forEach((topicName, conf) -> {\n                ConsumerConfig consumerConfig;\n                try {\n                    consumerConfig = OBJECT_MAPPER.readValue(conf, ConsumerConfig.class);\n                } catch (JsonProcessingException e) {\n                    throw new IllegalArgumentException(\n                            String.format(\"Topic %s has an incorrect schema Info\", topicName));\n                }\n                ValidatorUtils.validateSchema(consumerConfig.getSchemaType(), typeArgs[0],\n                        validatableFunctionPackage.getTypePool(), true);\n            });\n        }\n\n        // Check if the Input serialization/deserialization class exists in jar or already loaded and that it\n        // implements Schema or SerDe classes\n\n        if (functionConfig.getInputSpecs() != null) {\n            functionConfig.getInputSpecs().forEach((topicName, conf) -> {\n                // Need to make sure that one and only one of schema/serde is set\n                if (!isEmpty(conf.getSchemaType()) && !isEmpty(conf.getSerdeClassName())) {\n                    throw new IllegalArgumentException(\n                        \"Only one of schemaType or serdeClassName should be set in inputSpec\");\n                }\n                if (!isEmpty(conf.getSerdeClassName())) {","sourceCodeStart":661,"sourceCodeEnd":697,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java#L661-L697","documentation":"For each entry in customSchemaInputs, doJavaChecks parses the JSON value into a ConsumerConfig. If the value is not valid JSON (JsonProcessingException), an IllegalArgumentException naming the topic is thrown before any schema validation occurs.","triggerScenarios":"Passing a custom schema config for a topic whose value is malformed JSON, e.g. missing quotes around the type, trailing characters, or passing a raw string instead of a JSON object.","commonSituations":"Hand-editing customSchemaInputs maps; shell escaping mangles quotes in CLI --custom-schema-inputs arguments; YAML/JSON mixing errors.","solutions":["Provide the schema info as valid JSON, e.g. {\"schemaType\":\"avro\"} or {\"schemaType\":\"string\"}","Quote the argument properly in the CLI so quotes survive shell parsing","Validate the JSON with a parser before submitting the function"],"exampleFix":"// before\n--custom-schema-inputs my-topic=string\n// after\n--custom-schema-inputs my-topic='{\"schemaType\":\"string\"}'","handlingStrategy":"validation","validationCode":"for (Map.Entry<String,String> e : config.getCustomSchemaInputs().entrySet()) { try { new ObjectMapper().readTree(e.getValue()); } catch (JsonProcessingException ex) { throw new IllegalArgumentException(\"customSchemaInputs entry for \" + e.getKey() + \" is not valid JSON\", ex); } }","typeGuard":"boolean isValidJson(String s) { try { new ObjectMapper().readTree(s); return true; } catch (JsonProcessingException e) { return false; } }","tryCatchPattern":"try { ... } catch (IllegalArgumentException e) { if (e.getMessage().contains(\"incorrect schema Info\")) { log.error(\"Fix JSON for topic: {}\", e.getMessage()); } }","preventionTips":["Always pass schema info as a JSON object like {\"schemaType\":\"string\"}","Quote CLI arguments to protect inner quotes from shell parsing","Validate all JSON fields in config before submission"],"tags":["pulsar","functions","json","schema"],"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"}