{"record":{"id":"46afd5594def64da","repo":"apache/seatunnel","slug":"commonerrorcode-unsupported-data-type","errorCode":"CommonErrorCode.UNSUPPORTED_DATA_TYPE","errorMessage":"Unsupported format: %s","messagePattern":"Unsupported format: (.+?)","errorType":"error_code","errorClass":"SeaTunnelJsonFormatException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java","lineNumber":288,"sourceCode":"\n        if (closeFailure != null) {\n            rethrowCloseFailure(closeFailure);\n        }\n    }\n\n    private SerializationSchema createSerializationSchema(\n            SeaTunnelRowType rowType, String format, String delimiter) {\n        if (PulsarSinkOptions.DEFAULT_FORMAT.equals(format)) {\n            return new JsonSerializationSchema(rowType);\n        } else if (PulsarSinkOptions.TEXT_FORMAT.equals(format)) {\n            return TextSerializationSchema.builder()\n                    .seaTunnelRowType(rowType)\n                    .delimiter(delimiter)\n                    .build();\n        } else if (PulsarSinkOptions.AVRO_FORMAT.equals(format)) {\n            return new AvroSerializationSchema(rowType);\n        } else {\n            throw new SeaTunnelJsonFormatException(\n                    CommonErrorCode.UNSUPPORTED_DATA_TYPE, \"Unsupported format: \" + format);\n        }\n    }\n\n    public static SerializationSchema createKeySerializationSchema(\n            List<String> keyFieldNames, SeaTunnelRowType seaTunnelRowType) {\n        if (keyFieldNames == null || keyFieldNames.isEmpty()) {\n            return null;\n        }\n        int[] keyFieldIndexArr = new int[keyFieldNames.size()];\n        SeaTunnelDataType[] keyFieldDataTypeArr = new SeaTunnelDataType[keyFieldNames.size()];\n        for (int i = 0; i < keyFieldNames.size(); i++) {\n            String keyFieldName = keyFieldNames.get(i);\n            int rowFieldIndex = seaTunnelRowType.indexOf(keyFieldName);\n            keyFieldIndexArr[i] = rowFieldIndex;\n            keyFieldDataTypeArr[i] = seaTunnelRowType.getFieldType(rowFieldIndex);\n        }\n        SeaTunnelRowType keyType =","sourceCodeStart":270,"sourceCodeEnd":306,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java#L270-L306","documentation":"PulsarSinkWriter.createSerializationSchema supports a fixed set of value formats; when the configured 'format' matches none of them (and is not the Avro branch), it throws SeaTunnelJsonFormatException with UNSUPPORTED_DATA_TYPE. The format string determines how rows are serialized to bytes before publishing.","triggerScenarios":"Setting sink option format (PulsarSinkOptions format) to a value not in the supported switch (e.g. 'parquet', 'protobuf', or a typo like 'jsonn') in createSerializationSchema.","commonSituations":"Typo in format value; copying a format valid for other connectors (e.g. 'parquet') into the Pulsar sink; case mismatch ('JSON' vs 'json'); upgrading SeaTunnel where an old format alias was removed.","solutions":["Use one of the supported formats, e.g. 'json', 'text', 'canal-json' or 'avro' (check PulsarSinkOptions/constants for the exact accepted values).","Fix casing/spelling of the format value in the config.","Use AvroSerializationSchema by setting format to the Avro option if Avro output is needed."],"exampleFix":"// before\nsink {\n  Pulsar {\n    format = \"parquet\"\n  }\n}\n// after\nsink {\n  Pulsar {\n    format = \"json\"\n  }\n}","handlingStrategy":"validation","validationCode":"Set<String> supported = Set.of(\"json\", \"text\", \"canal-json\", \"avro\");\nif (!supported.contains(format.toLowerCase())) {\n    throw new IllegalArgumentException(\"Unsupported Pulsar sink format: \" + format);\n}","typeGuard":null,"tryCatchPattern":"try {\n    schema = PulsarSinkWriter.createSerializationSchema(rowType, format, delimiter);\n} catch (SeaTunnelJsonFormatException e) {\n    log.error(\"Unsupported format '{}'; use json/text/canal-json/avro\", format);\n    throw e;\n}","preventionTips":["Keep a list of supported Pulsar sink formats handy and validate configs against it.","Avoid copying 'format' values from other connectors (e.g. parquet).","Pin format values in shared config templates."],"tags":["pulsar","serialization","format"],"backgroundTag":"invalid-enum-value","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}