From 35d3c0ff28d5aea9a59eb7eed11a8783e248c378 Mon Sep 17 00:00:00 2001
From: Hongshun Wang <125648852+loserwang1024@users.noreply.github.com>
Date: Sat, 13 Jan 2024 12:03:20 +0800
Subject: [PATCH] [cdc-base][hotfix] Fix SourceSplitSerializer to get
 totalFinishedSplitSize from recovery in 4th version

This closes #2992.
---
 .../base/source/meta/split/SourceSplitSerializer.java           | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/meta/split/SourceSplitSerializer.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/meta/split/SourceSplitSerializer.java
index 378ea21cf..12a172377 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/meta/split/SourceSplitSerializer.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/meta/split/SourceSplitSerializer.java
@@ -159,7 +159,7 @@ public abstract class SourceSplitSerializer
                     readFinishedSplitsInfo(version, in);
             Map<TableId, TableChange> tableChangeMap = readTableSchemas(version, in);
             int totalFinishedSplitSize = finishedSplitsInfo.size();
-            if (version == 3) {
+            if (version >= 3) {
                 totalFinishedSplitSize = in.readInt();
             }
             in.releaseArrays();