[hotfix][cdc-common] Remove duplicated code to improve performance

 This closes #3840.

Co-authored-by: zhangchaoming.zcm <[email protected]>

Author: ChaomingZhangCN

Diff for: flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaUtils.java

@@ -37,6 +37,7 @@
 import org.apache.flink.cdc.common.types.TimestampType;
 import org.apache.flink.cdc.common.types.ZonedTimestampType;
 
+import javax.annotation.CheckReturnValue;
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -56,6 +57,7 @@ public class SchemaUtils {
      * create a list of {@link RecordData.FieldGetter} from given {@link Schema} to get Object from
      * RecordData.
      */
+    @CheckReturnValue
     public static List<RecordData.FieldGetter> createFieldGetters(Schema schema) {
         return createFieldGetters(schema.getColumns());
     }
@@ -64,6 +66,7 @@ public static List<RecordData.FieldGetter> createFieldGetters(Schema schema) {
      * create a list of {@link RecordData.FieldGetter} from given {@link Column} to get Object from
      * RecordData.
      */
+    @CheckReturnValue
     public static List<RecordData.FieldGetter> createFieldGetters(List<Column> columns) {
         List<RecordData.FieldGetter> fieldGetters = new ArrayList<>(columns.size());
         for (int i = 0; i < columns.size(); i++) {
@@ -73,6 +76,7 @@ public static List<RecordData.FieldGetter> createFieldGetters(List<Column> colum
     }
 
     /** Restore original data fields from RecordData structure. */
+    @CheckReturnValue
     public static List<Object> restoreOriginalData(
             @Nullable RecordData recordData, List<RecordData.FieldGetter> fieldGetters) {
         if (recordData == null) {
@@ -86,6 +90,7 @@ public static List<Object> restoreOriginalData(
     }
 
     /** apply SchemaChangeEvent to the old schema and return the schema after changing. */
+    @CheckReturnValue
     public static Schema applySchemaChangeEvent(Schema schema, SchemaChangeEvent event) {
         return SchemaChangeEventVisitor.visit(
                 event,
@@ -210,6 +215,7 @@ private static Schema applyAlterColumnTypeEvent(AlterColumnTypeEvent event, Sche
      * position indicators. This is necessary since extra calculated columns might be added, and
      * `FIRST` / `LAST` position might differ.
      */
+    @CheckReturnValue
     public static Optional<SchemaChangeEvent> transformSchemaChangeEvent(
             boolean hasAsterisk, List<String> referencedColumns, SchemaChangeEvent event) {
         Optional<SchemaChangeEvent> evolvedSchemaChangeEvent = Optional.empty();

Diff for: flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java

@@ -311,7 +311,6 @@ public Optional<DataChangeEvent> coerceDataRecord(
 
         List<RecordData.FieldGetter> upstreamSchemaReader =
                 upstreamRecordGetterCache.getUnchecked(upstreamSchema);
-        SchemaUtils.createFieldGetters(upstreamSchema);
         BinaryRecordDataGenerator evolvedSchemaWriter =
                 evolvedRecordWriterCache.getUnchecked(evolvedSchema);