{"record":{"id":"2bf0eecdd9077187","repo":"apache/beam","slug":"format-s-is-not-supported-supported-formats-are-s","errorCode":null,"errorMessage":"Format %s is not supported. Supported formats are: %s","messagePattern":"Format (.+?) is not supported\\. Supported formats are: (.+?)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java","lineNumber":94,"sourceCode":"  public static final Set<String> SUPPORTED_FORMATS =\n      Sets.newHashSet(SUPPORTED_FORMATS_STR.split(\",\"));\n  public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};\n  public static final TupleTag<KV<byte @Nullable [], byte[]>> OUTPUT_TAG =\n      new TupleTag<KV<byte @Nullable [], byte[]>>() {};\n  public static final TupleTag<KV<byte @Nullable [], GenericRecord>> RECORD_OUTPUT_TAG =\n      new TupleTag<KV<byte @Nullable [], GenericRecord>>() {};\n  private static final Logger LOG =\n      LoggerFactory.getLogger(KafkaWriteSchemaTransformProvider.class);\n\n  @Override\n  protected Class<KafkaWriteSchemaTransformConfiguration> configurationClass() {\n    return KafkaWriteSchemaTransformConfiguration.class;\n  }\n\n  @Override\n  protected SchemaTransform from(KafkaWriteSchemaTransformConfiguration configuration) {\n    if (!SUPPORTED_FORMATS.contains(configuration.getFormat())) {\n      throw new IllegalArgumentException(\n          \"Format \"\n              + configuration.getFormat()\n              + \" is not supported. \"\n              + \"Supported formats are: \"\n              + String.join(\", \", SUPPORTED_FORMATS));\n    }\n    return new KafkaWriteSchemaTransform(configuration);\n  }\n\n  static final class KafkaWriteSchemaTransform extends SchemaTransform implements Serializable {\n    final KafkaWriteSchemaTransformConfiguration configuration;\n\n    KafkaWriteSchemaTransform(KafkaWriteSchemaTransformConfiguration configuration) {\n      this.configuration = configuration;\n    }\n\n    Row getConfigurationRow() {\n      try {","sourceCodeStart":76,"sourceCodeEnd":112,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java#L76-L112","documentation":"KafkaWriteSchemaTransformProvider.from() validates the requested output format against SUPPORTED_FORMATS and throws IllegalArgumentException naming the unsupported format and the supported list. The format string chosen in KafkaWriteSchemaTransformConfiguration drives how Rows are serialized to bytes for the Kafka sink.","triggerScenarios":"Building the transform with KafkaWriteSchemaTransformConfiguration.builder().setFormat(<unknown>).build() where format is not in SUPPORTED_FORMATS (e.g. 'avro', 'json' lowercase, or a typo like 'RAWW'), typically via from() in the provider or through a YAML/schema-transform pipeline config.","commonSituations":"YAML pipeline configs with case-sensitive format strings; users writing 'AVRO' when AVRO needs toRawBytes handling and isn't in the supported list; drift after upgrading Beam when format set was changed.","solutions":["Use exactly one of the supported formats listed in the error message (e.g. RAW, JSON, PROTO, AVRO variants per version).","Check SUPPORTED_FORMATS in KafkaWriteSchemaTransformProvider for the installed Beam version.","Fix casing/typos in the format field of your config/YAML.","For Avro/Proto, ensure you also supply the required schema fields (schema/fileDescriptorPath/messageName) as applicable."],"exampleFix":"// before\nKafkaWriteSchemaTransformConfiguration.builder().setFormat(\"avro\").build();\n// after\nKafkaWriteSchemaTransformConfiguration.builder().setFormat(\"AVRO\").build(); // must be a member of SUPPORTED_FORMATS","handlingStrategy":"validation","validationCode":"if (!Arrays.asList(\"RAW\",\"JSON\",\"PROTO\",\"AVRO\").contains(format)) throw new IllegalArgumentException(format);","typeGuard":null,"tryCatchPattern":"try { provider.from(cfg); } catch (IllegalArgumentException e) { LOG.error(\"Invalid format: {}\", cfg.getFormat(), e); }","preventionTips":["Copy format strings from SUPPORTED_FORMATS exactly (case-sensitive)","Pin the Beam version and re-check supported formats after upgrades"],"tags":["kafka","schema-transform","illegal-argument","format-validation"],"backgroundTag":"invalid-enum-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}