{"record":{"id":"42dd6ff8fcd484ad","repo":"apache/beam","slug":"expected-input-schema-with-a-data-bytes-field-a-payload","errorCode":null,"errorMessage":"Expected input schema with a 'data' (BYTES) field, a 'payload' (BYTES/STRING) field, or a single bytes/string field, but got: ","messagePattern":"Expected input schema with a 'data' \\(BYTES\\) field, a 'payload' \\(BYTES/STRING\\) field, or a single bytes/string field, but got: ","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisWriteSchemaTransformProvider.java","lineNumber":270,"sourceCode":"      return PCollectionRowTuple.empty(inputRows.getPipeline());\n    }\n\n    private static int resolveDataFieldIndex(Schema schema) {\n      if (schema.hasField(\"data\")\n          && schema.getField(\"data\").getType().equals(Schema.FieldType.BYTES)) {\n        return schema.indexOf(\"data\");\n      }\n      if (schema.hasField(\"payload\")\n          && (schema.getField(\"payload\").getType().equals(Schema.FieldType.BYTES)\n              || schema.getField(\"payload\").getType().equals(Schema.FieldType.STRING))) {\n        return schema.indexOf(\"payload\");\n      }\n      if (schema.getFieldCount() == 1\n          && (schema.getField(0).getType().equals(Schema.FieldType.BYTES)\n              || schema.getField(0).getType().equals(Schema.FieldType.STRING))) {\n        return 0;\n      }\n      throw new IllegalArgumentException(\n          \"Expected input schema with a 'data' (BYTES) field, a 'payload' \"\n              + \"(BYTES/STRING) field, or a single bytes/string field, but got: \"\n              + schema);\n    }\n\n    private static boolean isStringField(Schema schema, int index) {\n      return schema.getField(index).getType().equals(Schema.FieldType.STRING);\n    }\n  }\n}\n","sourceCodeStart":252,"sourceCodeEnd":281,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisWriteSchemaTransformProvider.java#L252-L281","documentation":"KinesisWriteSchemaTransformProvider must know which input schema field contains the record bytes to send to Kinesis. resolveDataFieldIndex accepts a 'data' BYTES field, a 'payload' BYTES or STRING field, or a schema with exactly one BYTES/STRING field. Any other schema shape throws an IllegalArgumentException including the offending schema.","triggerScenarios":"Feeding the Kinesis write transform a PCollection whose schema has multiple fields without 'data'/'payload', or whose candidate field is another type (INT64, ROW, ARRAY, MAP).","commonSituations":"Passing a table row / JSON-derived schema with many columns; renamed field that is neither 'data' nor 'payload'; a single numeric field; upstream transform output changed shape after refactor.","solutions":["Rename the bytes column to 'data' (BYTES) or 'payload' (BYTES/STRING) in the upstream schema","Select or reshape with a map to a single BYTES field before the sink","Serialize complex rows explicitly (e.g. via JsonToBytes or Row serialization) so the input has one BYTES field"],"exampleFix":"// before\nrows.apply(\"write\", KinesisWriteSchemaTransformProvider...) // schema: id STRING, body STRING\n// after\nrows.apply(MapElements.into(Schema.FieldType.BYTES)... // or rename 'body' to 'payload'\n    .then(p -> p.apply(\"kinesis write\", ...))","handlingStrategy":"validation","validationCode":"Schema s = input.getSchema();\nboolean ok = (s.getField(\"data\") != null && s.getField(\"data\").getType().equals(Schema.FieldType.BYTES))\n || (s.getField(\"payload\") != null && (s.getField(\"payload\").getType().equals(Schema.FieldType.BYTES) || s.getField(\"payload\").getType().equals(Schema.FieldType.STRING)))\n || (s.getFieldCount()==1 && (s.getField(0).getType().equals(Schema.FieldType.BYTES) || s.getField(0).getType().equals(Schema.FieldType.STRING)));","typeGuard":"boolean hasWritableField(Schema s){ try { return findDataFieldIndex(s) >= 0; } catch (IllegalArgumentException e){ return false; } }","tryCatchPattern":"try { p.apply(Kinesis write transform); } catch (IllegalArgumentException e) { reshape/rename the input schema field to 'data' or 'payload'; rerun; }","preventionTips":["Name the payload column 'data' (BYTES) or 'payload' (BYTES/STRING)","Serialize complex rows to a single BYTES field before the sink","Log the input schema during pipeline development to catch shape drift early"],"tags":["kinesis","schema","bytes","schema-transform"],"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"}