Checkpoint 55 - Address comments

Author: voonhous

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/pom.xml

@@ -94,7 +94,7 @@ limitations under the License.
 
         <dependency>
             <groupId>org.apache.hudi</groupId>
-            <artifactId>hudi-flink1.20.x</artifactId>
+            <artifactId>hudi-flink${flink.major.version}.x</artifactId>
             <version>${hudi.version}</version>
         </dependency>
 

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

@@ -57,6 +57,17 @@ public HudiDataSink createDataSink(Context context) {
         FactoryHelper.DefaultContext factoryContext = (FactoryHelper.DefaultContext) context;
         Configuration config = factoryContext.getFactoryConfiguration();
 
+        // Validate that only BUCKET index type is used
+        String indexType = config.get(HudiConfig.INDEX_TYPE);
+        if (indexType != null && !indexType.equalsIgnoreCase("BUCKET")) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Unsupported index type '%s'. Currently only 'BUCKET' index type is supported. "
+                                    + "Other index types (e.g., FLINK_STATE, BLOOM, SIMPLE) are not yet implemented "
+                                    + "for multi-table CDC pipelines.",
+                            indexType));
+        }
+
         String schemaOperatorUid =
                 context.getPipelineConfiguration()
                         .get(PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID);
@@ -69,8 +80,6 @@ public Set<ConfigOption<?>> requiredOptions() {
         Set<ConfigOption<?>> options = new HashSet<>();
         options.add(HudiConfig.PATH);
         options.add(HudiConfig.RECORD_KEY_FIELD);
-        //        options.add(HudiConfig.PRECOMBINE_FIELD);
-        //        options.add(HudiConfig.BUCKET_INDEX_NUM_BUCKETS);
         return options;
     }
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.connectors.hudi.sink.bucket;
 
+import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.FlushEvent;
@@ -83,12 +84,12 @@ public class BucketAssignOperator extends AbstractStreamOperator<BucketWrapper>
     /** Cache of schemas per table for bucket calculation. */
     private final Map<TableId, Schema> schemaCache = new HashMap<>();
 
-    /** Cache of primary key fields per table. */
-    private final Map<TableId, List<String>> primaryKeyCache = new HashMap<>();
-
     /** RowDataKeyGen cache per table for key and partition extraction. */
     private final Map<TableId, RowDataKeyGen> keyGenCache = new HashMap<>();
 
+    /** Field getter cache per table - lazily created and invalidated on schema changes. */
+    private final Map<TableId, List<RecordData.FieldGetter>> fieldGetterCache = new HashMap<>();
+
     public BucketAssignOperator(Configuration conf, String schemaOperatorUid) {
         this.numBuckets = conf.getInteger(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS);
         this.schemaOperatorUid = schemaOperatorUid;
@@ -132,8 +133,8 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
             schemaCache.put(schemaEvent.tableId(), newSchema);
 
             // Clear caches when schema changes
-            primaryKeyCache.remove(schemaEvent.tableId());
             keyGenCache.remove(schemaEvent.tableId());
+            fieldGetterCache.remove(schemaEvent.tableId());
 
             // Broadcast to all tasks
             for (int i = 0; i < totalTasksNumber; i++) {
@@ -211,8 +212,7 @@ private int calculateTaskIndex(DataChangeEvent event) {
         final Schema finalSchema = schema;
 
         // Get or cache primary keys
-        List<String> primaryKeys =
-                primaryKeyCache.computeIfAbsent(tableId, k -> finalSchema.primaryKeys());
+        List<String> primaryKeys = finalSchema.primaryKeys();
 
         if (primaryKeys.isEmpty()) {
             throw new IllegalStateException(
@@ -223,11 +223,14 @@ private int calculateTaskIndex(DataChangeEvent event) {
         RowDataKeyGen keyGen =
                 keyGenCache.computeIfAbsent(tableId, k -> RowDataUtils.createKeyGen(finalSchema));
 
+        // Get or create field getters for this table, lazily cached
+        List<RecordData.FieldGetter> fieldGetters =
+                fieldGetterCache.computeIfAbsent(
+                        tableId,
+                        k -> RowDataUtils.createFieldGetters(finalSchema, ZoneId.systemDefault()));
+
         // Convert DataChangeEvent to RowData for key extraction
-        RowData rowData =
-                RowDataUtils.convertDataChangeEventToRowData(
-                        event,
-                        RowDataUtils.createFieldGetters(finalSchema, ZoneId.systemDefault()));
+        RowData rowData = RowDataUtils.convertDataChangeEventToRowData(event, fieldGetters);
 
         // Use RowDataKeyGen to extract record key and partition path
         String recordKey = keyGen.getRecordKey(rowData);
@@ -237,9 +240,9 @@ private int calculateTaskIndex(DataChangeEvent event) {
         String tableIndexKeyFields = String.join(",", primaryKeys);
         int bucketNumber = BucketIdentifier.getBucketId(recordKey, tableIndexKeyFields, numBuckets);
 
+        // partitionIndexFunc is designed for single table, events may come from different tables,
+        // prefix them with tableId e.g. tableId + "_" + partition
         // Use partition function to map bucket to task index for balanced distribution
-        int taskIndex = partitionIndexFunc.apply(numBuckets, partition, bucketNumber);
-
-        return taskIndex;
+        return partitionIndexFunc.apply(numBuckets, tableId + "_" + partition, bucketNumber);
     }
 }

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

@@ -63,7 +63,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.io.Serializable;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -489,6 +493,9 @@ private void handleCreateTableEvent(CreateTableOperatorEvent createTableOperator
                         String tablePath = tableConfig.getString(FlinkOptions.PATH);
                         pathToTableId.put(tablePath, tId);
 
+                        // Create physical directory for Hudi table before initializing
+                        createHudiTablePath(tableConfig);
+
                         StreamerUtil.initTableIfNotExists(tableConfig);
                         HoodieFlinkWriteClient<?> writeClient =
                                 FlinkWriteClients.createWriteClient(tableConfig);
@@ -920,6 +927,26 @@ public void subtaskReady(int i, SubtaskGateway subtaskGateway) {
     }
 
     // --- Helper Methods ---
+
+    /**
+     * Creates the physical directory for a Hudi table if it doesn't exist. This must be done on the
+     * coordinator side to avoid race conditions when multiple task managers try to create the same
+     * directory simultaneously.
+     *
+     * @param config The table-specific configuration containing the path
+     * @throws IOException if directory creation fails
+     */
+    private static void createHudiTablePath(Configuration config) throws IOException {
+        String tablePath = config.get(FlinkOptions.PATH);
+        Path path = Paths.get(tablePath);
+        if (!Files.exists(path)) {
+            Files.createDirectories(path);
+            LOG.info("Created physical directory for Hudi table at: {}", tablePath);
+        } else {
+            LOG.debug("Hudi table directory already exists at: {}", tablePath);
+        }
+    }
+
     private Configuration createTableSpecificConfig(TableId tableId) {
         Configuration tableConfig = new Configuration(baseConfig);
         String coordinatorPath = baseConfig.getString(FlinkOptions.PATH);

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

@@ -17,7 +17,6 @@
 
 package org.apache.flink.cdc.connectors.hudi.sink.event;
 
-import org.apache.flink.cdc.common.data.RecordData;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 import org.apache.flink.cdc.common.event.Event;
@@ -32,7 +31,6 @@
 
 import java.time.ZoneId;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 /**
@@ -56,9 +54,6 @@ public class HudiRecordEventSerializer implements HudiRecordSerializer<Event> {
     /** Schema cache per table - populated from CreateTableEvent and SchemaChangeEvent. */
     private final Map<TableId, Schema> schemaMaps;
 
-    /** Field getter cache per table for efficient conversion. */
-    private final Map<TableId, List<RecordData.FieldGetter>> fieldGetterCache;
-
     /** RowDataKeyGen cache per table for key and partition extraction. */
     private final Map<TableId, RowDataKeyGen> keyGenCache;
 
@@ -67,7 +62,6 @@ public class HudiRecordEventSerializer implements HudiRecordSerializer<Event> {
 
     public HudiRecordEventSerializer(ZoneId zoneId) {
         this.schemaMaps = new HashMap<>();
-        this.fieldGetterCache = new HashMap<>();
         this.keyGenCache = new HashMap<>();
         this.zoneId = zoneId;
     }
@@ -87,8 +81,7 @@ public HoodieFlinkInternalRow serialize(Event event, String fileId, String insta
         if (event instanceof CreateTableEvent) {
             CreateTableEvent createTableEvent = (CreateTableEvent) event;
             schemaMaps.put(createTableEvent.tableId(), createTableEvent.getSchema());
-            // Clear caches for this table since schema changed
-            fieldGetterCache.remove(createTableEvent.tableId());
+            // Clear keyGenCache for this table since schema changed
             keyGenCache.remove(createTableEvent.tableId());
             // Schema events don't produce records
             return null;
@@ -102,8 +95,7 @@ public HoodieFlinkInternalRow serialize(Event event, String fileId, String insta
                 Schema newSchema =
                         SchemaUtils.applySchemaChangeEvent(existingSchema, schemaChangeEvent);
                 schemaMaps.put(schemaChangeEvent.tableId(), newSchema);
-                // Clear caches for this table since schema changed
-                fieldGetterCache.remove(schemaChangeEvent.tableId());
+                // Clear keyGenCache for this table since schema changed
                 keyGenCache.remove(schemaChangeEvent.tableId());
             }
             // Schema events don't produce records
@@ -180,7 +172,6 @@ public void setSchema(TableId tableId, Schema schema) {
         schemaMaps.put(tableId, schema);
         // Clear cached field getters and key gens for this table so they get recreated with the new
         // schema
-        fieldGetterCache.remove(tableId);
         keyGenCache.remove(tableId);
     }
 }

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

@@ -18,8 +18,12 @@
 package org.apache.flink.cdc.connectors.hudi.sink.function;
 
 import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.FlushEvent;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
 
 /**
  * Template interface for processing CDC events in a standardized way. Provides a consistent event
@@ -29,22 +33,34 @@
  * clear separation of concerns:
  *
  * <ul>
- *   <li>{@link #processDataChange(DataChangeEvent)} - Handles INSERT/UPDATE/DELETE operations
+ *   <li>{@link #processDataChange(DataChangeEvent, ProcessFunction.Context, Collector)} - Handles
+ *       DML operations (INSERT, UPDATE, DELETE)
  *   <li>{@link #processSchemaChange(SchemaChangeEvent)} - Handles DDL operations (CREATE TABLE, ADD
  *       COLUMN, etc.)
  *   <li>{@link #processFlush(FlushEvent)} - Handles coordinated flushing of buffered data
  * </ul>
+ *
+ * <p>Implementations of this interface are used in multi-table CDC pipelines to route and process
+ * events from different source tables to their corresponding Hudi tables.
+ *
+ * @see MultiTableEventStreamWriteFunction
  */
 public interface EventProcessorFunction {
 
     /**
-     * Process data change events (INSERT/UPDATE/DELETE). This is where actual data is buffered and
-     * written.
+     * Process data change events (INSERT, UPDATE, DELETE operations).
      *
-     * @param event The data change event
-     * @throws Exception if processing fails
+     * <p>This method handles DML operations from the CDC stream, converting them into Hudi records
+     * and collecting them for writing to the appropriate table.
+     *
+     * @param event The data change event containing the operation type and data
+     * @param ctx The process function context for accessing runtime information
+     * @param out The collector for emitting processed RowData records
      */
-    void processDataChange(DataChangeEvent event) throws Exception;
+    void processDataChange(
+            DataChangeEvent event,
+            ProcessFunction<Event, RowData>.Context ctx,
+            Collector<RowData> out);
 
     /**
      * Process schema change events (CREATE TABLE, ADD COLUMN, etc.).