From 7aef3a69c8c97688a7e65fe2c57a8a30564694cc Mon Sep 17 00:00:00 2001
From: joyCurry30 <curry30play@gmail.com>
Date: Fri, 29 Dec 2023 15:45:55 +0800
Subject: [PATCH] [cdc-common] Address comment.

---
 .../schema/ColumnWithPositionSerializer.java   | 18 +++++-------------
 1 file changed, 5 insertions(+), 13 deletions(-)

diff --git a/flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/serializer/schema/ColumnWithPositionSerializer.java b/flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/serializer/schema/ColumnWithPositionSerializer.java
index 60e5d0cc7eb..87776b85aca 100644
--- a/flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/serializer/schema/ColumnWithPositionSerializer.java
+++ b/flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/serializer/schema/ColumnWithPositionSerializer.java
@@ -44,7 +44,7 @@ public class ColumnWithPositionSerializer
     private final TypeSerializer<Column> addColumnSerializer =
             new NullableSerializerWrapper<>(ColumnSerializer.INSTANCE);
 
-    private final TypeSerializer<String> existedColumnSerializer = new StringSerializer();
+    private final TypeSerializer<String> existedColumnNameSerializer = StringSerializer.INSTANCE;
 
     private final EnumSerializer<AddColumnEvent.ColumnPosition> positionEnumSerializer =
             new EnumSerializer<>(AddColumnEvent.ColumnPosition.class);
@@ -65,7 +65,7 @@ public AddColumnEvent.ColumnWithPosition copy(AddColumnEvent.ColumnWithPosition
         return new AddColumnEvent.ColumnWithPosition(
                 addColumnSerializer.copy(from.getAddColumn()),
                 from.getPosition(),
-                existedColumnSerializer.copy(from.getExistedColumnName()));
+                existedColumnNameSerializer.copy(from.getExistedColumnName()));
     }
 
     @Override
@@ -84,23 +84,15 @@ public void serialize(AddColumnEvent.ColumnWithPosition record, DataOutputView t
             throws IOException {
         addColumnSerializer.serialize(record.getAddColumn(), target);
         positionEnumSerializer.serialize(record.getPosition(), target);
-        if (record.getExistedColumnName() == null) {
-            target.writeInt(0);
-        } else {
-            target.writeInt(1);
-            existedColumnSerializer.serialize(record.getExistedColumnName(), target);
-        }
+        existedColumnNameSerializer.serialize(record.getExistedColumnName(), target);
     }
 
     @Override
     public AddColumnEvent.ColumnWithPosition deserialize(DataInputView source) throws IOException {
         Column addColumn = addColumnSerializer.deserialize(source);
         AddColumnEvent.ColumnPosition position = positionEnumSerializer.deserialize(source);
-        if (source.readInt() == 1) {
-            return new AddColumnEvent.ColumnWithPosition(
-                    addColumn, position, existedColumnSerializer.deserialize(source));
-        }
-        return new AddColumnEvent.ColumnWithPosition(addColumn, position, null);
+        return new AddColumnEvent.ColumnWithPosition(
+                addColumn, position, existedColumnNameSerializer.deserialize(source));
     }
 
     @Override