{"record":{"id":"22d288c039144e2d","repo":"apache/flink","slug":"failed-to-serialize-row","errorCode":null,"errorMessage":"Failed to serialize row.","messagePattern":"Failed to serialize row\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java","lineNumber":122,"sourceCode":"\n    @Override\n    public void open(InitializationContext context) throws Exception {\n        this.nestedSchema.open(context);\n        if (this.nestedSchema instanceof AvroSerializationSchema) {\n            this.schema = ((AvroSerializationSchema<GenericRecord>) this.nestedSchema).getSchema();\n        } else {\n            this.schema = AvroSchemaConverter.convertToSchema(rowType);\n        }\n    }\n\n    @Override\n    public byte[] serialize(RowData row) {\n        try {\n            // convert to record\n            final GenericRecord record = (GenericRecord) runtimeConverter.convert(schema, row);\n            return nestedSchema.serialize(record);\n        } catch (Exception e) {\n            throw new RuntimeException(\"Failed to serialize row.\", e);\n        }\n    }\n\n    @Override\n    public boolean equals(Object o) {\n        if (this == o) {\n            return true;\n        }\n        if (o == null || getClass() != o.getClass()) {\n            return false;\n        }\n        AvroRowDataSerializationSchema that = (AvroRowDataSerializationSchema) o;\n        return nestedSchema.equals(that.nestedSchema) && rowType.equals(that.rowType);\n    }\n\n    @Override\n    public int hashCode() {\n        return Objects.hash(nestedSchema, rowType);","sourceCodeStart":104,"sourceCodeEnd":140,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java#L104-L140","documentation":"AvroRowDataSerializationSchema.serialize wraps any exception from the RowData->GenericRecord conversion or the nested Avro encoder in a RuntimeException('Failed to serialize row.'). It almost always means a field value in the RowData does not fit the declared row type / Avro schema (null in a non-nullable field, wrong physical type, unsupported logical type).","triggerScenarios":"A null value in a field whose Avro schema is not a nullable union; a RowData field carrying a different physical type than the LogicalType the converters were built from; a TIMESTAMP_WITH_LOCAL_TIME_ZONE column when legacy timestamp mapping is on (propagates from RowDataToAvroConverters).","commonSituations":"Upstream operator changed nullability or types after the format was built; SQL DDL column types drifted from the actual Avro schema; mixed Flink versions where timestamp mapping defaults changed.","solutions":["Log the failing RowData (the cause carries the per-field message 'Fail to serialize at field: %s' when it comes from the row converter) and fix that field's value or type.","Make the corresponding Avro schema field nullable (union with null) if nulls are legitimate, or filter/replace nulls before serialization.","Align the RowType used to construct the schema with the actual RowData produced upstream (no silent type mismatches)."],"exampleFix":"// before\n// DDL: name STRING (non-nullable in Avro), but rows contain null\n\n// after\n// DDL: name STRING, or make the Avro field nullable:\n// {\"name\":\"name\",\"type\":[\"null\",\"string\"],\"default\":null}","handlingStrategy":"validation","validationCode":"for (int i = 0; i < rowType.getFieldCount(); i++) {\n    LogicalType ft = rowType.getTypeAt(i);\n    if (!ft.isNullable() && row.isNullAt(i)) {\n        throw new IllegalArgumentException(\"Null in non-nullable field: \" + rowType.getFieldNames().get(i));\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    out.collect(serializer.serialize(row));\n} catch (RuntimeException e) { // 'Failed to serialize row.'\n    if (e.getCause() != null && e.getCause().getMessage() != null\n            && e.getCause().getMessage().startsWith(\"Fail to serialize at field:\")) {\n        // field-level cause available; route record to DLQ\n        ctx.output(dlqTag, row);\n    } else {\n        throw e;\n    }\n}","preventionTips":["Validate nullability of RowData fields against the RowType before serialization in tests.","Keep the table DDL and the Avro sink schema in one review unit so they cannot drift.","Log failing rows with their field names (the nested cause provides them)."],"tags":["avro","flink","serialization","nullability","rowdata"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}