apache/pulsar · error · IllegalArgumentException

Incorrect custom schema inputs,Topic %s

Error message

Incorrect custom schema inputs,Topic %s 

What it means

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.

Source

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

                        .setSerdeClassName(serdeClassName)
                        .setIsRegexPattern(false);
            });
        }
        if (functionConfig.getCustomSchemaInputs() != null) {
            functionConfig.getCustomSchemaInputs().forEach((topicName, conf) -> {
                try {
                    ConsumerConfig consumerConfig = OBJECT_MAPPER.readValue(conf, ConsumerConfig.class);
                    ConsumerSpec cs = sourceSpec.putInputSpecs(topicName)
                            .setSchemaType(consumerConfig.getSchemaType())
                            .setIsRegexPattern(false);
                    if (consumerConfig.getSchemaProperties() != null) {
                        consumerConfig.getSchemaProperties().forEach(cs::putSchemaProperties);
                    }
                    if (consumerConfig.getConsumerProperties() != null) {
                        consumerConfig.getConsumerProperties().forEach(cs::putConsumerProperties);
                    }
                } catch (JsonProcessingException e) {
                    throw new IllegalArgumentException(
                            String.format("Incorrect custom schema inputs,Topic %s ", topicName));
                }
            });
        }
        if (functionConfig.getInputSpecs() != null) {
            functionConfig.getInputSpecs().forEach((topicName, consumerConf) -> {
                ConsumerSpec bldr = sourceSpec.putInputSpecs(topicName)
                        .setIsRegexPattern(consumerConf.isRegexPattern());
                if (isNotBlank(consumerConf.getSchemaType())) {
                    bldr.setSchemaType(consumerConf.getSchemaType());
                } else if (isNotBlank(consumerConf.getSerdeClassName())) {
                    bldr.setSerdeClassName(consumerConf.getSerdeClassName());
                }
                if (consumerConf.getReceiverQueueSize() != null) {
                    bldr.setReceiverQueueSize().setValue(consumerConf.getReceiverQueueSize());
                }
                if (consumerConf.getSchemaProperties() != null) {
                    consumerConf.getSchemaProperties().forEach(bldr::putSchemaProperties);

View on GitHub (pinned to 820761864e)

Solutions

1. Validate that each inputSpec's schema type string is a valid schema type and its schema/consumer properties are simple string maps
2. Fix malformed JSON supplied in the schema configuration (quote keys, escape strings, remove comments/trailing commas)
3. Use supported schema types (e.g. JSON, AVRO, STRING, INT64) or fully qualified class names for AutoConsume/Custom schemas
4. Test the schema config in isolation (serialize it with the same ObjectMapper) before submitting the function

Example fix

// before
inputSpecs.put("topic-1", new ConsumerConfig().setSchemaType("AVRO; bad json {")); // fails JSON processing
// after
ConsumerConfig cc = new ConsumerConfig();
cc.setSchemaType("AVRO");
cc.setSchemaProperties(Collections.singletonMap("avroSchemaLocation", "/schemas/t1.avsc"));
inputSpecs.put("topic-1", cc);

// Pre-validate input custom schema config is JSON-friendly
ConsumerConfig cc = inputSpecs.get(topic);
if (cc != null && cc.getSchemaType() != null) {
  try {
    new ObjectMapper().writeValueAsString(cc);
  } catch (JsonProcessingException e) {
    throw new IllegalArgumentException("Input schema config for topic " + topic + " is not JSON-serializable");
  }
}

try {
  FunctionDetails d = FunctionConfigUtils.convert(cfg, pkg);
} catch (IllegalArgumentException e) {
  if (e.getMessage() != null && e.getMessage().startsWith("Incorrect custom schema inputs")) {
    String topic = e.getMessage().replace("Incorrect custom schema inputs,Topic ", "").trim();
    log.error("Fix custom schema input config for topic {}", topic);
  } else {
    throw e;
  }
}

When it happens

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

Common situations: 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.

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

• Incorrect custom schema outputs,Topic %s
• Invalid schema class %s
• Failed to parse tlsFactoryConfig as a JSON object
• Malformed authentication parameters
• JSON Record Builder doesn't support non-JSON record as a fie

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