apache/beam · error · IllegalArgumentException

Expected input schema with a 'data' (BYTES) field, a…

Error message

Expected input schema with a 'data' (BYTES) field, a 'payload' (BYTES/STRING) field, or a single bytes/string field, but got: 

What it means

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.

Solutions

1. Rename the bytes column to 'data' (BYTES) or 'payload' (BYTES/STRING) in the upstream schema
2. Select or reshape with a map to a single BYTES field before the sink
3. Serialize complex rows explicitly (e.g. via JsonToBytes or Row serialization) so the input has one BYTES field

Example fix

// before
rows.apply("write", KinesisWriteSchemaTransformProvider...) // schema: id STRING, body STRING
// after
rows.apply(MapElements.into(Schema.FieldType.BYTES)... // or rename 'body' to 'payload'
    .then(p -> p.apply("kinesis write", ...))

Schema s = input.getSchema();
boolean ok = (s.getField("data") != null && s.getField("data").getType().equals(Schema.FieldType.BYTES))
 || (s.getField("payload") != null && (s.getField("payload").getType().equals(Schema.FieldType.BYTES) || s.getField("payload").getType().equals(Schema.FieldType.STRING)))
 || (s.getFieldCount()==1 && (s.getField(0).getType().equals(Schema.FieldType.BYTES) || s.getField(0).getType().equals(Schema.FieldType.STRING)));

boolean hasWritableField(Schema s){ try { return findDataFieldIndex(s) >= 0; } catch (IllegalArgumentException e){ return false; } }

try { p.apply(Kinesis write transform); } catch (IllegalArgumentException e) { reshape/rename the input schema field to 'data' or 'payload'; rerun; }

When it happens

Trigger: 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).

Common situations: 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.

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

• Service endpoint must be a URI, got
• Service endpoint must be a URI, got
• A schema is required to write non-schema'd data.
• All dicts in batch must have the same keys. extra keys
• An explicit schema is required to write non-schema'd…

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

Appendix: source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisWriteSchemaTransformProvider.java:270

      return PCollectionRowTuple.empty(inputRows.getPipeline());
    }

    private static int resolveDataFieldIndex(Schema schema) {
      if (schema.hasField("data")
          && schema.getField("data").getType().equals(Schema.FieldType.BYTES)) {
        return schema.indexOf("data");
      }
      if (schema.hasField("payload")
          && (schema.getField("payload").getType().equals(Schema.FieldType.BYTES)
              || schema.getField("payload").getType().equals(Schema.FieldType.STRING))) {
        return schema.indexOf("payload");
      }
      if (schema.getFieldCount() == 1
          && (schema.getField(0).getType().equals(Schema.FieldType.BYTES)
              || schema.getField(0).getType().equals(Schema.FieldType.STRING))) {
        return 0;
      }
      throw new IllegalArgumentException(
          "Expected input schema with a 'data' (BYTES) field, a 'payload' "
              + "(BYTES/STRING) field, or a single bytes/string field, but got: "
              + schema);
    }

    private static boolean isStringField(Schema schema, int index) {
      return schema.getField(index).getType().equals(Schema.FieldType.STRING);
    }
  }
}

View on GitHub (pinned to 12126d8942)