WIP: modify source properties on meta

ci/scripts/e2e-source-test.sh

@@ -95,6 +95,7 @@ sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.check.slt'
 
 # kill cluster and the connector node
 cargo make kill
+echo "cluster killed "
 
 # insert new rows
 mysql --host=mysql --port=3306 -u root -p123456 < ./e2e_test/source/cdc/mysql_cdc_insert.sql

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/common/DbzConnectorConfig.java

@@ -22,7 +22,6 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
-import java.util.concurrent.atomic.AtomicLong;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.text.StringSubstitutor;
 import org.slf4j.Logger;
@@ -77,8 +76,6 @@ private static Map<String, String> extractDebeziumProperties(
         return dbzProps;
     }
 
-    private static final AtomicLong serverId = new AtomicLong(20220410);
-
     private final long sourceId;
 
     private final SourceTypeE sourceType;
@@ -130,8 +127,6 @@ public DbzConnectorConfig(
                 mysqlProps.setProperty("snapshot.locking.mode", "none");
             }
 
-            mysqlProps.setProperty("database.server.id", String.valueOf(serverId.getAndAdd(1L)));
-
             dbzProps.putAll(mysqlProps);
         } else if (source == SourceTypeE.POSTGRES || source == SourceTypeE.CITUS) {
             var postgresProps = initiateDbConfig(POSTGRES_CONFIG_FILE, substitutor);

src/connector/src/macros.rs

@@ -234,12 +234,13 @@ macro_rules! impl_cdc_source_type {
             $(
                 $cdc_source_type,
             )*
+            Unspecified,
         }
 
         impl From<PbSourceType> for CdcSourceType {
             fn from(value: PbSourceType) -> Self {
                 match value {
-                    PbSourceType::Unspecified => unreachable!(),
+                    PbSourceType::Unspecified => CdcSourceType::Unspecified,
                     $(
                         PbSourceType::$cdc_source_type => CdcSourceType::$cdc_source_type,
                     )*
@@ -253,8 +254,10 @@ macro_rules! impl_cdc_source_type {
                     $(
                         CdcSourceType::$cdc_source_type => PbSourceType::$cdc_source_type,
                     )*
+                   CdcSourceType::Unspecified => PbSourceType::Unspecified,
                 }
             }
         }
+
     }
 }

src/connector/src/source/base.rs

@@ -51,7 +51,7 @@ use crate::{
 
 const SPLIT_TYPE_FIELD: &str = "split_type";
 const SPLIT_INFO_FIELD: &str = "split_info";
-const UPSTREAM_SOURCE_KEY: &str = "connector";
+pub const UPSTREAM_SOURCE_KEY: &str = "connector";
 
 pub trait TryFromHashmap: Sized {
     fn try_from_hashmap(props: HashMap<String, String>) -> Result<Self>;

src/connector/src/source/cdc/mod.rs

@@ -42,6 +42,17 @@ pub trait CdcSourceTypeTrait: Send + Sync + Clone + 'static {
 
 for_all_classified_sources!(impl_cdc_source_type);
 
+impl<'a> From<&'a str> for CdcSourceType {
+    fn from(name: &'a str) -> Self {
+        match name {
+            MYSQL_CDC_CONNECTOR => CdcSourceType::Mysql,
+            POSTGRES_CDC_CONNECTOR => CdcSourceType::Postgres,
+            CITUS_CDC_CONNECTOR => CdcSourceType::Citus,
+            _ => CdcSourceType::Unspecified,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Default)]
 pub struct CdcProperties<T: CdcSourceTypeTrait> {
     /// Properties specified in the WITH clause by user

src/connector/src/source/cdc/source/reader.rs

@@ -87,6 +87,9 @@ impl<T: CdcSourceTypeTrait> SplitReader for CdcSplitReader<T> {
                 parser_config,
                 source_ctx,
             }),
+            CdcSourceType::Unspecified => {
+                unreachable!();
+            }
         }
     }
 

src/connector/src/source/mod.rs

@@ -24,7 +24,7 @@ pub mod monitor;
 pub mod nats;
 pub mod nexmark;
 pub mod pulsar;
-pub use base::*;
+pub use base::{UPSTREAM_SOURCE_KEY, *};
 pub(crate) use common::*;
 pub use google_pubsub::GOOGLE_PUBSUB_CONNECTOR;
 pub use kafka::KAFKA_CONNECTOR;

src/meta/service/src/ddl_service.rs

@@ -19,6 +19,8 @@ use anyhow::anyhow;
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
 use risingwave_common::util::stream_graph_visitor::visit_fragment;
 use risingwave_connector::sink::catalog::SinkId;
+use risingwave_connector::source::cdc::CdcSourceType;
+use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
 use risingwave_pb::catalog::connection::private_link_service::{
     PbPrivateLinkProvider, PrivateLinkProvider,
 };
@@ -428,6 +430,17 @@ impl DdlService for DdlServiceImpl {
             // Generate source id.
             let source_id = self.gen_unique_id::<{ IdCategory::Table }>().await?; // TODO: Use source category
             fill_table_source(source, source_id, &mut mview, table_id, &mut fragment_graph);
+
+            // Modify properties for cdc sources if needed
+            if let Some(connector) = source.properties.get(UPSTREAM_SOURCE_KEY) {
+                if matches!(
+                    CdcSourceType::from(connector.as_str()),
+                    CdcSourceType::Mysql
+                ) {
+                    let server_id = self.gen_unique_id_u64::<{ IdCategory::MySqlCdc }>().await?;
+                    fill_cdc_mysql_server_id(server_id, &mut fragment_graph);
+                }
+            }
         }
 
         let mut stream_job = StreamingJob::Table(source, mview);
@@ -740,6 +753,11 @@ impl DdlServiceImpl {
         Ok(id)
     }
 
+    async fn gen_unique_id_u64<const C: IdCategoryType>(&self) -> MetaResult<u64> {
+        let id = self.env.id_gen_manager().generate::<C>().await?;
+        Ok(id)
+    }
+
     async fn validate_connection(&self, connection_id: ConnectionId) -> MetaResult<()> {
         let connection = self
             .catalog_manager
@@ -798,3 +816,22 @@ fn fill_table_source(
     table.optional_associated_source_id =
         Some(OptionalAssociatedSourceId::AssociatedSourceId(source_id));
 }
+
+fn fill_cdc_mysql_server_id(server_id: u64, fragment_graph: &mut PbStreamFragmentGraph) {
+    for fragment in fragment_graph.fragments.values_mut() {
+        visit_fragment(fragment, |node_body| {
+            if let NodeBody::Source(source_node) = node_body {
+                let props = &mut source_node.source_inner.as_mut().unwrap().properties;
+
+                // Modify properties for cdc sources if needed
+                if props.contains_key("server.id") {
+                    return;
+                }
+                props
+                    .entry("server.id".to_string())
+                    .or_insert(server_id.to_string());
+                tracing::debug!("server.id no set, generate one {}", server_id);
+            }
+        });
+    }
+}

src/meta/src/manager/id.rs

@@ -136,6 +136,7 @@ pub mod IdCategory {
     pub const CompactionGroup: IdCategoryType = 15;
     pub const Function: IdCategoryType = 16;
     pub const Connection: IdCategoryType = 17;
+    pub const MySqlCdc: IdCategoryType = 18;
 }
 
 pub type IdGeneratorManagerRef = Arc<IdGeneratorManager>;
@@ -159,6 +160,7 @@ pub struct IdGeneratorManager {
     parallel_unit: Arc<StoredIdGenerator>,
     compaction_group: Arc<StoredIdGenerator>,
     connection: Arc<StoredIdGenerator>,
+    mysql_cdc: Arc<StoredIdGenerator>,
 }
 
 impl IdGeneratorManager {
@@ -215,6 +217,9 @@ impl IdGeneratorManager {
             connection: Arc::new(
                 StoredIdGenerator::new(meta_store.clone(), "connection", None).await,
             ),
+            mysql_cdc: Arc::new(
+                StoredIdGenerator::new(meta_store.clone(), "mysql_cdc", Some(20210401)).await,
+            ),
         }
     }
 
@@ -236,6 +241,7 @@ impl IdGeneratorManager {
             IdCategory::HummockCompactionTask => &self.hummock_compaction_task,
             IdCategory::CompactionGroup => &self.compaction_group,
             IdCategory::Connection => &self.connection,
+            IdCategory::MySqlCdc => &self.mysql_cdc,
             _ => unreachable!(),
         }
     }

src/source/src/connector_source.rs

@@ -155,7 +155,7 @@ impl ConnectorSource {
                 let to_reader_splits = splits.into_iter().map(|split| vec![split]);
 
                 try_join_all(to_reader_splits.into_iter().map(|splits| {
-                    tracing::debug!("spawning connector split reader for split {:?}", splits);
+                    tracing::debug!(?splits, ?prop, "spawning connector split reader");
                     let props = prop.clone();
                     let data_gen_columns = data_gen_columns.clone();
                     let parser_config = parser_config.clone();