[FLINK-36891[[source-connector][mysql] Fix corrupted state in case of serialization failure in MySQL CDC Source

This closes #3794.

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializer.java

@@ -80,6 +80,7 @@ public class PendingSplitsStateSerializer implements SimpleVersionedSerializer<P
         }
         final DataOutputSerializer out = SERIALIZER_CACHE.get();
 
+        try {
             out.writeInt(splitSerializer.getVersion());
             if (state instanceof SnapshotPendingSplitsState) {
                 out.writeInt(SNAPSHOT_PENDING_SPLITS_STATE_FLAG);
@@ -100,8 +101,11 @@ public class PendingSplitsStateSerializer implements SimpleVersionedSerializer<P
             // optimization: cache the serialized from, so we avoid the byte work during repeated
             // serialization
             state.serializedFormCache = result;
-        out.clear();
+
             return result;
+        } finally {
+            out.clear();
+        }
     }
 
     @Override

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializerTest.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.base.source.assigner.state;
+
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitSerializer;
+
+import io.debezium.relational.TableId;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertThrows;
+
+/** Tests for {@link PendingSplitsStateSerializer}. */
+public class PendingSplitsStateSerializerTest {
+
+    private final TableId tableId = TableId.parse("catalog.schema.table1");
+
+    @Test
+    public void testOutputIsFinallyCleared() throws Exception {
+        PendingSplitsStateSerializer serializer =
+                new PendingSplitsStateSerializer(constructSourceSplitSerializer());
+        StreamPendingSplitsState state = new StreamPendingSplitsState(true);
+
+        final byte[] ser1 = serializer.serialize(state);
+        state.serializedFormCache = null;
+
+        PendingSplitsState unsupportedState = new UnsupportedPendingSplitsState();
+
+        assertThrows(IOException.class, () -> serializer.serialize(unsupportedState));
+
+        final byte[] ser2 = serializer.serialize(state);
+        assertArrayEquals(ser1, ser2);
+    }
+
+    private SourceSplitSerializer constructSourceSplitSerializer() {
+        return new SourceSplitSerializer() {
+            @Override
+            public OffsetFactory getOffsetFactory() {
+                return new OffsetFactory() {
+                    @Override
+                    public Offset newOffset(Map<String, String> offset) {
+                        return null;
+                    }
+
+                    @Override
+                    public Offset newOffset(String filename, Long position) {
+                        return null;
+                    }
+
+                    @Override
+                    public Offset newOffset(Long position) {
+                        return null;
+                    }
+
+                    @Override
+                    public Offset createTimestampOffset(long timestampMillis) {
+                        return null;
+                    }
+
+                    @Override
+                    public Offset createInitialOffset() {
+                        return null;
+                    }
+
+                    @Override
+                    public Offset createNoStoppingOffset() {
+                        return null;
+                    }
+                };
+            }
+        };
+    }
+
+    /** An implementation for {@link PendingSplitsState} which will cause a serialization error. */
+    static class UnsupportedPendingSplitsState extends PendingSplitsState {}
+}

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializer.java

@@ -72,6 +72,7 @@ public class PendingSplitsStateSerializer implements SimpleVersionedSerializer<P
         }
         final DataOutputSerializer out = SERIALIZER_CACHE.get();
 
+        try {
             out.writeInt(splitSerializer.getVersion());
             if (state instanceof SnapshotPendingSplitsState) {
                 out.writeInt(SNAPSHOT_PENDING_SPLITS_STATE_FLAG);
@@ -92,8 +93,11 @@ public class PendingSplitsStateSerializer implements SimpleVersionedSerializer<P
             // optimization: cache the serialized from, so we avoid the byte work during repeated
             // serialization
             state.serializedFormCache = result;
-        out.clear();
+
             return result;
+        } finally {
+            out.clear();
+        }
     }
 
     @Override

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/state/PendingSplitsStateSerializerTest.java

@@ -33,6 +33,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -42,8 +43,10 @@ import java.util.List;
 import java.util.Map;
 
 import static org.apache.flink.cdc.connectors.mysql.source.split.MySqlSnapshotSplit.generateSplitId;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
 
 /**
  * Tests for {@link
@@ -102,6 +105,22 @@ public class PendingSplitsStateSerializerTest {
         assertSame(ser1, ser3);
     }
 
+    @Test
+    public void testOutputIsFinallyCleared() throws Exception {
+        final PendingSplitsStateSerializer serializer =
+                new PendingSplitsStateSerializer(MySqlSplitSerializer.INSTANCE);
+
+        final byte[] ser1 = serializer.serialize(state);
+        state.serializedFormCache = null;
+
+        PendingSplitsState unsupportedState = new UnsupportedPendingSplitsState();
+
+        assertThrows(IOException.class, () -> serializer.serialize(unsupportedState));
+
+        final byte[] ser2 = serializer.serialize(state);
+        assertArrayEquals(ser1, ser2);
+    }
+
     static PendingSplitsState serializeAndDeserializeSourceEnumState(PendingSplitsState state)
             throws Exception {
         final PendingSplitsStateSerializer serializer =
@@ -270,4 +289,7 @@ public class PendingSplitsStateSerializerTest {
             throw new UnsupportedOperationException("Not implemented.");
         }
     }
+
+    /** An implementation for {@link PendingSplitsState} which will cause a serialization error. */
+    static class UnsupportedPendingSplitsState extends PendingSplitsState {}
 }