{"record":{"id":"8f2262ebc832a1c8","repo":"apache/beam","slug":"unable-to-convert-field","errorCode":null,"errorMessage":"UNABLE TO CONVERT FIELD ","messagePattern":"UNABLE TO CONVERT FIELD ","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java","lineNumber":120,"sourceCode":"                record.valueSchema(), record.sourceOffset()));\n  }\n\n  public static SourceRecordMapper<Row> beamRowFromSourceRecordFn(final Schema recordSchema) {\n    return new SourceRecordMapper<Row>() {\n      @Override\n      public Row mapSourceRecord(SourceRecord sourceRecord) throws Exception {\n        return beamRowFromKafkaStruct((Struct) sourceRecord.value(), recordSchema);\n      }\n\n      private Row beamRowFromKafkaStruct(Struct kafkaStruct, Schema beamSchema) {\n        Row.Builder rowBuilder = Row.withSchema(beamSchema);\n        for (Schema.Field f : beamSchema.getFields()) {\n          Object structField = kafkaStruct.getWithoutDefault(f.getName());\n          switch (kafkaStruct.schema().field(f.getName()).schema().type()) {\n            case ARRAY:\n            case MAP:\n              // TODO(pabloem): Handle nested structs\n              throw new IllegalArgumentException(\"UNABLE TO CONVERT FIELD \" + f);\n            case STRUCT:\n              Schema fieldSchema = f.getType().getRowSchema();\n              if (fieldSchema == null) {\n                throw new IllegalArgumentException(\n                    String.format(\n                        \"Improper schema for Beam record: %s has no row schema to build a Row from.\",\n                        f.getName()));\n              }\n              if (structField == null) {\n                // If the field is null, then we must add a null field to ensure we encode things\n                // properly.\n                rowBuilder = rowBuilder.addValue(null);\n                break;\n              }\n              rowBuilder =\n                  rowBuilder.addValue(beamRowFromKafkaStruct((Struct) structField, fieldSchema));\n              break;\n            default:","sourceCodeStart":102,"sourceCodeEnd":138,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaConnectUtils.java#L102-L138","documentation":"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.","triggerScenarios":"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().","commonSituations":"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.","solutions":["Upgrade Apache Beam, where nested ARRAY/MAP conversion has been implemented.","Flatten or drop array/map columns in the source table view or via Debezium's column masking/exclude filters.","Convert the field to a serialized STRING before ingestion."],"exampleFix":"// before\ncolumns: my_array_col\n// after (connector config)\n\"column.exclude.list\": \"mydb.mytable.my_array_col\"","handlingStrategy":"validation","validationCode":"for (Field f : beamSchema.getFields()) {\n  Schema.Type t = connectSchema.field(f.getName()).schema().type();\n  if (t == Schema.Type.ARRAY || t == Schema.Type.MAP) {\n    throw new IllegalStateException(\"Nested collection field not supported: \" + f.getName());\n  }\n}","typeGuard":"boolean isConvertible(Struct s, Schema beamSchema) {\n  return beamSchema.getFields().stream()\n      .noneMatch(f -> {\n        Schema.Type t = s.schema().field(f.getName()).schema().type();\n        return t == Schema.Type.ARRAY || t == Schema.Type.MAP;\n      });\n}","tryCatchPattern":"try {\n  Row row = KafkaConnectUtils.beamRowFromKafkaStruct(beamSchema, kafkaStruct);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().startsWith(\"UNABLE TO CONVERT FIELD\")) {\n    LOG.warn(\"Skipping record with unsupported nested field\");\n    return;\n  }\n  throw e;\n}","preventionTips":["Upgrade to a Beam version with nested struct support","Exclude array/map/JSON columns via column.exclude.list","Serialize complex columns to strings upstream"],"tags":["java","schema-conversion","kafka-connect","apache-beam"],"backgroundTag":"unsupported-operation","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"}