Checkpoint 50 - Remove code duplication via overloading

Author: voonhous

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/event/HudiRecordEventSerializer.java

@@ -134,49 +134,7 @@ public HoodieFlinkInternalRow serialize(Event event, String fileId, String insta
      */
     @Override
     public HoodieFlinkInternalRow serialize(Event event) {
-        if (event instanceof CreateTableEvent) {
-            CreateTableEvent createTableEvent = (CreateTableEvent) event;
-            schemaMaps.put(createTableEvent.tableId(), createTableEvent.getSchema());
-            // Clear field getter cache for this table since schema changed
-            fieldGetterCache.remove(createTableEvent.tableId());
-            // Schema events don't produce records
-            return null;
-
-        } else if (event instanceof SchemaChangeEvent) {
-            SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
-            Schema existingSchema = schemaMaps.get(schemaChangeEvent.tableId());
-            if (existingSchema != null
-                    && !SchemaUtils.isSchemaChangeEventRedundant(
-                            existingSchema, schemaChangeEvent)) {
-                Schema newSchema =
-                        SchemaUtils.applySchemaChangeEvent(existingSchema, schemaChangeEvent);
-                schemaMaps.put(schemaChangeEvent.tableId(), newSchema);
-                // Clear field getter cache for this table since schema changed
-                fieldGetterCache.remove(schemaChangeEvent.tableId());
-            }
-            // Schema events don't produce records
-            return null;
-
-        } else if (event instanceof DataChangeEvent) {
-            DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
-            Schema schema = schemaMaps.get(dataChangeEvent.tableId());
-
-            if (schema == null) {
-                throw new IllegalStateException(
-                        "No schema available for table "
-                                + dataChangeEvent.tableId()
-                                + ". CreateTableEvent should arrive before DataChangeEvent.");
-            }
-
-            // Convert DataChangeEvent to HoodieFlinkInternalRow using utility function
-            // Use temporary values that will be overridden later
-            return RowDataUtils.convertDataChangeEventToHoodieFlinkInternalRow(
-                    dataChangeEvent, schema, zoneId, "temp", "temp");
-        } else {
-            throw new IllegalArgumentException(
-                    "Unsupported event type for Hudi serialization: "
-                            + event.getClass().getSimpleName());
-        }
+        return serialize(event, "temp", "temp");
     }
 
     /**
@@ -199,22 +157,6 @@ public boolean hasSchema(TableId tableId) {
         return schemaMaps.containsKey(tableId);
     }
 
-    /**
-     * Get cached field getters for a table, creating them if needed.
-     *
-     * @param tableId The table identifier
-     * @return List of field getters or null if schema not available
-     */
-    public List<RecordData.FieldGetter> getFieldGetters(TableId tableId) {
-        Schema schema = schemaMaps.get(tableId);
-        if (schema == null) {
-            return null;
-        }
-
-        return fieldGetterCache.computeIfAbsent(
-                tableId, k -> RowDataUtils.createFieldGetters(schema, zoneId));
-    }
-
     /**
      * Set schema for a table. Used to initialize table-specific serializers with schema.
      *

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/function/MultiTableEventStreamWriteFunction.java

@@ -436,8 +436,8 @@ private EventBucketStreamWriteFunction createTableFunction(TableId tableId) thro
         tableFunction.setCorrespondent(tableCorrespondent);
         tableFunction.setTableId(tableId);
 
-        // This is the key change: instead of passing the raw gateway, we pass a proxy
-        // that intercepts and enhances events with the table path.
+        // Instead of passing the raw gateway, we pass a proxy that intercepts and enhances events
+        // with the table path
         String tablePath = tableConfig.getString(FlinkOptions.PATH);
         tableFunction.setOperatorEventGateway(
                 new InterceptingGateway(this.getOperatorEventGateway(), tablePath));