Checkpoint 51 - Refactor and remove Event*Functions

Author: voonhous

.idea/vcs.xml

@@ -31,7 +31,7 @@ limitations under the License.
     <name>flink-cdc-pipeline-connector-hudi</name>
 
     <properties>
-        <hudi.version>1.1.0-SNAPSHOT</hudi.version>
+        <hudi.version>1.2.0-SNAPSHOT</hudi.version>
         <hadoop.version>2.10.2</hadoop.version>
         <mockito.version>3.4.6</mockito.version>
         <metrics.version>4.1.1</metrics.version>

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

@@ -22,6 +22,7 @@
 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.util.RowDataUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.jobgraph.OperatorID;
@@ -453,6 +454,8 @@ public void handleEventFromOperator(
                 () -> {
                     if (operatorEvent instanceof CreateTableOperatorEvent) {
                         handleCreateTableEvent((CreateTableOperatorEvent) operatorEvent);
+                    } else if (operatorEvent instanceof SchemaChangeOperatorEvent) {
+                        handleSchemaChangeEvent((SchemaChangeOperatorEvent) operatorEvent);
                     } else if (operatorEvent instanceof EnhancedWriteMetadataEvent) {
                         handleEnhancedWriteMetadataEvent(
                                 (EnhancedWriteMetadataEvent) operatorEvent);
@@ -507,6 +510,71 @@ private void handleCreateTableEvent(CreateTableOperatorEvent createTableOperator
                 });
     }
 
+    /**
+     * Handles schema change events from the sink functions. Updates the cached schema and recreates
+     * the write client to ensure it uses the new schema.
+     *
+     * @param event The schema change event containing the table ID and new schema
+     */
+    private void handleSchemaChangeEvent(SchemaChangeOperatorEvent event) {
+        TableId tableId = event.getTableId();
+        Schema newSchema = event.getNewSchema();
+
+        LOG.info(
+                "Received schema change event for table {}: {} columns",
+                tableId,
+                newSchema.getColumnCount());
+
+        // Update the cached schema
+        tableSchemas.put(tableId, newSchema);
+        LOG.info("Updated coordinator's schema cache for table: {}", tableId);
+
+        // Get the existing table context
+        TableContext oldContext = tableContexts.get(tableId);
+        if (oldContext == null) {
+            LOG.warn(
+                    "Received schema change for unknown table: {}. Skipping write client update.",
+                    tableId);
+            return;
+        }
+
+        try {
+            // Close the old write client
+            if (oldContext.writeClient != null) {
+                oldContext.writeClient.close();
+                LOG.info("Closed old write client for table: {}", tableId);
+            }
+
+            // Create new configuration with updated schema
+            Configuration tableConfig = createTableSpecificConfig(tableId);
+
+            // Create new write client with updated schema
+            HoodieFlinkWriteClient<?> newWriteClient =
+                    FlinkWriteClients.createWriteClient(tableConfig);
+            LOG.info("Created new write client with updated schema for table: {}", tableId);
+
+            // Update the table context with the new write client
+            // Keep the same eventBuffers, tableState, and tablePath
+            TableContext newContext =
+                    new TableContext(
+                            newWriteClient,
+                            oldContext.eventBuffers,
+                            oldContext.tableState,
+                            oldContext.tablePath);
+            tableContexts.put(tableId, newContext);
+
+            LOG.info("Successfully updated write client for table {} after schema change", tableId);
+        } catch (Exception e) {
+            LOG.error("Failed to update write client for table {} after schema change", tableId, e);
+            context.failJob(
+                    new HoodieException(
+                            "Failed to update write client for table "
+                                    + tableId
+                                    + " after schema change",
+                            e));
+        }
+    }
+
     private void handleEnhancedWriteMetadataEvent(EnhancedWriteMetadataEvent enhancedEvent) {
         String tablePath = enhancedEvent.getTablePath();
         WriteMetadataEvent event = enhancedEvent.getOriginalEvent();

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

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.hudi.sink.event;
+
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+
+/**
+ * An operator event that encapsulates a schema change and the resulting new schema.
+ *
+ * <p>This event is sent from the {@code MultiTableEventStreamWriteFunction} to the {@code
+ * MultiTableStreamWriteOperatorCoordinator} to signal that a table's schema has changed in the CDC
+ * stream. The coordinator uses this event to update its cached schema and recreate the write client
+ * to ensure subsequent operations use the correct schema.
+ */
+public class SchemaChangeOperatorEvent implements OperatorEvent {
+
+    private static final long serialVersionUID = 1L;
+
+    private final TableId tableId;
+    private final Schema newSchema;
+
+    /**
+     * Constructs a new SchemaChangeOperatorEvent.
+     *
+     * @param tableId The ID of the table whose schema changed
+     * @param newSchema The new schema after applying the schema change
+     */
+    public SchemaChangeOperatorEvent(TableId tableId, Schema newSchema) {
+        this.tableId = tableId;
+        this.newSchema = newSchema;
+    }
+
+    /**
+     * Gets the ID of the table whose schema changed.
+     *
+     * @return The table ID
+     */
+    public TableId getTableId() {
+        return tableId;
+    }
+
+    /**
+     * Gets the new schema after the change.
+     *
+     * @return The new schema
+     */
+    public Schema getNewSchema() {
+        return newSchema;
+    }
+
+    @Override
+    public String toString() {
+        return "SchemaChangeOperatorEvent{"
+                + "tableId="
+                + tableId
+                + ", newSchema columns="
+                + newSchema.getColumnCount()
+                + '}';
+    }
+}