{"record":{"id":"90b75aac7c95b776","repo":"apache/flink","slug":"failed-to-serialize-schema-registry","errorCode":null,"errorMessage":"Failed to serialize schema registry.","messagePattern":"Failed to serialize schema registry\\.","errorType":"exception","errorClass":"WrappingRuntimeException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroSerializationSchema.java","lineNumber":174,"sourceCode":"            this.schema = new Parser().parse(schemaString);\n        }\n    }\n\n    @Override\n    public byte[] serialize(T object) {\n        checkAvroInitialized();\n\n        if (object == null) {\n            return null;\n        } else {\n            try {\n                datumWriter.write(object, encoder);\n                encoder.flush();\n                byte[] bytes = arrayOutputStream.toByteArray();\n                arrayOutputStream.reset();\n                return bytes;\n            } catch (IOException e) {\n                throw new WrappingRuntimeException(\"Failed to serialize schema registry.\", e);\n            }\n        }\n    }\n\n    protected void checkAvroInitialized() {\n        if (datumWriter != null) {\n            return;\n        }\n        ClassLoader cl = Thread.currentThread().getContextClassLoader();\n        if (SpecificRecord.class.isAssignableFrom(recordClazz)) {\n            Schema schema = SpecificData.get().getSchema(recordClazz);\n            this.datumWriter = new SpecificDatumWriter<>(schema);\n            this.schema = schema;\n        } else {\n            this.schema = new Schema.Parser().parse(this.schemaString);\n            GenericData genericData = new GenericData(cl);\n\n            this.datumWriter = new GenericDatumWriter<>(schema, genericData);","sourceCodeStart":156,"sourceCodeEnd":192,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroSerializationSchema.java#L156-L192","documentation":"Despite the message, this is not only about schema registries: AvroSerializationSchema.serialize throws WrappingRuntimeException('Failed to serialize schema registry.') whenever datumWriter.write() or encoder.flush() fails with an IOException while encoding the record to the in-memory stream. Registry-specific I/O is handled elsewhere; here the cause is typically a record/schema mismatch or encoder state corruption.","triggerScenarios":"AvroSerializationSchema.forSpecific/forGeneric(...).serialize(record) where the record does not conform to the writer schema (missing fields, wrong field type), or the schema changed between serialization calls without re-initialization.","commonSituations":"Passing a GenericRecord built with schema v2 while the schema object pinned at construction is v1; reusing a serialized schema object across schema upgrades; records produced by different producers.","solutions":["Check e.getCause() for Avro's encoder exception and reconcile the record's schema with the schema the AvroSerializationSchema was created with (they must match).","Recreate AvroSerializationSchema (or rebuild it via its factory) whenever the schema evolves; the datum writer is lazily cached per schema.","For SpecificRecord, ensure the class on the classpath of the task matches the class used to build records at runtime."],"exampleFix":"// before\nGenericRecord rec = new GenericData.Record(schemaV2);\n... serializer.serialize(rec); // serializer built with schemaV1\n\n// after\nGenericRecord rec = new GenericData.Record(schemaV1); // or rebuild serializer with schemaV2\n... serializer.serialize(rec);","handlingStrategy":"try-catch","validationCode":"// ensure record conforms to the writer schema before serializing\nif (record instanceof IndexedRecord\n        && !((IndexedRecord) record).getSchema().equals(expectedWriterSchema)) {\n    throw new IllegalArgumentException(\"Record schema != serializer schema\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    bytes = serializer.serialize(record);\n} catch (WrappingRuntimeException e) { // message: Failed to serialize schema registry.\n    log.error(\"Avro encode failed for schema={} cause={}\",\n        record.getSchema(), e.getCause().getMessage());\n    throw e; // data error: do not silently drop\n}","preventionTips":["Build the GenericRecord with exactly the schema the AvroSerializationSchema was constructed with.","Rebuild serializers after schema evolution instead of reusing them.","Wrap serialize() in tests with representative records."],"tags":["avro","flink","serialization","schema-mismatch"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}