apache/beam · error · IllegalArgumentException

Improper schema for Beam record

Error message

Improper schema for Beam record: %s has no row schema to build a Row from.

What it means

When converting a STRUCT-typed field, beamRowFromKafkaStruct needs the Beam FieldType's row schema to build the nested Row. If f.getType().getRowSchema() returns null the Beam field was declared as a generic STRUCT without a logical type/row schema, and the conversion cannot proceed.

Solutions

1. Ensure nested Beam fields are built with Schema.FieldType.row(...) carrying the nested Schema.
2. Upgrade Beam if the schema inference for nested Debezium structs is at fault.
3. Inspect the Beam schema produced by KafkaConnectUtils.beamSchema and fix the field definition at its source.

Example fix

// before
Schema.FieldType.structType() // no row schema attached
// after
Schema.FieldType.row(nestedSchema)

for (Field f : beamSchema.getFields()) {
  if (f.getType().getTypeName() == Schema.TypeName.ROW
      && f.getType().getRowSchema() == null) {
    throw new IllegalStateException("Field " + f.getName() + " lacks row schema");
  }
}

boolean hasRowSchema(Schema.Field f) {
  return f.getType().getTypeName() != Schema.TypeName.ROW
      || f.getType().getRowSchema() != null;
}

When it happens

Trigger: A Beam schema field typed as STRUCT whose FieldType was created without attaching a RowSchema, encountered during beamRowFromKafkaStruct conversion.

Common situations: A bug in upstream schema inference (beamField producing STRUCT without row schema), or custom schema registration where nested rows weren't fully specified.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• UNABLE TO CONVERT FIELD
• Unsupported thrift type code
• A function must be provided to convert the input type into…
• A PValue contained in
• A schema was provided without a data format (or viceversa)…

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

Appendix: source

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

    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:
              rowBuilder = rowBuilder.addValue(structField);
              break;
          }
        }

View on GitHub (pinned to 12126d8942)