Checkpoint 61 - fix table initialization

Author: cshuo

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

@@ -20,9 +20,11 @@
 import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.schema.Schema;
-import org.apache.flink.cdc.connectors.hudi.sink.event.CreateTableOperatorEvent;
 import org.apache.flink.cdc.connectors.hudi.sink.event.EnhancedWriteMetadataEvent;
 import org.apache.flink.cdc.connectors.hudi.sink.event.SchemaChangeOperatorEvent;
+import org.apache.flink.cdc.connectors.hudi.sink.event.TableAwareCorrespondent.CreateTableRequest;
+import org.apache.flink.cdc.connectors.hudi.sink.event.TableAwareCorrespondent.CreateTableResponse;
+import org.apache.flink.cdc.connectors.hudi.sink.event.TableAwareCorrespondent.MultiTableInstantTimeRequest;
 import org.apache.flink.cdc.connectors.hudi.sink.util.RowDataUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -96,29 +98,6 @@ public class MultiTableStreamWriteOperatorCoordinator extends StreamWriteOperato
     private static final Logger LOG =
             LoggerFactory.getLogger(MultiTableStreamWriteOperatorCoordinator.class);
 
-    /**
-     * A custom coordination request that includes the TableId to request an instant for a specific
-     * table.
-     */
-    public static class MultiTableInstantTimeRequest implements CoordinationRequest, Serializable {
-        private static final long serialVersionUID = 1L;
-        private final long checkpointId;
-        private final TableId tableId;
-
-        public MultiTableInstantTimeRequest(long checkpointId, TableId tableId) {
-            this.checkpointId = checkpointId;
-            this.tableId = tableId;
-        }
-
-        public long getCheckpointId() {
-            return checkpointId;
-        }
-
-        public TableId getTableId() {
-            return tableId;
-        }
-    }
-
     /**
      * Encapsulates all state and resources for a single table. This simplifies management by
      * grouping related objects, making the coordinator logic cleaner and less prone to errors.
@@ -406,9 +385,20 @@ public CompletableFuture<CoordinationResponse> handleCoordinationRequest(
                                         Correspondent.InstantTimeResponse.getInstance(
                                                 instantTime)));
                     },
-                    "handling instant time request for checkpoint %d",
+                    "Handling instant time request for checkpoint %d",
                     ((MultiTableInstantTimeRequest) request).getCheckpointId());
             return future;
+        } else if (request instanceof CreateTableRequest) {
+            CompletableFuture<CoordinationResponse> future = new CompletableFuture<>();
+            executor.execute(
+                    () -> {
+                        boolean isSuccess = handleCreateTableEvent(((CreateTableRequest) request));
+                        future.complete(
+                                CoordinationResponseSerDe.wrap(new CreateTableResponse(isSuccess)));
+                    },
+                    "Handling create table request: ",
+                    request);
+            return future;
         } else {
             LOG.warn("Received an unknown coordination request: {}", request.getClass().getName());
             return super.handleCoordinationRequest(request);
@@ -433,9 +423,7 @@ public void handleEventFromOperator(
             int subtask, int attemptNumber, OperatorEvent operatorEvent) {
         executor.execute(
                 () -> {
-                    if (operatorEvent instanceof CreateTableOperatorEvent) {
-                        handleCreateTableEvent((CreateTableOperatorEvent) operatorEvent);
-                    } else if (operatorEvent instanceof SchemaChangeOperatorEvent) {
+                    if (operatorEvent instanceof SchemaChangeOperatorEvent) {
                         handleSchemaChangeEvent((SchemaChangeOperatorEvent) operatorEvent);
                     } else if (operatorEvent instanceof EnhancedWriteMetadataEvent) {
                         handleEnhancedWriteMetadataEvent(
@@ -450,8 +438,8 @@ public void handleEventFromOperator(
                 operatorEvent);
     }
 
-    private void handleCreateTableEvent(CreateTableOperatorEvent createTableOperatorEvent) {
-        CreateTableEvent event = createTableOperatorEvent.getCreateTableEvent();
+    private boolean handleCreateTableEvent(CreateTableRequest createTableRequest) {
+        CreateTableEvent event = createTableRequest.getCreateTableEvent();
         TableId tableId = event.tableId();
 
         // Store the schema for this table
@@ -461,37 +449,44 @@ private void handleCreateTableEvent(CreateTableOperatorEvent createTableOperator
                 tableId,
                 event.getSchema().getColumnCount());
 
-        tableContexts.computeIfAbsent(
-                tableId,
-                tId -> {
-                    LOG.info("New table detected: {}. Initializing Hudi resources.", tId);
-                    try {
-                        Configuration tableConfig = createTableSpecificConfig(tId);
-                        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);
-                        TableState tableState = new TableState(tableConfig);
-                        EventBuffers eventBuffers = EventBuffers.getInstance(tableConfig);
-
-                        LOG.info(
-                                "Successfully initialized resources for table: {} at path: {}",
-                                tId,
-                                tablePath);
-                        return new TableContext(writeClient, eventBuffers, tableState, tablePath);
-                    } catch (Exception e) {
-                        LOG.error("Failed to initialize Hudi table resources for: {}", tId, e);
-                        context.failJob(
-                                new HoodieException(
-                                        "Failed to initialize Hudi writer for table " + tId, e));
-                        return null;
-                    }
-                });
+        TableContext tableContext =
+                tableContexts.computeIfAbsent(
+                        tableId,
+                        tId -> {
+                            LOG.info("New table detected: {}. Initializing Hudi resources.", tId);
+                            try {
+                                Configuration tableConfig = createTableSpecificConfig(tId);
+                                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);
+                                TableState tableState = new TableState(tableConfig);
+                                EventBuffers eventBuffers = EventBuffers.getInstance(tableConfig);
+
+                                LOG.info(
+                                        "Successfully initialized resources for table: {} at path: {}",
+                                        tId,
+                                        tablePath);
+                                return new TableContext(
+                                        writeClient, eventBuffers, tableState, tablePath);
+                            } catch (Exception e) {
+                                LOG.error(
+                                        "Failed to initialize Hudi table resources for: {}",
+                                        tId,
+                                        e);
+                                context.failJob(
+                                        new HoodieException(
+                                                "Failed to initialize Hudi writer for table " + tId,
+                                                e));
+                                return null;
+                            }
+                        });
+        return tableContext != null;
     }
 
     /**

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

@@ -18,10 +18,13 @@
 
 package org.apache.flink.cdc.connectors.hudi.sink.event;
 
+import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.connectors.hudi.sink.coordinator.MultiTableStreamWriteOperatorCoordinator;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway;
+import org.apache.flink.runtime.operators.coordination.CoordinationRequest;
+import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
 import org.apache.flink.util.SerializedValue;
 
 import org.apache.hudi.exception.HoodieException;
@@ -66,10 +69,8 @@ public static TableAwareCorrespondent getInstance(
     @Override
     public String requestInstantTime(long checkpointId) {
         try {
-            MultiTableStreamWriteOperatorCoordinator.MultiTableInstantTimeRequest request =
-                    new MultiTableStreamWriteOperatorCoordinator.MultiTableInstantTimeRequest(
-                            checkpointId, tableId);
-
+            MultiTableInstantTimeRequest request =
+                    new MultiTableInstantTimeRequest(checkpointId, tableId);
             Correspondent.InstantTimeResponse response =
                     CoordinationResponseSerDe.unwrap(
                             this.gateway
@@ -83,4 +84,89 @@ public String requestInstantTime(long checkpointId) {
                     e);
         }
     }
+
+    /**
+     * A custom coordination request that includes the TableId to request an instant for a specific
+     * table.
+     */
+    public static class MultiTableInstantTimeRequest implements CoordinationRequest {
+        private static final long serialVersionUID = 1L;
+        private final long checkpointId;
+        private final TableId tableId;
+
+        public MultiTableInstantTimeRequest(long checkpointId, TableId tableId) {
+            this.checkpointId = checkpointId;
+            this.tableId = tableId;
+        }
+
+        public long getCheckpointId() {
+            return checkpointId;
+        }
+
+        public TableId getTableId() {
+            return tableId;
+        }
+    }
+
+    /**
+     * Send a request to coordinator to create a hudi table.
+     *
+     * @param createTableEvent The creating table event.
+     * @return Whether the table is created successfully.
+     */
+    public boolean requestCreatingTable(CreateTableEvent createTableEvent) {
+        try {
+            CreateTableRequest request = new CreateTableRequest(createTableEvent);
+            CreateTableResponse response =
+                    CoordinationResponseSerDe.unwrap(
+                            this.gateway
+                                    .sendRequestToCoordinator(
+                                            this.operatorID, new SerializedValue<>(request))
+                                    .get());
+            return response.isSuccess();
+        } catch (Exception e) {
+            throw new HoodieException(
+                    "Error requesting the instant time from the coordinator for table " + tableId,
+                    e);
+        }
+    }
+
+    /**
+     * An CoordinationRequest that encapsulates a {@link CreateTableEvent}.
+     *
+     * <p>This request is sent from the {@code MultiTableEventStreamWriteFunction} to the {@code
+     * MultiTableStreamWriteOperatorCoordinator} to signal that a new table has been discovered in
+     * the CDC stream. The coordinator uses this event to initialize all necessary resources for the
+     * new table, such as its dedicated write client and event buffers, before any data is written.
+     */
+    public static class CreateTableRequest implements CoordinationRequest {
+        private static final long serialVersionUID = 1L;
+        private final CreateTableEvent createTableEvent;
+
+        public CreateTableRequest(CreateTableEvent createtableEvent) {
+            this.createTableEvent = createtableEvent;
+        }
+
+        public CreateTableEvent getCreateTableEvent() {
+            return createTableEvent;
+        }
+
+        @Override
+        public String toString() {
+            return "CreateTableRequest{" + "tableId=" + createTableEvent.tableId() + '}';
+        }
+    }
+
+    public static class CreateTableResponse implements CoordinationResponse {
+        private static final long serialVersionUID = 1L;
+        private final boolean success;
+
+        public CreateTableResponse(boolean success) {
+            this.success = success;
+        }
+
+        public boolean isSuccess() {
+            return success;
+        }
+    }
 }