{"record":{"id":"ccea3f5948c22198","repo":"apache/pulsar","slug":"incorrect-custom-schema-outputs-topic-s","errorCode":null,"errorMessage":"Incorrect custom schema outputs,Topic %s ","messagePattern":"Incorrect custom schema outputs,Topic (.+?) ","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java","lineNumber":255,"sourceCode":"            sinkSpec.setSchemaType(functionConfig.getOutputSchemaType());\n        }\n        if (functionConfig.getForwardSourceMessageProperty() == Boolean.TRUE) {\n            sinkSpec.setForwardSourceMessageProperty(functionConfig.getForwardSourceMessageProperty());\n        }\n        if (functionConfig.getCustomSchemaOutputs() != null && functionConfig.getOutput() != null) {\n            String conf = functionConfig.getCustomSchemaOutputs().get(functionConfig.getOutput());\n            try {\n                if (StringUtils.isNotEmpty(conf)) {\n                    ConsumerConfig consumerConfig = OBJECT_MAPPER.readValue(conf, ConsumerConfig.class);\n                    if (consumerConfig.getSchemaProperties() != null) {\n                        consumerConfig.getSchemaProperties().forEach(sinkSpec::putSchemaProperties);\n                    }\n                    if (consumerConfig.getConsumerProperties() != null) {\n                        consumerConfig.getConsumerProperties().forEach(sinkSpec::putConsumerProperties);\n                    }\n                }\n            } catch (JsonProcessingException e) {\n                throw new IllegalArgumentException(\n                        String.format(\"Incorrect custom schema outputs,Topic %s \", functionConfig.getOutput()));\n            }\n        }\n        if (extractedDetails.getTypeArg1() != null) {\n            sinkSpec.setTypeClassName(extractedDetails.getTypeArg1());\n        } else if (StringUtils.isNotEmpty(functionConfig.getOutputTypeClassName())) {\n            sinkSpec.setTypeClassName(functionConfig.getOutputTypeClassName());\n        }\n        if (functionConfig.getProducerConfig() != null) {\n            sinkSpec.setProducerSpec()\n                    .copyFrom(convertProducerConfigToProducerSpec(functionConfig.getProducerConfig()));\n        }\n        if (functionConfig.getBatchBuilder() != null) {\n            ProducerSpec producerSpec;\n            if (sinkSpec.hasProducerSpec()) {\n                producerSpec = sinkSpec.getProducerSpec();\n            } else {\n                producerSpec = sinkSpec.setProducerSpec();","sourceCodeStart":237,"sourceCodeEnd":273,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java#L237-L273","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the output schema type is a supported string (JSON, AVRO, STRING, etc.) and output schema properties are Map<String,String>","Fix malformed JSON in the output schema configuration","Simplify the output schema config to schema type + class name (setOutputSerdeClassName/setOutputTypeClassName) if custom properties are unnecessary","Pre-validate serialization of your schema config with a Jackson ObjectMapper before submission"],"exampleFix":"// before\ncfg.setOutput(\"topic-out\");\ncfg.setOutputSchemaProperties(Map.of(\"schema\", new Object())); // not JSON-serializable -> throws\n// after\ncfg.setOutput(\"topic-out\");\ncfg.setOutputTypeClassName(\"com.example.OutMsg\"); // rely on schema inference instead of broken custom schema","handlingStrategy":"try-catch","validationCode":"// Pre-validate output custom schema config is JSON-friendly\nif (cfg.getOutput() != null && cfg.getOutputSchemaProperties() != null) {\n  try {\n    new ObjectMapper().writeValueAsString(cfg.getOutputSchemaProperties());\n  } catch (JsonProcessingException e) {\n    throw new IllegalArgumentException(\"Output schema properties are not JSON-serializable\");\n  }\n}","typeGuard":null,"tryCatchPattern":"try {\n  FunctionDetails d = FunctionConfigUtils.convert(cfg, pkg);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage() != null && e.getMessage().startsWith(\"Incorrect custom schema outputs\")) {\n    log.error(\"Fix custom schema output config for topic {}\", cfg.getOutput());\n  } else {\n    throw e;\n  }\n}","preventionTips":["Store output schema properties as plain Map<String,String> only","Prefer setOutputTypeClassName/setOutputSerdeClassName over hand-built custom schema JSON","Round-trip the output schema config through Jackson before submitting","Run convert() in CI for function configs"],"tags":["pulsar-functions","schema","json","configuration"],"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-14T00:17:10.932Z"}