apache/pulsar · error · IllegalArgumentException

Output Schema mismatch

Error message

Output Schema mismatch

What it means

The declared output schema type of a function is immutable on update. When the new config provides a non-empty outputSchemaType that differs from the existing config's, validateUpdate throws this error because consumers depend on the published output schema.

Source

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

        if (!newConfig.getInputSpecs().isEmpty()) {
            newConfig.getInputSpecs().forEach((topicName, consumerConfig) -> {
                if (!existingConfig.getInputSpecs().containsKey(topicName)) {
                    throw new IllegalArgumentException("Input Topics cannot be altered");
                }
                if (consumerConfig.isRegexPattern() != existingConfig.getInputSpecs().get(topicName).isRegexPattern()) {
                    throw new IllegalArgumentException(
                            "isRegexPattern for input topic " + topicName + " cannot be altered");
                }
                mergedConfig.getInputSpecs().put(topicName, consumerConfig);
            });
        }
        if (!StringUtils.isEmpty(newConfig.getOutputSerdeClassName()) && !newConfig.getOutputSerdeClassName()
                .equals(existingConfig.getOutputSerdeClassName())) {
            throw new IllegalArgumentException("Output Serde mismatch");
        }
        if (!StringUtils.isEmpty(newConfig.getOutputSchemaType()) && !newConfig.getOutputSchemaType()
                .equals(existingConfig.getOutputSchemaType())) {
            throw new IllegalArgumentException("Output Schema mismatch");
        }
        if (!StringUtils.isEmpty(newConfig.getLogTopic())) {
            mergedConfig.setLogTopic(newConfig.getLogTopic());
        }
        if (newConfig.getProcessingGuarantees() != null && !newConfig.getProcessingGuarantees()
                .equals(existingConfig.getProcessingGuarantees())) {
            throw new IllegalArgumentException("Processing Guarantees cannot be altered");
        }
        if (newConfig.getRetainOrdering() != null && !newConfig.getRetainOrdering()
                .equals(existingConfig.getRetainOrdering())) {
            throw new IllegalArgumentException("Retain Ordering cannot be altered");
        }
        if (newConfig.getRetainKeyOrdering() != null && !newConfig.getRetainKeyOrdering()
                .equals(existingConfig.getRetainKeyOrdering())) {
            throw new IllegalArgumentException("Retain Key Ordering cannot be altered");
        }
        if (!StringUtils.isEmpty(newConfig.getOutput())) {
            mergedConfig.setOutput(newConfig.getOutput());

View on GitHub (pinned to 820761864e)

Solutions

  1. Pass the existing outputSchemaType unchanged, or omit it (empty) in the update.
  2. Delete and recreate the function if the output schema must change.
  3. Ensure any schema-migration path uses a new function (or a compatibility-checked recreate) rather than an in-place update.

Example fix

// before
newConfig.setOutputSchemaType("JSON"); // existing: "AVRO"
// after
newConfig.setOutputSchemaType(""); // keep existing schema on update
// or recreate the function with the new schema type
Defensive patterns

Strategy: validation

Validate before calling

if (newConfig.getOutputSchemaType() != null && !newConfig.getOutputSchemaType().isEmpty()
        && !newConfig.getOutputSchemaType().equals(existingConfig.getOutputSchemaType())) {
    throw new IllegalArgumentException("outputSchemaType is immutable on update");
}

Type guard

boolean outputSchemaUnchanged(FunctionConfig existing, FunctionConfig updated) {
    String n = updated.getOutputSchemaType();
    return n == null || n.isEmpty() || n.equals(existing.getOutputSchemaType());
}

Try / catch

try {
    merged = FunctionConfigUtils.validateUpdate(existing, updated);
} catch (IllegalArgumentException e) {
    if (e.getMessage().equals("Output Schema mismatch")) {
        updated.setOutputSchemaType("");
        merged = FunctionConfigUtils.validateUpdate(existing, updated);
    } else { throw e; }
}

Prevention

When it happens

Trigger: Updating a function where newConfig.getOutputSchemaType() is non-empty and not equal to existingConfig.getOutputSchemaType() (e.g. AVRO -> JSON or a schema class name change).

Common situations: Evolving a function's output schema in place; migrating from serdeClassName-based config to schemaType-based config where one field is set inconsistently; copy-pasted configs from other functions.

Related errors

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