{"record":{"id":"eca69a92d31da220","repo":"apache/flink","slug":"corrupted-data-to-deserialize","errorCode":null,"errorMessage":"Corrupted data to deserialize","messagePattern":"Corrupted data to deserialize","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"critical","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterStateSerializer.java","lineNumber":97,"sourceCode":"            for (int i = 0; i < size; i++) {\n                long requestSize = in.readLong();\n                RequestEntryT request = deserializeRequestFromStream(requestSize, in);\n                serializedState.add(new RequestEntryWrapper<>(request, requestSize));\n            }\n\n            return new BufferedRequestState<>(serializedState);\n        }\n    }\n\n    protected abstract void serializeRequestToStream(RequestEntryT request, DataOutputStream out)\n            throws IOException;\n\n    protected abstract RequestEntryT deserializeRequestFromStream(\n            long requestSize, DataInputStream in) throws IOException;\n\n    private void validateIdentifier(DataInputStream in) throws IOException {\n        if (in.readLong() != DATA_IDENTIFIER) {\n            throw new IllegalStateException(\"Corrupted data to deserialize\");\n        }\n    }\n}\n","sourceCodeStart":79,"sourceCodeEnd":101,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterStateSerializer.java#L79-L101","documentation":"Thrown as an IllegalStateException by AsyncSinkWriterStateSerializer.validateIdentifier when the first 8 bytes of the serialized state stream do not equal DATA_IDENTIFIER (-1L). This identifier is a sentinel written at the start of every serialized state blob to verify data integrity. A mismatch indicates the byte stream is not a valid AsyncSinkWriter state payload — it is either corrupted, truncated, or produced by an incompatible serializer.","triggerScenarios":"Deserializing BufferedRequestState from bytes that were not produced by AsyncSinkWriterStateSerializer.serialize, or from bytes that have been corrupted/truncated (e.g., partial checkpoint write, network transfer error).","commonSituations":"Restoring from a checkpoint/savepoint that was written by a different or older serializer version; checkpoint data corruption due to storage failure; manual tampering or incorrect state migration; restoring from a savepoint taken with a different sink implementation that uses the same state serializer class.","solutions":["Verify the checkpoint/savepoint was produced by a compatible version of the same AsyncSinkWriter subclass.","If the checkpoint is from an incompatible version, start fresh without restore or implement a custom state migration.","Check checkpoint storage integrity — re-copy or re-upload the checkpoint if corruption is suspected.","Ensure no other serializer is writing to the same state handle namespace."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Validate data identifier before full deserialization\ntry (DataInputStream in = new DataInputStream(new ByteArrayInputStream(serialized))) {\n    long identifier = in.readLong();\n    if (identifier != -1L) { // DATA_IDENTIFIER\n        throw new IllegalStateException(\"Invalid state data — not an AsyncSinkWriter state payload\");\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    BufferedRequestState<?> state = serializer.deserialize(version, bytes);\n} catch (IllegalStateException e) {\n    if (e.getMessage().equals(\"Corrupted data to deserialize\")) {\n        // checkpoint corruption — cannot recover, start fresh\n        log.error(\"Checkpoint state is corrupted, starting without restore\");\n    }\n}","preventionTips":["Only restore from checkpoints produced by the same serializer version.","Verify checkpoint integrity after writing (checksums, size validation).","Do not manually edit or truncate checkpoint files."],"tags":["async-sink","state-serialization","corruption","checkpoint-restore"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}