apache/flink · error · IOException

Invalid version %d

Error message

Invalid version %d

What it means

HybridSourceSplitSerializer#deserialize only knows how to read version 0 of the serialized split format. Any other version number (passed by the framework from SimpleVersionedSerializer) is rejected with an IOException. This guards against reading a split that was serialized by a newer or older incompatible serializer revision.

Source

Thrown at flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitSerializer.java:58

    public byte[] serialize(HybridSourceSplit split) throws IOException {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
                DataOutputStream out = new DataOutputStream(baos)) {
            out.writeInt(split.sourceIndex());
            out.writeUTF(split.splitId());
            out.writeInt(split.wrappedSplitSerializerVersion());
            out.writeInt(split.wrappedSplitBytes().length);
            out.write(split.wrappedSplitBytes());
            out.flush();
            return baos.toByteArray();
        }
    }

    @Override
    public HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {
        if (version == 0) {
            return deserializeV0(serialized);
        }
        throw new IOException(String.format("Invalid version %d", version));
    }

    private HybridSourceSplit deserializeV0(byte[] serialized) throws IOException {
        try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
                DataInputStream in = new DataInputStream(bais)) {
            int sourceIndex = in.readInt();
            String splitId = in.readUTF();
            int nestedVersion = in.readInt();
            int length = in.readInt();
            byte[] splitBytes = new byte[length];
            in.readFully(splitBytes);
            return new HybridSourceSplit(sourceIndex, splitBytes, nestedVersion, splitId);
        }
    }
}

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Check the JobManager logs for the exact version number reported and compare against the current HybridSourceSplitSerializer constant (0).
2. If restoring across an incompatible Flink version, start a fresh job without the savepoint because the split bytes cannot be decoded.
3. If you maintain a fork that changed the version, add a deserializeV<n> branch mirroring the existing deserializeV0 pattern.
4. Ensure no custom code is calling serialize with a version other than the one returned by getVersion().

Example fix

// before
@Override
public HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {
    if (version == 0) return deserializeV0(serialized);
    throw new IOException(String.format("Invalid version %d", version));
}
// after: handle a new version explicitly
@Override
public HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {
    switch (version) {
        case 0: return deserializeV0(serialized);
        case 1: return deserializeV1(serialized);
        default: throw new IOException(String.format("Invalid version %d", version));
    }
}

int expected = new HybridSourceSplitSerializer().getVersion(); // 0
if (version != expected) {
    throw new IOException("Cannot restore HybridSourceSplit: version " + version + " != " + expected);
}

try {
    return serializer.deserialize(version, bytes);
} catch (IOException e) {
    if (e.getMessage().startsWith("Invalid version")) {
        // incompatible checkpoint; cannot recover this split
        throw new FlinkRuntimeException("Incompatible HybridSourceSplit version: " + version, e);
    }
    throw e;
}

When it happens

Trigger: Restoring a checkpoint/savepoint whose HybridSourceSplit bytes were written with a serializer version other than 0; a custom serializer wrapper that bumps the outer version without handling it; downgrading Flink after splits were persisted under a newer format.

Common situations: Downgrading Flink versions where the split format version changed; mixing Flink versions in a recovery path; manual corruption of checkpoint metadata.

Related errors

• The bytes are serialized with version %d, while this deseria
• Failed to create enumerator for sourceIndex={currentSourceIn
• Unrecognized version or corrupt state: {}
• Incompatible version: found {}, compatible versions are {}
• Serialized data with version %d cannot be read by serializer

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