[hotfix] Fix union read can't restore issue

[luoyuxia]

Purpose

Linked issue: close #xxx

Brief change log

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This PR fixes an issue with union read restoration by introducing a new serialization version (VERSION_1) for SourceEnumeratorState. The key change is that VERSION_1 always serializes and deserializes remainingHybridLakeFlussSplits regardless of the lakeSource flag, fixing a bug where VERSION_0 would only serialize/deserialize these splits when lakeSource was non-null.

Key Changes:

• Introduced VERSION_1 serialization format that always handles remainingHybridLakeFlussSplits
• Refactored serialization logic to extract common bucket/partition serialization into a helper method
• Updated deserialization to handle both VERSION_0 (backward compatible) and VERSION_1 formats
• Added comprehensive tests for V0 compatibility and cross-version scenarios

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.

File	Description
FlussSourceEnumeratorStateSerializer.java	Introduces VERSION_1 serialization format, refactors serialization logic into helper methods, and updates deserialization to handle version-specific behavior
SourceEnumeratorStateSerializerTest.java	Adds tests for V0 backward compatibility and cross-version deserialization scenarios

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[loserwang1024]

I have left some message.

[loserwang1024]

Would like to add what's the different with different version? such as:

[loserwang1024]

version != VERSION_0 && version != CURRENT_VERSION

When we support version 3 later, state with version 3 cannot downgrade. We have met this problem in many connectors. @leonardBang , CC, should we stop higher version's state?

And I also found a problem : we can never downgrade later to version 1. Because in later code will version 1 state: version != VERSION_0 is always true. Maybe we should add this to upgrade document later.

[luoyuxia]

1: so, what's the best pratice in here? Always make the version check pass and make sure version3 compitable with v2? From kafka connector code, seems the kafka connector will still have the problem?

2: what do you mean by saying we can never downgrade later to version 1? Seems it still the same problem to 1?

[loserwang1024]

I understand it. But in your follower comment "for VERSION_1 and later" maybe inconsistency.

Maybe you can use switch.. case as kafka does. It's clearer

[luoyuxia]

Agree, that's clearer

[luoyuxia]

@loserwang1024 Comments is addressed. PTAL

[loserwang1024]

LGTM