apache/beam · error · IllegalArgumentException

Type of @Element must match the DoFn type

Error message

Type of @Element must match the DoFn type

What it means

When a DoFn's @ProcessElement declares schema element parameters (e.g. @SchemaElementParameter annotations on @Element), the input PCollection must have a schema so Beam can map schema fields onto the element type. getDoFnSchemaInformation throws IllegalArgumentException if schema parameters are present but the input PCollection has no schema.

Solutions

1. Make the input PCollection schema-bearing: use a schema-registered type (POJO with @DefaultSchema, Avro record, or Row) via PCollection.setSchema(...)/beam converters (SetSchema/Convert.to(Row.class))
2. Remove the schema element parameters from the @Element parameter if schema mapping is not needed
3. Convert the input to Row first (e.g. apply Convert.toRow()) before the DoFn

Example fix

// before
PCollection<KV<String, V>> kv = ...;
kv.apply(ParDo.of(doFnWithSchemaElementParams));
// after
PCollection<Row> rows = kv.apply(Convert.toRow());
rows.apply(ParDo.of(doFnWithSchemaElementParams));

if (hasSchemaElementParameters(fn) && !input.hasSchema()) {
  throw new IllegalArgumentException("Input PCollection must have a schema for @SchemaElementParameter");
}

boolean schemaCompatible(DoFn<?, ?> fn, PCollection<?> input) {
  DoFnSignature sig = DoFnSignatures.getSignature(fn.getClass());
  return sig.processElement().getSchemaElementParameters().isEmpty() || input.hasSchema();
}

try {
  return ParDo.getDoFnSchemaInformation(fn, input);
} catch (IllegalArgumentException e) {
  if (e.getMessage().startsWith("Type of @Element must match the DoFn type")) {
    throw new IllegalStateException("Convert input to a schema-bearing PCollection (Row/POJO) first", e);
  }
  throw e;
}

When it happens

Trigger: Calling ParDo.getDoFnSchemaInformation(fn, input) (during pipeline construction/validation) where fn's @Element parameter carries schema element parameters but input.hasSchema() is false — e.g. input is a KV, primitive, or unregistered type rather than a schema-bearing Row/POJO.

Common situations: Using @SchemaElementParameter (e.g. field access ordering hints) on a DoFn fed by a non-schema PCollection; feeding an Avro/Row-typed pipeline stage into a DoFn expecting schema conversion of a non-schema input.

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 call getFromRowFunction when there is no schema
• Cannot call getSchema when there is no schema
• Cannot call getToRowFunction when there is no schema
• Cannot provide a coder for a Beam Row. Please provide a…
• Collection element type cannot be null for type: " +…

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

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java:654

      throw new UnsupportedOperationException(
          String.format(
              "%s is splittable and uses timer family, but these are not compatible",
              fn.getClass().getName()));
    }
  }

  /**
   * Extract information on how the DoFn uses schemas. In particular, if the schema of an element
   * parameter does not match the input PCollection's schema, convert.
   */
  @Internal
  public static DoFnSchemaInformation getDoFnSchemaInformation(
      DoFn<?, ?> fn, PCollection<?> input) {
    DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());
    DoFnSignature.ProcessElementMethod processElementMethod = signature.processElement();
    if (!processElementMethod.getSchemaElementParameters().isEmpty()) {
      if (!input.hasSchema()) {
        throw new IllegalArgumentException("Type of @Element must match the DoFn type" + input);
      }
    }

    SchemaRegistry schemaRegistry = input.getPipeline().getSchemaRegistry();
    DoFnSchemaInformation doFnSchemaInformation = DoFnSchemaInformation.create();
    for (SchemaElementParameter parameter : processElementMethod.getSchemaElementParameters()) {
      TypeDescriptor<?> elementT = parameter.elementT();
      FieldAccessDescriptor accessDescriptor =
          getFieldAccessDescriptorFromParameter(
              parameter.fieldAccessString(),
              input.getSchema(),
              signature.fieldAccessDeclarations(),
              fn);
      doFnSchemaInformation = doFnSchemaInformation.withFieldAccessDescriptor(accessDescriptor);
      Schema selectedSchema = SelectHelpers.getOutputSchema(input.getSchema(), accessDescriptor);
      ConvertHelpers.ConvertedSchemaInformation converted =
          ConvertHelpers.getConvertedSchemaInformation(selectedSchema, elementT, schemaRegistry);
      if (converted.outputSchemaCoder != null) {

View on GitHub (pinned to 12126d8942)