feat: add ZooKeeper dual-active registry sync connector

Cargo.toml

@@ -71,6 +71,7 @@ opendal = { version = "0.54.1", features = ["services-s3"] }
 governor = "0.10.4"
 indexmap = "2"
 libc = "0.2"
+zookeeper-client = "0.8"
 
 [profile.release]
 panic = 'unwind'

dt-common/src/config/config_enums.rs

@@ -34,6 +34,8 @@ pub enum DbType {
     Doris,
     #[strum(serialize = "tidb")]
     Tidb,
+    #[strum(serialize = "zk")]
+    Zk,
 }
 
 #[derive(Display, EnumString, IntoStaticStr, Debug, Clone, Hash, PartialEq, Eq)]
@@ -106,6 +108,8 @@ pub enum ConflictPolicyEnum {
     #[default]
     #[strum(serialize = "interrupt")]
     Interrupt,
+    #[strum(serialize = "last_write_wins")]
+    LastWriteWins,
 }
 
 #[derive(Display, EnumString, IntoStaticStr, PartialEq)]

dt-common/src/config/extractor_config.rs

@@ -202,6 +202,13 @@ pub enum ExtractorConfig {
         offset: i64,
         ack_interval_secs: u64,
     },
+    Zk {
+        url: String,
+        watch_paths: Vec<String>,
+        scan_interval_secs: u64,
+        include_ephemeral: bool,
+        heartbeat_interval_secs: u64,
+    },
 }
 
 #[derive(Clone, Debug, Hash)]

dt-common/src/config/mod.rs

@@ -20,6 +20,7 @@ pub mod s3_config;
 pub mod sinker_config;
 pub mod ssl_config;
 pub mod task_config;
+pub mod zk_filter_config;
 
 #[cfg(feature = "metrics")]
 pub mod metrics_config;

dt-common/src/config/router_config.rs

@@ -6,4 +6,7 @@ pub enum RouterConfig {
         col_map: String,
         topic_map: String,
     },
+    Zk {
+        path_prefix_map: String,
+    },
 }

dt-common/src/config/sinker_config.rs

@@ -122,6 +122,14 @@ pub enum SinkerConfig {
     Sql {
         reverse: bool,
     },
+
+    Zk {
+        url: String,
+        batch_size: usize,
+        create_if_not_exists: bool,
+        sync_ephemeral_as_persistent: bool,
+        conflict_policy: ConflictPolicyEnum,
+    },
 }
 
 #[derive(Clone, Debug, Hash)]

dt-common/src/config/task_config.rs

@@ -42,6 +42,7 @@ use super::{
     runtime_config::RuntimeConfig,
     s3_config::S3Config,
     sinker_config::{BasicSinkerConfig, SinkerConfig},
+    zk_filter_config::ZkFilterConfig,
 };
 
 #[derive(Clone)]
@@ -61,6 +62,7 @@ pub struct TaskConfig {
     pub meta_center: Option<MetaCenterConfig>,
     pub data_marker: Option<DataMarkerConfig>,
     pub processor: Option<ProcessorConfig>,
+    pub zk_filter: ZkFilterConfig,
     #[cfg(feature = "metrics")]
     pub metrics: MetricsConfig,
 }
@@ -80,6 +82,7 @@ const FILTER: &str = "filter";
 const ROUTER: &str = "router";
 const RESUMER: &str = "resumer";
 const DATA_MARKER: &str = "data_marker";
+const ZK_FILTER: &str = "zk_filter";
 const PROCESSOR: &str = "processor";
 const CHECKER: &str = "checker";
 const META_CENTER: &str = "metacenter";
@@ -256,6 +259,7 @@ impl TaskConfig {
             checker,
             data_marker: Self::load_data_marker_config(&loader)?,
             processor: Self::load_processor_config(&loader)?,
+            zk_filter: Self::load_zk_filter_config(&loader),
             meta_center: Self::load_meta_center_config(&loader)?,
             #[cfg(feature = "metrics")]
             metrics: Self::load_metrics_config(&loader)?,
@@ -748,6 +752,33 @@ impl TaskConfig {
                 ack_interval_secs: loader.get_optional(EXTRACTOR, "ack_interval_secs"),
             },
 
+            DbType::Zk => {
+                let watch_paths_str: String = loader.get_optional(EXTRACTOR, "watch_paths");
+                let watch_paths = if watch_paths_str.is_empty() {
+                    vec!["/".to_string()]
+                } else {
+                    watch_paths_str
+                        .split(',')
+                        .map(|s| s.trim().to_string())
+                        .collect()
+                };
+                ExtractorConfig::Zk {
+                    url,
+                    watch_paths,
+                    scan_interval_secs: loader.get_with_default(
+                        EXTRACTOR,
+                        "scan_interval_secs",
+                        60,
+                    ),
+                    include_ephemeral: loader.get_with_default(
+                        EXTRACTOR,
+                        "include_ephemeral",
+                        false,
+                    ),
+                    heartbeat_interval_secs,
+                }
+            }
+
             db_type => {
                 bail! {Error::ConfigError(format!(
                     "extractor db type: {} not supported",
@@ -995,6 +1026,24 @@ impl TaskConfig {
 
                 _ => bail! { not_supported_err },
             },
+            DbType::Zk => match sink_type {
+                SinkType::Write => SinkerConfig::Zk {
+                    url,
+                    batch_size,
+                    create_if_not_exists: loader.get_with_default(
+                        SINKER,
+                        "create_if_not_exists",
+                        true,
+                    ),
+                    sync_ephemeral_as_persistent: loader.get_with_default(
+                        SINKER,
+                        "sync_ephemeral_as_persistent",
+                        false,
+                    ),
+                    conflict_policy,
+                },
+                _ => bail! { not_supported_err },
+            },
         };
         Ok((basic, sinker))
     }
@@ -1332,6 +1381,18 @@ impl TaskConfig {
         }
     }
 
+    fn load_zk_filter_config(loader: &IniLoader) -> ZkFilterConfig {
+        ZkFilterConfig {
+            do_paths: loader.get_optional(ZK_FILTER, "do_paths"),
+            ignore_paths: loader.get_with_default(
+                ZK_FILTER,
+                "ignore_paths",
+                "/zookeeper".to_string(),
+            ),
+            include_ephemeral: loader.get_with_default(ZK_FILTER, "include_ephemeral", false),
+        }
+    }
+
     fn load_data_marker_config(loader: &IniLoader) -> anyhow::Result<Option<DataMarkerConfig>> {
         if !loader.ini.sections().contains(&DATA_MARKER.to_string()) {
             return Ok(None);

dt-common/src/config/zk_filter_config.rs

@@ -0,0 +1,6 @@
+#[derive(Clone, Default, Hash)]
+pub struct ZkFilterConfig {
+    pub do_paths: String,
+    pub ignore_paths: String,
+    pub include_ephemeral: bool,
+}

dt-common/src/lib.rs

@@ -9,3 +9,4 @@ pub mod rdb_filter;
 pub mod system_dbs;
 pub mod time_filter;
 pub mod utils;
+pub mod zk_filter;

dt-common/src/meta/dt_data.rs

@@ -1,7 +1,10 @@
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 
-use super::{ddl_meta::ddl_data::DdlData, row_data::RowData, struct_meta::struct_data::StructData};
+use super::{
+    ddl_meta::ddl_data::DdlData, row_data::RowData, struct_meta::struct_data::StructData,
+    zk::zk_entry::ZkEntry,
+};
 use crate::meta::dcl_meta::dcl_data::DclData;
 use crate::meta::row_type::RowSqlType;
 use crate::meta::{position::Position, redis::redis_entry::RedisEntry};
@@ -58,6 +61,9 @@ pub enum DtData {
     Redis {
         entry: RedisEntry,
     },
+    Zk {
+        entry: ZkEntry,
+    },
 }
 
 impl DtData {
@@ -83,6 +89,7 @@ impl DtData {
             DtData::Dcl { dcl_data } => dcl_data.get_malloc_size(),
             DtData::Ddl { ddl_data } => ddl_data.get_malloc_size(),
             DtData::Redis { entry } => entry.get_data_malloc_size() as u64,
+            DtData::Zk { entry } => entry.get_data_size(),
             // ignore other item types
             _ => 0,
         }

dt-common/src/meta/mod.rs

@@ -23,3 +23,4 @@ pub mod struct_meta;
 pub mod syncer;
 pub mod tagged_col_value_map;
 pub mod time;
+pub mod zk;

dt-common/src/meta/position.rs

@@ -1,4 +1,4 @@
-use std::str::FromStr;
+use std::{collections::HashMap, str::FromStr};
 
 use anyhow::Context;
 use chrono::{DateTime, NaiveDateTime};
@@ -56,6 +56,12 @@ pub enum Position {
         now_db_id: i64,
         timestamp: String,
     },
+    Zk {
+        path_versions: HashMap<String, (i32, i64)>,
+        last_scan_timestamp: i64,
+        high_water_zxid: i64,
+        total_paths: usize,
+    },
 }
 
 impl Position {
@@ -123,6 +129,10 @@ impl Position {
                 }
                 0
             }
+            Position::Zk {
+                last_scan_timestamp,
+                ..
+            } => *last_scan_timestamp as u64,
             _ => 0,
         }
     }

dt-common/src/meta/zk/mod.rs

@@ -0,0 +1,3 @@
+pub mod zk_entry;
+pub mod zk_event_type;
+pub mod zk_stat;

dt-common/src/meta/zk/zk_entry.rs

@@ -0,0 +1,41 @@
+use serde::{Deserialize, Serialize};
+
+use super::{zk_event_type::ZkEventType, zk_stat::ZkStat};
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
+pub enum ZkOrderOrigin {
+    #[default]
+    SourceStatMtime,
+    WatcherObserved,
+    ReconcileObserved,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ZkEntry {
+    pub path: String,
+    pub data: Option<Vec<u8>>,
+    pub stat: ZkStat,
+    pub ephemeral: bool,
+    pub event_type: ZkEventType,
+    pub source_id: String,
+    #[serde(default)]
+    pub source_order_millis: i64,
+    #[serde(default)]
+    pub source_zxid: i64,
+    #[serde(default)]
+    pub order_origin: ZkOrderOrigin,
+}
+
+impl ZkEntry {
+    pub fn get_data_size(&self) -> u64 {
+        self.path.len() as u64 + self.data.as_ref().map_or(0, |d| d.len() as u64)
+    }
+
+    pub fn source_order_millis(&self) -> i64 {
+        if self.source_order_millis != 0 {
+            self.source_order_millis
+        } else {
+            self.stat.mtime
+        }
+    }
+}

dt-common/src/meta/zk/zk_event_type.rs

@@ -0,0 +1,9 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub enum ZkEventType {
+    Created,
+    Updated,
+    Deleted,
+    ChildrenChanged,
+}

dt-common/src/meta/zk/zk_stat.rs

@@ -0,0 +1,13 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct ZkStat {
+    pub version: i32,
+    pub cversion: i32,
+    pub aversion: i32,
+    pub mtime: i64,
+    pub ctime: i64,
+    pub czxid: i64,
+    pub mzxid: i64,
+    pub ephemeral_owner: i64,
+}