apache/flink · critical · IllegalStateException

Corrupted data to deserialize

Error message

Corrupted data to deserialize

What it means

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.

Source

Thrown at flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterStateSerializer.java:97

            for (int i = 0; i < size; i++) {
                long requestSize = in.readLong();
                RequestEntryT request = deserializeRequestFromStream(requestSize, in);
                serializedState.add(new RequestEntryWrapper<>(request, requestSize));
            }

            return new BufferedRequestState<>(serializedState);
        }
    }

    protected abstract void serializeRequestToStream(RequestEntryT request, DataOutputStream out)
            throws IOException;

    protected abstract RequestEntryT deserializeRequestFromStream(
            long requestSize, DataInputStream in) throws IOException;

    private void validateIdentifier(DataInputStream in) throws IOException {
        if (in.readLong() != DATA_IDENTIFIER) {
            throw new IllegalStateException("Corrupted data to deserialize");
        }
    }
}

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Verify the checkpoint/savepoint was produced by a compatible version of the same AsyncSinkWriter subclass.
2. If the checkpoint is from an incompatible version, start fresh without restore or implement a custom state migration.
3. Check checkpoint storage integrity — re-copy or re-upload the checkpoint if corruption is suspected.
4. Ensure no other serializer is writing to the same state handle namespace.

// Validate data identifier before full deserialization
try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(serialized))) {
    long identifier = in.readLong();
    if (identifier != -1L) { // DATA_IDENTIFIER
        throw new IllegalStateException("Invalid state data — not an AsyncSinkWriter state payload");
    }
}

try {
    BufferedRequestState<?> state = serializer.deserialize(version, bytes);
} catch (IllegalStateException e) {
    if (e.getMessage().equals("Corrupted data to deserialize")) {
        // checkpoint corruption — cannot recover, start fresh
        log.error("Checkpoint state is corrupted, starting without restore");
    }
}

When it happens

Trigger: 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).

Common situations: 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.

Related errors

• The bytes are serialized with version %d, while this deseria
• Unrecognized Kryo registration serializer definition type: {
• Error while opening jar file '{jarFile.getPath()}'. {ioex.ge
• The Manifest in the jar file could not be accessed '{jarFile
• Cannot create empty classifier chain.

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