apache/flink · error · IOException

Corrupt data: Unexpected magic number.

Error message

Corrupt data: Unexpected magic number.

What it means

Thrown by S3RecoverableSerializer.deserializeV1 when deserializing the persisted state of an in-progress S3 recoverable write (a CommitRecoverable snapshot). The serialized payload starts with a fixed MAGIC_NUMBER int (little-endian); if the first 4 bytes do not match, the bytes are not a valid S3Recoverable payload. This almost always means the state/savepoint data is truncated, corrupted, or was written by a different serializer.

Source

Thrown at flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java:130

        return targetBytes;
    }

    @Override
    public S3Recoverable deserialize(int version, byte[] serialized) throws IOException {
        switch (version) {
            case 1:
                return deserializeV1(serialized);
            default:
                throw new IOException("Unrecognized version or corrupt state: " + version);
        }
    }

    private static S3Recoverable deserializeV1(byte[] serialized) throws IOException {
        final ByteBuffer bb = ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN);

        if (bb.getInt() != MAGIC_NUMBER) {
            throw new IOException("Corrupt data: Unexpected magic number.");
        }

        final byte[] keyBytes = new byte[bb.getInt()];
        bb.get(keyBytes);

        final byte[] uploadIdBytes = new byte[bb.getInt()];
        bb.get(uploadIdBytes);

        final int numParts = bb.getInt();
        final ArrayList<PartETag> parts = new ArrayList<>(numParts);
        for (int i = 0; i < numParts; i++) {
            final int partNum = bb.getInt();
            final byte[] buffer = new byte[bb.getInt()];
            bb.get(buffer);
            parts.add(new PartETag(partNum, new String(buffer, CHARSET)));
        }

        final long numBytes = bb.getLong();

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Verify the byte[] passed to deserialize came from S3RecoverableSerializer.serialize() of the same Flink version; do not mix recoverables across filesystem implementations.
2. Check the preceding version byte: the switch only accepts version 1, so payloads serialized by any other version fail earlier with 'Unrecognized version or corrupt state' — confirm you are on a matching Flink version on both write and restore.
3. If state corruption is suspected, verify checkpoint/savepoint integrity (sizes, checksums, storage durability) and restore from the last known-good checkpoint instead.
4. As a last resort, discard the in-progress recoverable data and rewrite the output (losing exactly-once append for that file), then re-run.

Example fix

// before
S3Recoverable rec = (S3Recoverable) S3RecoverableSerializer.INSTANCE.deserialize(corruptBytes);

// after
CommitRecoverable rec = S3RecoverableSerializer.INSTANCE.deserialize(bytes);
// bytes must originate from S3RecoverableSerializer.serialize() of the same
// Flink version and the same S3 RecoverableWriter; validate length first:
if (bytes == null || bytes.length < 5) {
    throw new IOException("State too short to be an S3Recoverable payload");
}

// before deserializing recoverable state
private static final int MAGIC = S3RecoverableSerializer.MAGIC_NUMBER; // or inline constant
static boolean looksLikeS3Recoverable(byte[] bytes) {
    if (bytes == null || bytes.length < 5) return false; // version byte + 4 magic bytes
    ByteBuffer bb = ByteBuffer.wrap(bytes, 1, 4).order(ByteOrder.LITTLE_ENDIAN);
    return bb.getInt() == MAGIC;
}

try {
    S3Recoverable rec = (S3Recoverable) S3RecoverableSerializer.INSTANCE.deserialize(bytes);
} catch (IOException e) {
    // treat as unrecoverable state: log checkpoint/savepoint id, do not resume,
    // fall back to full rewrite of the output file
}

When it happens

Trigger: Calling S3RecoverableSerializer.deserialize() (directly or via RecoverableWriter.recover/resume/commit after checkpoint restore) with a byte[] that is not a version-1 S3Recoverable: truncated state, garbage bytes, little-endian/big-endian mismatch, or a payload produced by a different RecoverableSerializer (e.g. a different filesystem's writer).

Common situations: Restoring a job from a savepoint or checkpoint where the persisted recoverable-serializer state was corrupted (partial write to state backend), upgrading Flink across versions that changed the serializer format, or handing a recoverable from one filesystem implementation (e.g. HadoopS3) to the wrong writer. Also happens when operator state was manually edited or the state backend lost data.

Related errors

• Unrecognized version or corrupt state: {version}
• Corrupt data: Unexpected magic number %08X
• Corrupt data: Unexpected magic number %08X
• Could not create a restore serializer for enum {}. Probably
• Cannot instantiate class.

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