Checkpoint 57 - Remove manual embedded timeline server management

Author: voonhous

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

@@ -39,12 +39,10 @@
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
 import org.apache.hudi.common.util.CommitUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.SerializationUtils;
 import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.configuration.OptionsResolver;
 import org.apache.hudi.exception.HoodieException;
@@ -275,12 +273,6 @@ boolean shouldTriggerCompaction() {
      */
     private transient SubtaskGateway[] gateways;
 
-    /**
-     * A dedicated write client whose only job is to run the embedded timeline server. This ensures
-     * there is only one timeline server for the entire job.
-     */
-    private transient HoodieFlinkWriteClient<?> timelineServerClient;
-
     /** A single-thread executor to handle instant time requests, mimicking the parent behavior. */
     private transient NonThrownExecutor instantRequestExecutor;
 
@@ -328,20 +320,6 @@ public void start() throws Exception {
         // Initialize the gateways array to avoid NullPointerException when subtasks are ready.
         this.gateways = new SubtaskGateway[context.currentParallelism()];
 
-        // Initialize a single write client for the coordinator path.
-        // Its primary role is to start and manage the embedded timeline server.
-        try {
-            // The baseConfig points to the dummy coordinator path.
-            // A .hoodie directory is required for the timeline server to start.
-            StreamerUtil.initTableIfNotExists(this.baseConfig);
-            this.timelineServerClient = FlinkWriteClients.createWriteClient(this.baseConfig);
-            LOG.info("Successfully started timeline server on coordinator.");
-        } catch (Exception e) {
-            LOG.error("Failed to start timeline server on coordinator.", e);
-            context.failJob(e);
-            return;
-        }
-
         // Re-initialize transient fields after deserialization from a Flink checkpoint.
         // When the coordinator is restored, the `tableContexts` map is deserialized, but all
         // `writeClient` fields within it will be null because they are transient.
@@ -903,9 +881,6 @@ private void scheduleTableServicesIfNeeded(TableId tableId, TableContext tableCo
 
     @Override
     public void close() throws Exception {
-        if (timelineServerClient != null) {
-            timelineServerClient.close();
-        }
         if (instantRequestExecutor != null) {
             instantRequestExecutor.close();
         }
@@ -948,9 +923,7 @@ private static void createHudiTablePath(Configuration config) throws IOException
 
     private Configuration createTableSpecificConfig(TableId tableId) {
         Configuration tableConfig = new Configuration(baseConfig);
-        String coordinatorPath = baseConfig.getString(FlinkOptions.PATH);
-        // Use the same logic as MultiTableEventStreamWriteFunction to strip "/coordinator"
-        String rootPath = coordinatorPath.split("/coordinator")[0];
+        String rootPath = baseConfig.getString(FlinkOptions.PATH);
         String tablePath =
                 String.format(
                         "%s/%s/%s", rootPath, tableId.getSchemaName(), tableId.getTableName());
@@ -974,13 +947,6 @@ private Configuration createTableSpecificConfig(TableId tableId) {
                     tableId);
         }
 
-        // Disable both embedded timeline server and metadata table for per-table clients.
-        // The central coordinator manages the only timeline server.
-        tableConfig.setBoolean(HoodieWriteConfig.EMBEDDED_TIMELINE_SERVER_ENABLE.key(), false);
-
-        // Use memory-based file system view since each client is lightweight.
-        tableConfig.setString(FileSystemViewStorageConfig.VIEW_TYPE.key(), "MEMORY");
-
         return tableConfig;
     }
 

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

@@ -52,12 +52,10 @@
 import org.apache.flink.util.OutputTag;
 
 import org.apache.hudi.client.model.HoodieFlinkInternalRow;
-import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.sink.common.AbstractStreamWriteFunction;
 import org.apache.hudi.sink.event.WriteMetadataEvent;
 import org.apache.hudi.util.AvroSchemaConverter;
-import org.apache.hudi.util.ViewStorageProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -107,7 +105,12 @@ public MultiTableEventStreamWriteFunction(Configuration config) {
 
     @Override
     public void initializeState(FunctionInitializationContext context) throws Exception {
-        super.initializeState(context);
+        // NOTE: Do NOT call super.initializeState(context) here.
+        // The parent class (AbstractStreamWriteFunction) expects to manage a single Hudi table
+        // and tries to create a HoodieTableMetaClient during initialization.
+        // MultiTableEventStreamWriteFunction is a dispatcher that manages multiple tables
+        // dynamically, so it doesn't have a single table path. Each child function
+        // (ExtendedBucketStreamWriteFunction) handles its own state initialization.
         this.functionInitializationContext = context;
 
         // Initialize schema map before restoring state
@@ -523,28 +526,13 @@ private Configuration createTableSpecificConfig(TableId tableId) {
                             AvroSchemaConverter.convertToSchema(rowType).toString();
                     localTableConfig.set(FlinkOptions.SOURCE_AVRO_SCHEMA, tableAvroSchema);
 
-                    String rootPath = this.config.get(FlinkOptions.PATH).split("/coordinator")[0];
-                    if (rootPath != null) {
-                        String tableBasePath =
-                                String.format(
-                                        "%s/%s/%s",
-                                        rootPath, tableId.getSchemaName(), tableId.getTableName());
-                        localTableConfig.set(FlinkOptions.PATH, tableBasePath);
-                    }
+                    String rootPath = this.config.get(FlinkOptions.PATH);
+                    String tableBasePath =
+                            String.format(
+                                    "%s/%s/%s",
+                                    rootPath, tableId.getSchemaName(), tableId.getTableName());
+                    localTableConfig.set(FlinkOptions.PATH, tableBasePath);
 
-                    // Modify ViewStorageProperties to point to coordinator table
-                    FileSystemViewStorageConfig viewStorageConfig =
-                            ViewStorageProperties.loadFromProperties(
-                                    this.config.get(FlinkOptions.PATH), localTableConfig);
-                    localTableConfig.setString(
-                            FileSystemViewStorageConfig.VIEW_TYPE.key(),
-                            viewStorageConfig.getStorageType().name());
-                    localTableConfig.setString(
-                            FileSystemViewStorageConfig.REMOTE_HOST_NAME.key(),
-                            viewStorageConfig.getRemoteViewServerHost());
-                    localTableConfig.setString(
-                            FileSystemViewStorageConfig.REMOTE_PORT_NUM.key(),
-                            viewStorageConfig.getRemoteViewServerPort() + "");
                     return localTableConfig;
                 });
     }
@@ -634,23 +622,27 @@ protected void flushRemaining(boolean endInput) {
 
     @Override
     public void close() throws Exception {
-        for (ExtendedBucketStreamWriteFunction func : tableFunctions.values()) {
-            try {
-                func.close();
-            } catch (Exception e) {
-                LOG.error("Failed to close table function", e);
+        if (tableFunctions != null) {
+            for (ExtendedBucketStreamWriteFunction func : tableFunctions.values()) {
+                try {
+                    func.close();
+                } catch (Exception e) {
+                    LOG.error("Failed to close table function", e);
+                }
             }
         }
         super.close();
     }
 
     public void endInput() {
         super.endInput();
-        for (ExtendedBucketStreamWriteFunction func : tableFunctions.values()) {
-            try {
-                func.endInput();
-            } catch (Exception e) {
-                LOG.error("Failed to complete endInput for table function", e);
+        if (tableFunctions != null) {
+            for (ExtendedBucketStreamWriteFunction func : tableFunctions.values()) {
+                try {
+                    func.endInput();
+                } catch (Exception e) {
+                    LOG.error("Failed to complete endInput for table function", e);
+                }
             }
         }
     }

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

@@ -25,7 +25,6 @@
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway;
 
-import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.sink.common.AbstractWriteOperator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -105,56 +104,8 @@ public void open() throws Exception {
      */
     public static MultiTableWriteOperatorFactory<Event> getFactory(
             Configuration conf, String schemaOperatorUid) {
-        // Create coordinator-specific configuration with dummy table settings
-        // This satisfies the coordinator's requirement for table initialization
-        Configuration coordinatorConfig = createCoordinatorConfig(conf);
-
         LOG.info("Creating multi-table write operator factory with extended coordinator support");
         return MultiTableWriteOperatorFactory.instance(
-                coordinatorConfig,
-                new MultiTableWriteOperator(coordinatorConfig, schemaOperatorUid));
-    }
-
-    /**
-     * Creates a coordinator-specific configuration with dummy table settings. This satisfies the
-     * coordinator's requirement for table initialization while actual table routing happens
-     * dynamically based on incoming events. Uses deterministic naming to allow reuse of existing
-     * coordinator tables.
-     */
-    private static Configuration createCoordinatorConfig(Configuration originalConfig) {
-        Configuration coordinatorConfig = new Configuration();
-        coordinatorConfig.addAll(originalConfig);
-
-        // Create deterministic dummy table name based on base path hash for reusability
-        String originalPath = coordinatorConfig.get(FlinkOptions.PATH, "default");
-        String pathHash = String.valueOf(Math.abs(originalPath.hashCode()));
-        String dummyTableName = "coordinator_" + pathHash;
-
-        coordinatorConfig.set(FlinkOptions.TABLE_NAME, dummyTableName);
-        coordinatorConfig.set(FlinkOptions.DATABASE_NAME, "coordinator_db");
-
-        // Set deterministic path for coordinator table (allows reuse)
-        String coordinatorPath = originalPath + "/coordinator/" + dummyTableName;
-        coordinatorConfig.set(FlinkOptions.PATH, coordinatorPath);
-
-        // Set dummy Avro schema with a simple structure (id: int)
-        String dummyAvroSchema =
-                "{\n"
-                        + "  \"type\": \"record\",\n"
-                        + "  \"name\": \"coordinator_record\",\n"
-                        + "  \"fields\": [\n"
-                        + "    {\n"
-                        + "      \"name\": \"id\",\n"
-                        + "      \"type\": \"int\"\n"
-                        + "    }\n"
-                        + "  ]\n"
-                        + "}";
-        coordinatorConfig.set(FlinkOptions.SOURCE_AVRO_SCHEMA, dummyAvroSchema);
-
-        LOG.info(
-                "Created coordinator config with reusable dummy table: coordinator_db.{} at path: {}",
-                dummyTableName,
-                coordinatorPath);
-        return coordinatorConfig;
+                conf, new MultiTableWriteOperator(conf, schemaOperatorUid));
     }
 }