apache/pulsar · error · IllegalArgumentException

Topic %s has an incorrect schema Info

Error message

Topic %s has an incorrect schema Info

What it means

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.

Source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java:679

        // Check if the Input serialization/deserialization class exists in jar or already loaded and that it
        // implements SerDe class
        if (functionConfig.getCustomSerdeInputs() != null) {
            functionConfig.getCustomSerdeInputs().forEach((topicName, inputSerializer) -> {
                ValidatorUtils.validateSerde(inputSerializer, typeArgs[0], validatableFunctionPackage.getTypePool(),
                        true);
            });
        }

        // Check if the Input serialization/deserialization class exists in jar or already loaded and that it
        // implements SerDe class
        if (functionConfig.getCustomSchemaInputs() != null) {
            functionConfig.getCustomSchemaInputs().forEach((topicName, conf) -> {
                ConsumerConfig consumerConfig;
                try {
                    consumerConfig = OBJECT_MAPPER.readValue(conf, ConsumerConfig.class);
                } catch (JsonProcessingException e) {
                    throw new IllegalArgumentException(
                            String.format("Topic %s has an incorrect schema Info", topicName));
                }
                ValidatorUtils.validateSchema(consumerConfig.getSchemaType(), typeArgs[0],
                        validatableFunctionPackage.getTypePool(), true);
            });
        }

        // Check if the Input serialization/deserialization class exists in jar or already loaded and that it
        // implements Schema or SerDe classes

        if (functionConfig.getInputSpecs() != null) {
            functionConfig.getInputSpecs().forEach((topicName, conf) -> {
                // Need to make sure that one and only one of schema/serde is set
                if (!isEmpty(conf.getSchemaType()) && !isEmpty(conf.getSerdeClassName())) {
                    throw new IllegalArgumentException(
                        "Only one of schemaType or serdeClassName should be set in inputSpec");
                }
                if (!isEmpty(conf.getSerdeClassName())) {

View on GitHub (pinned to 820761864e)

Solutions

1. Provide the schema info as valid JSON, e.g. {"schemaType":"avro"} or {"schemaType":"string"}
2. Quote the argument properly in the CLI so quotes survive shell parsing
3. Validate the JSON with a parser before submitting the function

Example fix

// before
--custom-schema-inputs my-topic=string
// after
--custom-schema-inputs my-topic='{"schemaType":"string"}'

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); } }

boolean isValidJson(String s) { try { new ObjectMapper().readTree(s); return true; } catch (JsonProcessingException e) { return false; } }

try { ... } catch (IllegalArgumentException e) { if (e.getMessage().contains("incorrect schema Info")) { log.error("Fix JSON for topic: {}", e.getMessage()); } }

When it happens

Trigger: 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.

Common situations: Hand-editing customSchemaInputs maps; shell escaping mangles quotes in CLI --custom-schema-inputs arguments; YAML/JSON mixing errors.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• JSON Record Builder doesn't support non-JSON record as a fie
• Only one of schemaType or serdeClassName should be set in in
• Only one of outputSchemaType or outputSerdeClassName should
• Could not validate sink config:
• RestException(conversionError)

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/f855d34da6e4963e. Report an issue: GitHub.