apache/beam · error · IllegalArgumentException

UNABLE TO CONVERT FIELD

Error message

UNABLE TO CONVERT FIELD 

What it means

beamRowFromKafkaStruct converts a Kafka Connect Struct into a Beam Row according to the computed Beam schema. ARRAY and MAP fields are explicitly unimplemented (TODO: nested structs), so encountering one throws this error. It is a known limitation, not data corruption.

Solutions

1. Upgrade Apache Beam, where nested ARRAY/MAP conversion has been implemented.
2. Flatten or drop array/map columns in the source table view or via Debezium's column masking/exclude filters.
3. Convert the field to a serialized STRING before ingestion.

Example fix

// before
columns: my_array_col
// after (connector config)
"column.exclude.list": "mydb.mytable.my_array_col"

for (Field f : beamSchema.getFields()) {
  Schema.Type t = connectSchema.field(f.getName()).schema().type();
  if (t == Schema.Type.ARRAY || t == Schema.Type.MAP) {
    throw new IllegalStateException("Nested collection field not supported: " + f.getName());
  }
}

boolean isConvertible(Struct s, Schema beamSchema) {
  return beamSchema.getFields().stream()
      .noneMatch(f -> {
        Schema.Type t = s.schema().field(f.getName()).schema().type();
        return t == Schema.Type.ARRAY || t == Schema.Type.MAP;
      });
}

try {
  Row row = KafkaConnectUtils.beamRowFromKafkaStruct(beamSchema, kafkaStruct);
} catch (IllegalArgumentException e) {
  if (e.getMessage().startsWith("UNABLE TO CONVERT FIELD")) {
    LOG.warn("Skipping record with unsupported nested field");
    return;
  }
  throw e;
}

When it happens

Trigger: A Debezium record whose Beam schema contains an ARRAY or MAP-typed field, causing the switch in beamRowFromKafkaStruct to hit the ARRAY/MAP case while iterating beamSchema.getFields().

Common situations: Debezium tables with JSON/array columns (e.g. MySQL JSON, Postgres arrays) that map to Connect ARRAY/MAP types; CDC on tables with nested collections.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• Improper schema for Beam record
• Unable to convert Kafka field schema
• Unsupported thrift type code
• A function must be provided to convert the input type into…
• A PValue contained in

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

Appendix: source

Thrown at sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java:120

                record.valueSchema(), record.sourceOffset()));
  }

  public static SourceRecordMapper<Row> beamRowFromSourceRecordFn(final Schema recordSchema) {
    return new SourceRecordMapper<Row>() {
      @Override
      public Row mapSourceRecord(SourceRecord sourceRecord) throws Exception {
        return beamRowFromKafkaStruct((Struct) sourceRecord.value(), recordSchema);
      }

      private Row beamRowFromKafkaStruct(Struct kafkaStruct, Schema beamSchema) {
        Row.Builder rowBuilder = Row.withSchema(beamSchema);
        for (Schema.Field f : beamSchema.getFields()) {
          Object structField = kafkaStruct.getWithoutDefault(f.getName());
          switch (kafkaStruct.schema().field(f.getName()).schema().type()) {
            case ARRAY:
            case MAP:
              // TODO(pabloem): Handle nested structs
              throw new IllegalArgumentException("UNABLE TO CONVERT FIELD " + f);
            case STRUCT:
              Schema fieldSchema = f.getType().getRowSchema();
              if (fieldSchema == null) {
                throw new IllegalArgumentException(
                    String.format(
                        "Improper schema for Beam record: %s has no row schema to build a Row from.",
                        f.getName()));
              }
              if (structField == null) {
                // If the field is null, then we must add a null field to ensure we encode things
                // properly.
                rowBuilder = rowBuilder.addValue(null);
                break;
              }
              rowBuilder =
                  rowBuilder.addValue(beamRowFromKafkaStruct((Struct) structField, fieldSchema));
              break;
            default:

View on GitHub (pinned to 12126d8942)