{"record":{"id":"0aafff7e1d06d8dd","repo":"apache/seatunnel","slug":"common-unsupported-data-type-commonerrorcodedepre-0aafff","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/source/KafkaSourceConfig.java","lineNumber":450,"sourceCode":"                                \"avro_schema must be configured when strip_schema_registry_header is enabled for avro format\");\n                    }\n                    schema =\n                            new AvroDeserializationSchema(\n                                    catalogTable,\n                                    avroSchema.orElse(null),\n                                    stripAvroSchemaRegistryHeader);\n                    break;\n                case PROTOBUF:\n                    boolean stripSchemaRegistryHeader =\n                            readonlyConfig.get(STRIP_SCHEMA_REGISTRY_HEADER);\n                    if (stripSchemaRegistryHeader) {\n                        schema = new SchemaRegistryAwareProtobufDeserializationSchema(catalogTable);\n                    } else {\n                        schema = new ProtobufDeserializationSchema(catalogTable);\n                    }\n                    break;\n                default:\n                    throw new SeaTunnelJsonFormatException(\n                            CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE,\n                            \"Unsupported format: \" + format);\n            }\n        }\n\n        if (schema instanceof NativeKafkaConnectDeserializationSchema\n                || schema instanceof CompatibleKafkaConnectDeserializationSchema) {\n            return schema;\n        }\n\n        if (!headerFields.isEmpty() && format != MessageFormat.NATIVE) {\n            SeaTunnelRowType baseRowType = (SeaTunnelRowType) schema.getProducedType();\n            SeaTunnelRowType extendedRowType = buildExtendedRowType(baseRowType, headerFields);\n            schema = new KafkaHeadersDeserializationSchema(schema, headerFields, extendedRowType);\n        }\n\n        return new KafkaEventTimeDeserializationSchema(schema);\n    }","sourceCodeStart":432,"sourceCodeEnd":468,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java#L432-L468","documentation":"Kafka source supports a fixed set of value formats (json, canal_json, debezium_json, avro, protobuf, text, etc.). If the configured format does not match any supported case, createDeserializationSchema falls into the default branch and throws SeaTunnelJsonFormatException with UNSUPPORTED_DATA_TYPE.","triggerScenarios":"Setting format to an unlisted string (typo, wrong casing, or a format the connector version does not implement) so the switch's default branch executes.","commonSituations":"Typos like 'JSON' or 'csv'; using a format supported by other connectors but not Kafka; running an older connector version lacking a newer format (e.g. protobuf in old releases).","solutions":["Use one of the supported format values exactly (e.g. json, canal_json, debezium_json, avro, protobuf, text) — check the connector docs for your version.","Fix casing/typos in the format option.","Upgrade the connector if the desired format was added in a later release."],"exampleFix":"// before\nformat = csv   // unsupported for Kafka source\n// after\nformat = json","handlingStrategy":"validation","validationCode":"Set<String> supported = Set.of(\"json\",\"canal_json\",\"debezium_json\",\"avro\",\"protobuf\",\"text\");\nif (!supported.contains(cfg.get(\"format\"))) throw new IllegalArgumentException(\"Unsupported format: \" + cfg.get(\"format\"));","typeGuard":null,"tryCatchPattern":"try { buildSource(cfg); } catch (SeaTunnelJsonFormatException e) { if (e.getMessage().startsWith(\"Unsupported format\")) useDefaultFormat(cfg); else throw e; }","preventionTips":["Copy format values verbatim from the connector docs.","Check the supported list for your connector version before upgrading configs."],"tags":["kafka","format","unsupported","config"],"backgroundTag":"unsupported-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"}