{"record":{"id":"666f261eabc352d8","repo":"apache/beam","slug":"the-input-schema-must-have-exactly-one-field-of-type-byte-666f26","errorCode":null,"errorMessage":"The input schema must have exactly one field of type byte.","messagePattern":"The input schema must have exactly one field of type byte\\.","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":206,"sourceCode":"      }\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.\");\n        } else if (descriptorPath != null) {\n          toBytesFn = ProtoByteUtils.getRowToProtoBytes(descriptorPath, messageName);\n        } else if (schema != null) {","sourceCodeStart":188,"sourceCodeEnd":224,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaWriteSchemaTransformProvider.java#L188-L224","documentation":"Companion check to the RAW format validation in expand(): the single input schema field must be of type BYTES, because RAW mode writes the field's byte content directly to Kafka. Otherwise IllegalArgumentException is thrown.","triggerScenarios":"Format RAW with exactly one field whose FieldType is not BYTES (e.g. STRING, INT64).","commonSituations":"Emitting a single string field expecting the sink to encode it; using a numeric timestamp field as the payload.","solutions":["Convert the field to bytes before the sink (String.getBytes or explicit serialization) and declare it as BYTES in the schema.","Switch to JSON format which handles non-bytes single fields.","Adjust the PCollection schema so the lone field is FieldType.BYTES."],"exampleFix":"// before\nSchema schema = Schema.of(Schema.Field.of(\"message\", Schema.FieldType.STRING));\n// after\nSchema schema = Schema.of(Schema.Field.of(\"message\", Schema.FieldType.BYTES));\n// and convert: bytes b = row.getString(\"message\").getBytes(StandardCharsets.UTF_8);","handlingStrategy":"validation","validationCode":"if (format.equals(\"RAW\") && !pc.getSchema().getField(0).getType().equals(Schema.FieldType.BYTES)) { throw new IllegalArgumentException(\"RAW field must be BYTES\"); }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Encode payloads to bytes before the sink","Declare schema fields explicitly with FieldType.BYTES"],"tags":["kafka","schema-transform","raw-format","type-mismatch"],"backgroundTag":"type-mismatch","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"}