apache/seatunnel · error · SeaTunnelJsonFormatException

COMMON_UNSUPPORTED_DATA_TYPE (CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE)

COMMON_UNSUPPORTED_DATA_TYPE (CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE)

Error message

Unsupported format: %s

What it means

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.

Solutions

  1. 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.
  2. Fix casing/typos in the format option.
  3. Upgrade the connector if the desired format was added in a later release.

Example fix

// before
format = csv   // unsupported for Kafka source
// after
format = json
Defensive patterns

Strategy: validation

Validate before calling

Set<String> supported = Set.of("json","canal_json","debezium_json","avro","protobuf","text");
if (!supported.contains(cfg.get("format"))) throw new IllegalArgumentException("Unsupported format: " + cfg.get("format"));

Try / catch

try { buildSource(cfg); } catch (SeaTunnelJsonFormatException e) { if (e.getMessage().startsWith("Unsupported format")) useDefaultFormat(cfg); else throw e; }

Prevention

When it happens

Trigger: 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.

Common situations: 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).

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/0aafff7e1d06d8dd. Report an issue: GitHub.

Appendix: source

Thrown at seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java:450

                                "avro_schema must be configured when strip_schema_registry_header is enabled for avro format");
                    }
                    schema =
                            new AvroDeserializationSchema(
                                    catalogTable,
                                    avroSchema.orElse(null),
                                    stripAvroSchemaRegistryHeader);
                    break;
                case PROTOBUF:
                    boolean stripSchemaRegistryHeader =
                            readonlyConfig.get(STRIP_SCHEMA_REGISTRY_HEADER);
                    if (stripSchemaRegistryHeader) {
                        schema = new SchemaRegistryAwareProtobufDeserializationSchema(catalogTable);
                    } else {
                        schema = new ProtobufDeserializationSchema(catalogTable);
                    }
                    break;
                default:
                    throw new SeaTunnelJsonFormatException(
                            CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE,
                            "Unsupported format: " + format);
            }
        }

        if (schema instanceof NativeKafkaConnectDeserializationSchema
                || schema instanceof CompatibleKafkaConnectDeserializationSchema) {
            return schema;
        }

        if (!headerFields.isEmpty() && format != MessageFormat.NATIVE) {
            SeaTunnelRowType baseRowType = (SeaTunnelRowType) schema.getProducedType();
            SeaTunnelRowType extendedRowType = buildExtendedRowType(baseRowType, headerFields);
            schema = new KafkaHeadersDeserializationSchema(schema, headerFields, extendedRowType);
        }

        return new KafkaEventTimeDeserializationSchema(schema);
    }

View on GitHub (pinned to cf67b549a7)