apache/beam · error · IllegalArgumentException

Format is not supported. Supported formats are

Error message

Format %s is not supported. Supported formats are: %s

What it means

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.

Solutions

1. Use exactly one of the supported formats listed in the error message (e.g. RAW, JSON, PROTO, AVRO variants per version).
2. Check SUPPORTED_FORMATS in KafkaWriteSchemaTransformProvider for the installed Beam version.
3. Fix casing/typos in the format field of your config/YAML.
4. For Avro/Proto, ensure you also supply the required schema fields (schema/fileDescriptorPath/messageName) as applicable.

Example fix

// before
KafkaWriteSchemaTransformConfiguration.builder().setFormat("avro").build();
// after
KafkaWriteSchemaTransformConfiguration.builder().setFormat("AVRO").build(); // must be a member of SUPPORTED_FORMATS

if (!Arrays.asList("RAW","JSON","PROTO","AVRO").contains(format)) throw new IllegalArgumentException(format);

try { provider.from(cfg); } catch (IllegalArgumentException e) { LOG.error("Invalid format: {}", cfg.getFormat(), e); }

When it happens

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

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

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Expecting exactly one field, found
• Expecting messageName to be non-null.
• Table reference is not in…
• The input schema must have exactly one field of type byte.
• 2xx codes should not be exceptions. Got status code

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/2bf0eecdd9077187. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java:94

  public static final Set<String> SUPPORTED_FORMATS =
      Sets.newHashSet(SUPPORTED_FORMATS_STR.split(","));
  public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
  public static final TupleTag<KV<byte @Nullable [], byte[]>> OUTPUT_TAG =
      new TupleTag<KV<byte @Nullable [], byte[]>>() {};
  public static final TupleTag<KV<byte @Nullable [], GenericRecord>> RECORD_OUTPUT_TAG =
      new TupleTag<KV<byte @Nullable [], GenericRecord>>() {};
  private static final Logger LOG =
      LoggerFactory.getLogger(KafkaWriteSchemaTransformProvider.class);

  @Override
  protected Class<KafkaWriteSchemaTransformConfiguration> configurationClass() {
    return KafkaWriteSchemaTransformConfiguration.class;
  }

  @Override
  protected SchemaTransform from(KafkaWriteSchemaTransformConfiguration configuration) {
    if (!SUPPORTED_FORMATS.contains(configuration.getFormat())) {
      throw new IllegalArgumentException(
          "Format "
              + configuration.getFormat()
              + " is not supported. "
              + "Supported formats are: "
              + String.join(", ", SUPPORTED_FORMATS));
    }
    return new KafkaWriteSchemaTransform(configuration);
  }

  static final class KafkaWriteSchemaTransform extends SchemaTransform implements Serializable {
    final KafkaWriteSchemaTransformConfiguration configuration;

    KafkaWriteSchemaTransform(KafkaWriteSchemaTransformConfiguration configuration) {
      this.configuration = configuration;
    }

    Row getConfigurationRow() {
      try {

View on GitHub (pinned to 12126d8942)