apache/flink · error · IllegalArgumentException

Please use AvroParquetReaders.forGenericRecord(Class<T>) for

Error message

Please use AvroParquetReaders.forGenericRecord(Class<T>) for GenericRecord.Cannot read and create Avro GenericRecord without specifying the Avro Schema. This is because Flink needs to be able serialize the results in its data flow, which isvery inefficient without the schema. And while the Schema is stored in the Avro file header,Flink needs this schema during 'pre-flight' time when the data flow is set up and wired,which is before there is access to the files

What it means

AvroParquetReaders.forReflectRecord(Class) refuses classes implementing GenericRecord. Reading GenericRecords requires the Avro Schema up-front (at pipeline construction / 'pre-flight' time) because Flink must serialize the produced type in the data flow; the schema embedded in the parquet file header is not available that early.

Source

Thrown at flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/avro/AvroParquetReaders.java:75

    /**
     * Creates a new {@link AvroParquetRecordFormat} that reads the parquet file into Avro records
     * via reflection.
     *
     * <p>To read into Avro {@link GenericRecord GenericRecords}, use the {@link
     * #forGenericRecord(Schema)} method.
     *
     * <p>To read into Avro {@link org.apache.avro.specific.SpecificRecord SpecificRecords}, use the
     * {@link #forSpecificRecord(Class)} method.
     *
     * @see #forGenericRecord(Schema)
     * @see #forSpecificRecord(Class)
     */
    public static <T> StreamFormat<T> forReflectRecord(final Class<T> typeClass) {
        if (SpecificRecordBase.class.isAssignableFrom(typeClass)) {
            throw new IllegalArgumentException(
                    "Please use AvroParquetReaders.forSpecificRecord(Class<T>) for SpecificRecord.");
        } else if (GenericRecord.class.isAssignableFrom(typeClass)) {
            throw new IllegalArgumentException(
                    "Please use AvroParquetReaders.forGenericRecord(Class<T>) for GenericRecord."
                            + "Cannot read and create Avro GenericRecord without specifying the Avro Schema. "
                            + "This is because Flink needs to be able serialize the results in its data flow, which is"
                            + "very inefficient without the schema. And while the Schema is stored in the Avro file header,"
                            + "Flink needs this schema during 'pre-flight' time when the data flow is set up and wired,"
                            + "which is before there is access to the files");
        }

        // this is a PoJo that Avo will reader via reflect de-serialization
        // for Flink, this is just a plain PoJo type
        return new AvroParquetRecordFormat<>(
                TypeExtractor.createTypeInfo(typeClass), () -> ReflectData.get());
    }

    /**
     * Creates a new {@link AvroParquetRecordFormat} that reads the parquet file into Avro {@link
     * GenericRecord GenericRecords}.
     *

View on GitHub (pinned to 2f3c205e92)

Solutions

Use AvroParquetReaders.forGenericRecord(avroSchema) with an explicitly provided org.apache.avro.Schema
Load the schema from the .avsc resource at job construction time and pass it in
For specific generated classes, use forSpecificRecord(Class)

Example fix

// before
StreamFormat<GenericRecord> f = AvroParquetReaders.forReflectRecord(GenericData.Record.class);

// after
Schema schema = new Schema.Parser().parse(getResources().open("user.avsc"));
StreamFormat<GenericRecord> f = AvroParquetReaders.forGenericRecord(schema);

Defensive patterns

Strategy: validation

Validate before calling

Schema schema = new Schema.Parser().parse(new FileInputStream("/path/user.avsc")); // resolve schema at job construction, then: AvroParquetReaders.forGenericRecord(schema)

Prevention

Always load the .avsc into the job and call forGenericRecord(schema) for generic records
Never assume parquet file headers can substitute for an up-front schema in Flink
Beware: the thrown message concatenates without spaces - the garbled text is still this single error

When it happens

Trigger: Calling AvroParquetReaders.forReflectRecord(GenericData.Record.class) or any class implementing org.apache.avro.generic.GenericRecord.

Common situations: Developers migrating from parquet-avro's ParquetReader (which can lazily read the schema from the file) to Flink's StreamFormat and passing GenericData.Record.class.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/fa3a30fc597e39ad. Report an issue: GitHub.