refactor: move server.id for MySQL CDC from meta to fe

motivation is to simplify create stream job process in meta Signed-off-by: xxchan <xxchan22f@gmail.com>
risingwavelabs · Oct 29, 2024 · 3f64c85 · 3f64c85
1 parent 193e93f
commit 3f64c85
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 22 deletions.
diff --git a/src/connector/src/with_options.rs b/src/connector/src/with_options.rs
@@ -18,6 +18,7 @@ use risingwave_pb::secret::PbSecretRef;
 
 use crate::sink::catalog::SinkFormatDesc;
 use crate::source::cdc::external::CdcTableType;
+use crate::source::cdc::MYSQL_CDC_CONNECTOR;
 use crate::source::iceberg::ICEBERG_CONNECTOR;
 use crate::source::{
     AZBLOB_CONNECTOR, GCS_CONNECTOR, KAFKA_CONNECTOR, OPENDAL_S3_CONNECTOR, POSIX_FS_CONNECTOR,
@@ -104,6 +105,14 @@ pub trait WithPropertiesExt: Get + Sized {
         connector == KAFKA_CONNECTOR
     }
 
+    #[inline(always)]
+    fn is_mysql_cdc_connector(&self) -> bool {
+        let Some(connector) = self.get_connector() else {
+            return false;
+        };
+        connector == MYSQL_CDC_CONNECTOR
+    }
+
     #[inline(always)]
     fn is_cdc_connector(&self) -> bool {
         let Some(connector) = self.get_connector() else {

diff --git a/src/frontend/src/handler/create_source.rs b/src/frontend/src/handler/create_source.rs
@@ -21,6 +21,7 @@ use either::Either;
 use itertools::Itertools;
 use maplit::{convert_args, hashmap};
 use pgwire::pg_response::{PgResponse, StatementType};
+use rand::Rng;
 use risingwave_common::array::arrow::{arrow_schema_iceberg, IcebergArrowConvert};
 use risingwave_common::bail_not_implemented;
 use risingwave_common::catalog::{

diff --git a/src/frontend/src/handler/create_table.rs b/src/frontend/src/handler/create_table.rs
@@ -21,6 +21,7 @@ use either::Either;
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
+use rand::Rng;
 use risingwave_common::bail_not_implemented;
 use risingwave_common::catalog::{
     CdcTableDesc, ColumnCatalog, ColumnDesc, TableId, TableVersionId, DEFAULT_SCHEMA_NAME,
@@ -896,6 +897,13 @@ fn derive_with_options_for_cdc_table(
                 })?;
                 with_options.insert(DATABASE_NAME_KEY.into(), db_name.into());
                 with_options.insert(TABLE_NAME_KEY.into(), table_name.into());
+                // Generate a random server id for mysql cdc source if needed
+
+                // `server.id` (in the range from 1 to 2^32 - 1). This value MUST be unique across whole replication
+                // group (that is, different from any other server id being used by any master or slave)
+                with_options
+                    .entry("server.id".to_string())
+                    .or_insert(rand::thread_rng().gen_range(1..u32::MAX).to_string());
             }
             POSTGRES_CDC_CONNECTOR => {
                 let (schema_name, table_name) = external_table_name

diff --git a/src/meta/src/rpc/ddl_controller.rs b/src/meta/src/rpc/ddl_controller.rs
@@ -20,7 +20,6 @@ use std::time::Duration;
 
 use anyhow::{anyhow, Context};
 use itertools::Itertools;
-use rand::Rng;
 use risingwave_common::bitmap::Bitmap;
 use risingwave_common::config::DefaultParallelism;
 use risingwave_common::hash::{ActorMapping, VnodeCountCompat};
@@ -32,10 +31,8 @@ use risingwave_common::util::stream_graph_visitor::{
 };
 use risingwave_common::{bail, hash, must_match};
 use risingwave_connector::error::ConnectorError;
-use risingwave_connector::source::cdc::CdcSourceType;
 use risingwave_connector::source::{
     ConnectorProperties, SourceEnumeratorContext, SourceProperties, SplitEnumerator,
-    UPSTREAM_SOURCE_KEY,
 };
 use risingwave_connector::{dispatch_source_prop, WithOptionsSecResolved};
 use risingwave_meta_model::object::ObjectType;
@@ -2009,25 +2006,6 @@ pub fn fill_table_stream_graph_info(
                     source_node.source_inner.as_mut().unwrap().source_id = source.id;
                     source_count += 1;
 
-                    // Generate a random server id for mysql cdc source if needed
-                    // `server.id` (in the range from 1 to 2^32 - 1). This value MUST be unique across whole replication
-                    // group (that is, different from any other server id being used by any master or slave)
-                    if let Some(connector) = source.with_properties.get(UPSTREAM_SOURCE_KEY)
-                        && matches!(
-                            CdcSourceType::from(connector.as_str()),
-                            CdcSourceType::Mysql
-                        )
-                    {
-                        let props = &mut source_node.source_inner.as_mut().unwrap().with_properties;
-                        let rand_server_id = rand::thread_rng().gen_range(1..u32::MAX);
-                        props
-                            .entry("server.id".to_string())
-                            .or_insert(rand_server_id.to_string());
-
-                        // make these two `Source` consistent
-                        props.clone_into(&mut source.with_properties);
-                    }
-
                     assert_eq!(
                         source_count, 1,
                         "require exactly 1 external stream source when creating table with a connector"