apache/beam · error · IllegalArgumentException

Unexpected field '${fieldName}' in top level schema for Pubs

Error message

Unexpected field '${fieldName}' in top level schema for Pubsub message. Top level schema should only contain 'timestamp', 'attributes', and 'payload' fields

What it means

PubsubMessageToRow supports nested-schema mode where the top-level Row schema may only have the fields 'timestamp', 'attributes', and 'payload'. getValueForFieldNestedSchema hits an unknown field name and throws IllegalArgumentException, listing the only allowed top-level fields.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageToRow.java:236

          (k, v) -> rows.add(Row.withSchema(ATTRIBUTE_ARRAY_ENTRY_SCHEMA).attachValues(k, v)));
      return rows.build();
    }

    /** Get the value for a field int the order they're specified in the nested schema. */
    private @Nullable Object getValueForFieldNestedSchema(
        Schema.Field field,
        Instant timestamp,
        @Nullable Map<String, String> attributeMap,
        byte[] payload) {
      switch (field.getName()) {
        case TIMESTAMP_FIELD:
          return timestamp;
        case ATTRIBUTES_FIELD:
          return handleAttributes(attributeMap);
        case PAYLOAD_FIELD:
          return maybeDeserialize(payload);
        default:
          throw new IllegalArgumentException(
              "Unexpected field '"
                  + field.getName()
                  + "' in top level schema"
                  + " for Pubsub message. Top level schema should only contain "
                  + "'timestamp', 'attributes', and 'payload' fields");
      }
    }

    @ProcessElement
    public void processElement(
        @Element PubsubMessage element, @Timestamp Instant timestamp, MultiOutputReceiver o) {
      try {
        List<@Nullable Object> values =
            messageSchema.getFields().stream()
                .<@Nullable Object>map(
                    field ->
                        getValueForFieldNestedSchema(
                            field, timestamp, element.getAttributeMap(), element.getPayload()))

View on GitHub (pinned to 12126d8942)

Solutions

1. Restructure the top-level schema to contain only 'timestamp', 'attributes', and 'payload' fields; put custom fields inside 'payload'.
2. If you intended a flat mapping of attributes/payload to custom fields, use the flat schema mode (attribute/payload attribute configuration) instead of nested mode.
3. Fix field-name typos or casing (e.g. 'Timestamp' -> 'timestamp').

Example fix

// before
Schema.of(Field.of("event_time", ...), Field.of("attributes", ...)) // unknown top-level field
// after
Schema.of(
  Field.of("timestamp", ...),
  Field.of("attributes", ...),
  Field.of("payload", Schema.of(Field.of("event_time", ...))))

java.util.Set<String> allowed = new java.util.HashSet<>(java.util.Arrays.asList("timestamp", "attributes", "payload"));
for (Schema.Field f : schema.getFields()) {
  if (!allowed.contains(f.getName())) throw new IllegalArgumentException("Top-level field not allowed: " + f.getName());
}

boolean isNestedTopLevelField = java.util.Arrays.asList("timestamp","attributes","payload").contains(fieldName);

try {
  Row value = values(message, schema);
} catch (IllegalArgumentException e) {
  // fix schema definition; log field name from message
}

When it happens

Trigger: Supplying a top-level schema (via PubsubMessageToRow / PubsubIO.readMessagesWithAttributes with a schema) containing any field other than timestamp/attributes/payload in nested (non-flat) mode.

Common situations: Users define a schema where the message body fields are placed at the top level instead of nested under 'payload', or they rename the standard fields, or they confuse flat vs nested schema modes.

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

• Cannot merge schemas with different numbers of fields. schem
• Unable to generate coder for schema {schema}
• Expecting exactly one field, found
• The input schema must have exactly one field of type byte.
• Cannot merge two types: +fieldType1.getTypeName()+ and +fiel

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