apache/pulsar · error · IllegalArgumentException

Incorrect custom schema outputs,Topic %s

Error message

Incorrect custom schema outputs,Topic %s 

What it means

Symmetric to the input case: while converting FunctionConfig to FunctionDetails, the output topic's custom schema configuration is serialized to JSON to populate the SinkSpec. If JsonProcessingException is thrown while serializing the custom schema output info, the code rethrows this IllegalArgumentException naming the output topic.

Source

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

            sinkSpec.setSchemaType(functionConfig.getOutputSchemaType());
        }
        if (functionConfig.getForwardSourceMessageProperty() == Boolean.TRUE) {
            sinkSpec.setForwardSourceMessageProperty(functionConfig.getForwardSourceMessageProperty());
        }
        if (functionConfig.getCustomSchemaOutputs() != null && functionConfig.getOutput() != null) {
            String conf = functionConfig.getCustomSchemaOutputs().get(functionConfig.getOutput());
            try {
                if (StringUtils.isNotEmpty(conf)) {
                    ConsumerConfig consumerConfig = OBJECT_MAPPER.readValue(conf, ConsumerConfig.class);
                    if (consumerConfig.getSchemaProperties() != null) {
                        consumerConfig.getSchemaProperties().forEach(sinkSpec::putSchemaProperties);
                    }
                    if (consumerConfig.getConsumerProperties() != null) {
                        consumerConfig.getConsumerProperties().forEach(sinkSpec::putConsumerProperties);
                    }
                }
            } catch (JsonProcessingException e) {
                throw new IllegalArgumentException(
                        String.format("Incorrect custom schema outputs,Topic %s ", functionConfig.getOutput()));
            }
        }
        if (extractedDetails.getTypeArg1() != null) {
            sinkSpec.setTypeClassName(extractedDetails.getTypeArg1());
        } else if (StringUtils.isNotEmpty(functionConfig.getOutputTypeClassName())) {
            sinkSpec.setTypeClassName(functionConfig.getOutputTypeClassName());
        }
        if (functionConfig.getProducerConfig() != null) {
            sinkSpec.setProducerSpec()
                    .copyFrom(convertProducerConfigToProducerSpec(functionConfig.getProducerConfig()));
        }
        if (functionConfig.getBatchBuilder() != null) {
            ProducerSpec producerSpec;
            if (sinkSpec.hasProducerSpec()) {
                producerSpec = sinkSpec.getProducerSpec();
            } else {
                producerSpec = sinkSpec.setProducerSpec();

View on GitHub (pinned to 820761864e)

Solutions

1. Ensure the output schema type is a supported string (JSON, AVRO, STRING, etc.) and output schema properties are Map<String,String>
2. Fix malformed JSON in the output schema configuration
3. Simplify the output schema config to schema type + class name (setOutputSerdeClassName/setOutputTypeClassName) if custom properties are unnecessary
4. Pre-validate serialization of your schema config with a Jackson ObjectMapper before submission

Example fix

// before
cfg.setOutput("topic-out");
cfg.setOutputSchemaProperties(Map.of("schema", new Object())); // not JSON-serializable -> throws
// after
cfg.setOutput("topic-out");
cfg.setOutputTypeClassName("com.example.OutMsg"); // rely on schema inference instead of broken custom schema

// Pre-validate output custom schema config is JSON-friendly
if (cfg.getOutput() != null && cfg.getOutputSchemaProperties() != null) {
  try {
    new ObjectMapper().writeValueAsString(cfg.getOutputSchemaProperties());
  } catch (JsonProcessingException e) {
    throw new IllegalArgumentException("Output schema properties are not JSON-serializable");
  }
}

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

When it happens

Trigger: Configuring a custom/complex output schema on functionConfig.getOutput() whose schema type or properties cannot be JSON-serialized — e.g. an object placed in outputSchemaProperties that the ObjectMapper cannot handle, or a malformed schema definition string on the output topic.

Common situations: Setting outputSchemaType/properties programmatically with non-JSON-serializable objects (Date, streams, nested POJOs without getters); hand-written schema JSON with syntax errors in the output config; framework code auto-generating output schema metadata 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 inputs,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/ccea3f5948c22198. Report an issue: GitHub.