[lake] LakeSourceSplit should be backcompatible (#1703)

Author: caozhen1937

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitSerializer.java

@@ -47,6 +47,7 @@ public LakeSplitSerializer(SimpleVersionedSerializer<LakeSplit> sourceSplitSeria
     }
 
     public void serialize(DataOutputSerializer out, SourceSplitBase split) throws IOException {
+        out.writeInt(sourceSplitSerializer.getVersion());
         if (split instanceof LakeSnapshotSplit) {
             LakeSnapshotSplit lakeSplit = (LakeSnapshotSplit) split;
             out.writeInt(lakeSplit.getSplitIndex());
@@ -91,13 +92,12 @@ public SourceSplitBase deserialize(
             @Nullable String partition,
             DataInputDeserializer input)
             throws IOException {
+        int version = input.readInt();
         if (splitKind == LAKE_SNAPSHOT_SPLIT_KIND) {
             int splitIndex = input.readInt();
             byte[] serializeBytes = new byte[input.readInt()];
             input.read(serializeBytes);
-            LakeSplit lakeSplit =
-                    sourceSplitSerializer.deserialize(
-                            sourceSplitSerializer.getVersion(), serializeBytes);
+            LakeSplit lakeSplit = sourceSplitSerializer.deserialize(version, serializeBytes);
             return new LakeSnapshotSplit(tableBucket, partition, lakeSplit, splitIndex);
         } else if (splitKind == LAKE_SNAPSHOT_FLUSS_LOG_SPLIT_KIND) {
             List<LakeSplit> lakeSplits = null;
@@ -107,9 +107,7 @@ public SourceSplitBase deserialize(
                 for (int i = 0; i < lakeSplitSize; i++) {
                     byte[] serializeBytes = new byte[input.readInt()];
                     input.read(serializeBytes);
-                    lakeSplits.add(
-                            sourceSplitSerializer.deserialize(
-                                    sourceSplitSerializer.getVersion(), serializeBytes));
+                    lakeSplits.add(sourceSplitSerializer.deserialize(version, serializeBytes));
                 }
             }
             long startingOffset = input.readLong();

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/state/FlussSourceEnumeratorStateSerializer.java

@@ -142,12 +142,12 @@ private void serializeRemainingHybridLakeFlussSplits(
             out.writeBoolean(true);
             out.writeInt(remainingHybridLakeFlussSplits.size());
             SourceSplitSerializer sourceSplitSerializer = new SourceSplitSerializer(lakeSource);
+            out.writeInt(sourceSplitSerializer.getVersion());
             for (SourceSplitBase split : remainingHybridLakeFlussSplits) {
                 byte[] serializeBytes = sourceSplitSerializer.serialize(split);
                 out.writeInt(serializeBytes.length);
                 out.write(serializeBytes);
             }
-
         } else {
             // write that hybrid lake fluss splits is null
             out.writeBoolean(false);
@@ -161,13 +161,12 @@ private List<SourceSplitBase> deserializeRemainingHybridLakeFlussSplits(
             int numSplits = in.readInt();
             List<SourceSplitBase> splits = new ArrayList<>(numSplits);
             SourceSplitSerializer sourceSplitSerializer = new SourceSplitSerializer(lakeSource);
+            int version = in.readInt();
             for (int i = 0; i < numSplits; i++) {
                 int splitSizeInBytes = in.readInt();
                 byte[] splitBytes = new byte[splitSizeInBytes];
                 in.readFully(splitBytes);
-                splits.add(
-                        sourceSplitSerializer.deserialize(
-                                sourceSplitSerializer.getVersion(), splitBytes));
+                splits.add(sourceSplitSerializer.deserialize(version, splitBytes));
             }
             return splits;
         } else {

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/lake/LakeSplitSerializerTest.java

@@ -42,8 +42,6 @@
 class LakeSplitSerializerTest {
     private static final byte LAKE_SNAPSHOT_SPLIT_KIND = -1;
 
-    private static final int SERIALIZER_VERSION = 3;
-
     private static final byte[] TEST_DATA = "test-lake-split".getBytes();
 
     private static final int STOPPING_OFFSET = 1024;
@@ -61,8 +59,9 @@ class LakeSplitSerializerTest {
     @Test
     void testSerializeAndDeserializeLakeSnapshotSplit() throws IOException {
         // Prepare test data
+        int splitIndex = 1;
         LakeSnapshotSplit originalSplit =
-                new LakeSnapshotSplit(tableBucket, "2025-08-18", LAKE_SPLIT, 1);
+                new LakeSnapshotSplit(tableBucket, "2025-08-18", LAKE_SPLIT, splitIndex);
 
         DataOutputSerializer output = new DataOutputSerializer(STOPPING_OFFSET);
         serializer.serialize(output, originalSplit);
@@ -80,6 +79,40 @@ void testSerializeAndDeserializeLakeSnapshotSplit() throws IOException {
         assertThat(tableBucket).isEqualTo(result.getTableBucket());
         assertThat("2025-08-18").isEqualTo(result.getPartitionName());
         assertThat(LAKE_SPLIT).isEqualTo(result.getLakeSplit());
+        assertThat(splitIndex).isEqualTo(result.getSplitIndex());
+    }
+
+    @Test
+    void testSerializeAndDeserializeLakeSnapshotSplitBackwardCompatibility() throws IOException {
+        SimpleVersionedSerializer<LakeSplit> sourceSplitSerializerV1 =
+                new TestSimpleVersionedSerializer();
+        SimpleVersionedSerializer<LakeSplit> sourceSplitSerializerV2 =
+                new TestSimpleVersionedSerializerV2();
+        LakeSplitSerializer serializerV1 = new LakeSplitSerializer(sourceSplitSerializerV1);
+        LakeSplitSerializer serializerV2 = new LakeSplitSerializer(sourceSplitSerializerV2);
+
+        // Prepare test data
+        int splitIndex = 1;
+        LakeSnapshotSplit originalSplit =
+                new LakeSnapshotSplit(tableBucket, "2025-08-18", LAKE_SPLIT, splitIndex);
+
+        DataOutputSerializer output = new DataOutputSerializer(STOPPING_OFFSET);
+        serializerV1.serialize(output, originalSplit);
+
+        SourceSplitBase deserializedSplit =
+                serializerV2.deserialize(
+                        LAKE_SNAPSHOT_SPLIT_KIND,
+                        tableBucket,
+                        "2025-08-18",
+                        new DataInputDeserializer(output.getCopyOfBuffer()));
+
+        assertThat(deserializedSplit instanceof LakeSnapshotSplit).isTrue();
+        LakeSnapshotSplit result = (LakeSnapshotSplit) deserializedSplit;
+
+        assertThat(tableBucket).isEqualTo(result.getTableBucket());
+        assertThat("2025-08-18").isEqualTo(result.getPartitionName());
+        assertThat(LAKE_SPLIT).isEqualTo(result.getLakeSplit());
+        assertThat(splitIndex).isEqualTo(result.getSplitIndex());
     }
 
     @Test
@@ -137,6 +170,8 @@ void testDeserializeWithWrongSplitKind() throws IOException {
     private static class TestSimpleVersionedSerializer
             implements SimpleVersionedSerializer<LakeSplit> {
 
+        private static final int V1 = 1;
+
         @Override
         public byte[] serialize(LakeSplit split) throws IOException {
             return TEST_DATA;
@@ -149,7 +184,31 @@ public LakeSplit deserialize(int version, byte[] serialized) throws IOException
 
         @Override
         public int getVersion() {
-            return SERIALIZER_VERSION;
+            return V1;
+        }
+    }
+
+    private static class TestSimpleVersionedSerializerV2
+            implements SimpleVersionedSerializer<LakeSplit> {
+
+        private static final int V2 = 2;
+
+        @Override
+        public byte[] serialize(LakeSplit split) throws IOException {
+            return TEST_DATA;
+        }
+
+        @Override
+        public LakeSplit deserialize(int version, byte[] serialized) throws IOException {
+            if (version < V2) {
+                return LAKE_SPLIT;
+            }
+            return new TestLakeSplit(0, Collections.singletonList("2025-08-19"));
+        }
+
+        @Override
+        public int getVersion() {
+            return V2;
         }
     }