{"record":{"id":"9c023c504b96f53d","repo":"apache/beam","slug":"expecting-exactly-one-field-found-s","errorCode":null,"errorMessage":"Expecting exactly one field, found %s","messagePattern":"Expecting exactly one field, found (.+?)","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":203,"sourceCode":"          Schema errorSchema,\n          boolean handleErrors) {\n        super(name, toGenericRecordsFn, errorSchema, handleErrors, RECORD_OUTPUT_TAG);\n      }\n    }\n\n    @SuppressWarnings({\n      \"nullness\" // TODO(https://github.com/apache/beam/issues/20497)\n    })\n    @Override\n    public PCollectionRowTuple expand(PCollectionRowTuple input) {\n      Schema inputSchema = input.get(\"input\").getSchema();\n      org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(inputSchema);\n      final SerializableFunction<Row, byte[]> toBytesFn;\n      SerializableFunction<Row, GenericRecord> toGenericRecordsFn = null;\n      if (configuration.getFormat().equals(\"RAW\")) {\n        int numFields = inputSchema.getFields().size();\n        if (numFields != 1) {\n          throw new IllegalArgumentException(\"Expecting exactly one field, found \" + numFields);\n        }\n        if (!inputSchema.getField(0).getType().equals(Schema.FieldType.BYTES)) {\n          throw new IllegalArgumentException(\n              \"The input schema must have exactly one field of type byte.\");\n        }\n        toBytesFn = getRowToRawBytesFunction(inputSchema.getField(0).getName());\n      } else if (configuration.getFormat().equals(\"JSON\")) {\n        toBytesFn = JsonUtils.getRowToJsonBytesFunction(inputSchema);\n      } else if (configuration.getFormat().equals(\"PROTO\")) {\n        String descriptorPath = configuration.getFileDescriptorPath();\n        String schema = configuration.getSchema();\n        String messageName = configuration.getMessageName();\n        if (messageName == null) {\n          throw new IllegalArgumentException(\"Expecting messageName to be non-null.\");\n        }\n        if (descriptorPath != null && schema != null) {\n          throw new IllegalArgumentException(\n              \"You must include a descriptorPath or a proto Schema but not both.\");","sourceCodeStart":185,"sourceCodeEnd":221,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java#L185-L221","documentation":"In expand(), when the configured format is RAW, the input PCollection's schema must contain exactly one field, since RAW writes the single field's bytes verbatim to Kafka. The provider throws IllegalArgumentException stating how many fields were actually found.","triggerScenarios":"Applying the Kafka write SchemaTransform with format RAW on a PCollection whose schema has 0 or 2+ fields.","commonSituations":"Pipelines emitting multi-field Rows (e.g. key+value) directly to a RAW-format Kafka sink; forgetting to pre-map the record to a single BYTES field.","solutions":["Use a different format (JSON/AVRO/PROTO) that supports multi-field schemas.","Map/combine your Row into a single BYTES field before the sink (e.g. serialize your payload yourself).","Verify the PCollection's schema via pc.getSchema() before applying the transform."],"exampleFix":"// before\nrows.apply(\"write\", kafkaWrite.withFormat(\"RAW\")); // schema has 3 fields\n// after\nrows.apply(MapElements.into(Schema.FieldType.BYTES).via(row -> serialize(row)))\n    .setSchema(Schema.of(Schema.Field.of(\"payload\", Schema.FieldType.BYTES)))\n    .apply(\"write\", kafkaWrite.withFormat(\"RAW\"));","handlingStrategy":"validation","validationCode":"if (format.equals(\"RAW\") && pc.getSchema().getFields().size() != 1) { throw new IllegalArgumentException(\"RAW format requires exactly one field\"); }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Check PCollection schema before applying RAW sink","Choose JSON/AVRO/PROTO for multi-field records"],"tags":["kafka","schema-transform","raw-format","schema-validation"],"backgroundTag":"schema-validation-failed","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"}