{"record":{"id":"63e3952790bd729b","repo":"apache/seatunnel","slug":"common-unsupported-data-type-commonerrorcodedepre","errorCode":"COMMON_UNSUPPORTED_DATA_TYPE (CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE)","errorMessage":"Unsupported format: %s","messagePattern":"Unsupported format: (.+?)","errorType":"error_code","errorClass":"SeaTunnelJsonFormatException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java","lineNumber":438,"sourceCode":"            case CANAL_JSON:\n                return new CanalJsonSerializationSchema(rowType);\n            case OGG_JSON:\n                return new OggJsonSerializationSchema(rowType);\n            case DEBEZIUM_JSON:\n                return new DebeziumJsonSerializationSchema(rowType);\n            case MAXWELL_JSON:\n                return new MaxWellJsonSerializationSchema(rowType);\n            case COMPATIBLE_DEBEZIUM_JSON:\n                return new CompatibleDebeziumJsonSerializationSchema(rowType, isKey);\n            case AVRO:\n                return new AvroSerializationSchema(rowType);\n            case PROTOBUF:\n                String protobufMessageName = pluginConfig.get(PROTOBUF_MESSAGE_NAME);\n                String protobufSchema = pluginConfig.get(PROTOBUF_SCHEMA);\n                return new ProtobufSerializationSchema(\n                        rowType, protobufMessageName, protobufSchema);\n            default:\n                throw new SeaTunnelJsonFormatException(\n                        CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE,\n                        \"Unsupported format: \" + format);\n        }\n    }\n\n    private static Iterable<Header> convertToKafkaHeaders(Map<String, String> headersMap) {\n        if (MapUtils.isEmpty(headersMap)) {\n            return null;\n        }\n        RecordHeaders kafkaHeaders = new RecordHeaders();\n        for (Map.Entry<String, String> entry : headersMap.entrySet()) {\n            kafkaHeaders.add(\n                    new RecordHeader(\n                            entry.getKey(), entry.getValue().getBytes(StandardCharsets.UTF_8)));\n        }\n        return kafkaHeaders;\n    }\n}","sourceCodeStart":420,"sourceCodeEnd":456,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java#L420-L456","documentation":"DefaultSeaTunnelRowSerializer.createSerializationSchema switches on the configured 'format' option and throws SeaTunnelJsonFormatException with code UNSUPPORTED_DATA_TYPE for any format outside the supported set (JSON, Text, CANAL_JSON, DEBEZIUM_JSON, AVRO, PROTOBUF, etc.). Despite the exception class name, it is a format-dispatch failure, not a JSON parsing failure.","triggerScenarios":"Setting the Kafka sink 'format' option to an unsupported value (e.g. 'csv', 'parquet', a misspelled name like 'jsn', or a format the connector has no case for), so the default branch executes.","commonSituations":"Typos in the format config value; using a format supported only by source (not sink) serialization; copy-pasting configs between connectors with different supported format lists.","solutions":["Change the format option to a supported sink format: json, text, canal_json, debezium_json, avro, or protobuf.","Check the connector documentation for the exact list of supported serialization formats and their exact spelling.","If you need an unsupported format, add/enable a format plugin and extend the switch in createSerializationSchema."],"exampleFix":"# before\nformat = \"csv\"   # not a supported Kafka sink format\n# after\nformat = \"json\"","handlingStrategy":"validation","validationCode":"java.util.Set<String> supported = java.util.Set.of(\n    \"json\", \"text\", \"canal_json\", \"debezium_json\", \"avro\", \"protobuf\");\nString format = pluginConfig.get(\"format\");\nif (format == null || !supported.contains(format.toLowerCase())) {\n    throw new IllegalArgumentException(\"unsupported kafka sink format: \" + format);\n}","typeGuard":null,"tryCatchPattern":"try {\n    schema = DefaultSeaTunnelRowSerializer.createSerializationSchema(rowType, pluginConfig);\n} catch (SeaTunnelJsonFormatException e) {\n    // log valid format values and rethrow with guidance\n    throw e;\n}","preventionTips":["Copy format values from the official connector docs verbatim.","Validate the HOCON config (format is an enum-like option) before job submission.","Keep source/sink format lists separate — a source-supported format may not be sink-supported."],"tags":["kafka","config","serialization","unsupported-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-23T08:17:48.524Z"}