diff --git a/Cargo.lock b/Cargo.lock index 7893448f..b7c78f37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -39,7 +39,7 @@ dependencies = [ "flate2", "foldhash 0.1.5", "futures-core", - "h2 0.3.26", + "h2 0.3.27", "http 0.2.12", "httparse", "httpdate", @@ -65,7 +65,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e01ed3140b2f8d422c68afa1ed2e85d996ea619c988ac834d255db32138655cb" dependencies = [ "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -182,7 +182,7 @@ dependencies = [ "actix-router", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -351,8 +351,8 @@ dependencies = [ "regex-lite", "serde", "serde_json", - "strum", - "strum_macros", + "strum 0.25.0", + "strum_macros 0.25.3", "thiserror 1.0.69", "typed-builder 0.16.2", "uuid", @@ -368,7 +368,7 @@ dependencies = [ "proc-macro2", "quote", "serde_json", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -530,7 +530,7 @@ checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -586,7 +586,7 @@ checksum = "e539d3fca749fcee5236ab05e93a52867dd549cc157c8cb7f99595f3cedffdb5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -619,6 +619,28 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-lc-rs" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", +] + [[package]] name = "axum" version = "0.8.9" @@ -740,7 +762,7 @@ dependencies = [ "regex", "rustc-hash", "shlex", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -824,10 +846,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdd1d3c0c2f5833f22386f252fe8ed005c7f59fdcddeef025c01b4c3b9fd9ac3" dependencies = [ "once_cell", - "proc-macro-crate", + "proc-macro-crate 3.3.0", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -948,6 +970,15 @@ dependencies = [ "bytes", ] +[[package]] +name = "castaway" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" +dependencies = [ + "rustversion", +] + [[package]] name = "cc" version = "1.2.27" @@ -1051,7 +1082,7 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1095,7 +1126,7 @@ dependencies = [ "proc-macro2", "quote", "serde_derive_internals", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1127,6 +1158,17 @@ dependencies = [ "tokio-util 0.7.15", ] +[[package]] +name = "compact_str" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e33b5c3ee2b4ffa00ac2b00d1645cd9229ade668139bccf95f15fadcf374127b" +dependencies = [ + "castaway", + "itoa", + "ryu", +] + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -1208,6 +1250,27 @@ dependencies = [ "tiny-keccak", ] +[[package]] +name = "const_format" +version = "0.2.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4481a617ad9a412be3b97c5d403fef8ed023103368908b9c50af598ff467cc1e" +dependencies = [ + "const_format_proc_macros", + "konst", +] + +[[package]] +name = "const_format_proc_macros" +version = "0.2.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d57c2eccfb16dbac1f4e61e206105db5820c9d26c3c472bc17c774259ef7744" +dependencies = [ + "proc-macro2", + "quote", + "unicode-xid", +] + [[package]] name = "cookie" version = "0.16.2" @@ -1382,7 +1445,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1396,7 +1459,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1407,7 +1470,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ "darling_core 0.20.11", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1418,7 +1481,7 @@ checksum = "d38308df82d1080de0afee5d069fa14b0326a88c14f15c5ccda35b4a6c414c81" dependencies = [ "darling_core 0.21.3", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1486,7 +1549,7 @@ checksum = "d65d7ce8132b7c0e54497a4d9a55a1c2a0912a0d786cf894472ba818fba45762" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1497,7 +1560,7 @@ checksum = "d08b3a0bcc0d079199cd476b2cae8435016ec11d1c0986c6901c5ac223041534" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1517,7 +1580,7 @@ checksum = "bda628edc44c4bb645fbe0f758797143e4e07926f7ebf4e9bdfbd3d2ce621df3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", "unicode-xid", ] @@ -1562,7 +1625,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -1621,7 +1684,7 @@ dependencies = [ "serde_json", "serde_yaml", "sqlx", - "strum", + "strum 0.25.0", "thiserror 1.0.69", "tokio", "url", @@ -1667,12 +1730,13 @@ dependencies = [ "serde", "serde_json", "sqlx", - "strum", + "strum 0.25.0", "thiserror 1.0.69", "tokio", "tokio-postgres", "url", "uuid", + "zookeeper-client", ] [[package]] @@ -1742,7 +1806,7 @@ dependencies = [ "redis", "regex", "sqlx", - "strum", + "strum 0.25.0", "tokio", "url", ] @@ -1780,7 +1844,7 @@ dependencies = [ "serde_yaml", "serial_test", "sqlx", - "strum", + "strum 0.25.0", "tokio", ] @@ -1846,9 +1910,15 @@ dependencies = [ "darling 0.20.11", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "either" version = "1.15.0" @@ -1876,7 +1946,7 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -2073,7 +2143,7 @@ checksum = "a0b4095fc99e1d858e5b8c7125d2638372ec85aa0fe6c807105cf10b0265ca6c" dependencies = [ "frunk_proc_macro_helpers", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -2085,7 +2155,7 @@ dependencies = [ "frunk_core", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -2097,9 +2167,15 @@ dependencies = [ "frunk_core", "frunk_proc_macro_helpers", "quote", - "syn 2.0.117", + "syn 2.0.103", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "funty" version = "2.0.0" @@ -2201,7 +2277,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -2414,6 +2490,15 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hashlink" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7" +dependencies = [ + "hashbrown 0.14.5", +] + [[package]] name = "hashlink" version = "0.10.0" @@ -2436,6 +2521,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "heck" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d621efb26863f0e9924c6ac577e8275e5e6b77455db64ffa6c65c904e9e132c" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "heck" version = "0.4.1" @@ -2633,7 +2727,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "h2 0.3.26", + "h2 0.3.27", "http 0.2.12", "http-body 0.4.6", "httparse", @@ -2872,6 +2966,12 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "ignore-result" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "665ff4dce8edd10d490641ccb78949832f1ddbff02c584fb1f85ab888fe0e50c" + [[package]] name = "impl-more" version = "0.1.9" @@ -3005,6 +3105,21 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "konst" +version = "0.2.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "128133ed7824fcd73d6e7b17957c5eb7bacb885649bd8c69708b2331a10bcefb" +dependencies = [ + "konst_macro_rules", +] + +[[package]] +name = "konst_macro_rules" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4933f3f57a8e9d9da04db23fb153356ecaf00cbd14aee46279c33dc80925c37" + [[package]] name = "kv-log-macro" version = "1.0.7" @@ -3236,7 +3351,7 @@ dependencies = [ "macro_magic_core", "macro_magic_macros", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3250,7 +3365,7 @@ dependencies = [ "macro_magic_core_macros", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3261,7 +3376,7 @@ checksum = "b02abfe41815b5bd98dbd4260173db2c116dda171dc0fe7838cb206333b83308" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3272,7 +3387,7 @@ checksum = "73ea28ee64b88876bf45277ed9a5817c1817df061a74f2b988971a12570e5869" dependencies = [ "macro_magic_core", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3475,7 +3590,7 @@ dependencies = [ "macro_magic", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3492,7 +3607,7 @@ dependencies = [ "lazy_static", "log", "mysql_common", - "num_enum", + "num_enum 0.7.3", "openssl", "percent-encoding", "serde", @@ -3515,11 +3630,11 @@ dependencies = [ "darling 0.20.11", "heck 0.5.0", "num-bigint", - "proc-macro-crate", + "proc-macro-crate 3.3.0", "proc-macro-error2", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", "termcolor", "thiserror 1.0.69", ] @@ -3671,13 +3786,34 @@ dependencies = [ "libm", ] +[[package]] +name = "num_enum" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f646caf906c20226733ed5b1374287eb97e3c2a5c227ce668c1f2ce20ae57c9" +dependencies = [ + "num_enum_derive 0.5.11", +] + [[package]] name = "num_enum" version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e613fc340b2220f734a8595782c551f1250e969d87d3be1ae0579e8d4065179" dependencies = [ - "num_enum_derive", + "num_enum_derive 0.7.3", +] + +[[package]] +name = "num_enum_derive" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dcbff9bc912032c62bf65ef1d5aea88983b420f4f839db1e9b0c281a25c9c799" +dependencies = [ + "proc-macro-crate 1.3.1", + "proc-macro2", + "quote", + "syn 1.0.109", ] [[package]] @@ -3686,10 +3822,10 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af1844ef2428cc3e1cb900be36181049ef3d3193c63e43026cfe202983b27a56" dependencies = [ - "proc-macro-crate", + "proc-macro-crate 3.3.0", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3773,7 +3909,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -3936,7 +4072,7 @@ checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -4089,13 +4225,23 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "proc-macro-crate" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f4c021e1093a56626774e81216a4ce732a735e5bad4868a03f3ed65ca0c3919" +dependencies = [ + "once_cell", + "toml_edit 0.19.15", +] + [[package]] name = "proc-macro-crate" version = "3.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edce586971a4dfaa28950c6f18ed55e0406c1ab88bbce2c6f6293a7aaba73d35" dependencies = [ - "toml_edit", + "toml_edit 0.22.27", ] [[package]] @@ -4117,7 +4263,7 @@ dependencies = [ "proc-macro-error-attr2", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -4170,7 +4316,7 @@ dependencies = [ "itertools", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -4434,7 +4580,7 @@ dependencies = [ "cmake", "libc", "libz-sys", - "num_enum", + "num_enum 0.7.3", "pkg-config", ] @@ -4575,7 +4721,7 @@ dependencies = [ "encoding_rs", "futures-core", "futures-util", - "h2 0.3.26", + "h2 0.3.27", "http 0.2.12", "http-body 0.4.6", "hyper 0.14.32", @@ -4808,6 +4954,7 @@ version = "0.23.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6a9586e9ee2b4f8fab52a0048ca7334d7024eef48e2cb9407e3497bb7cab7fa7" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring", @@ -4833,6 +4980,7 @@ version = "0.103.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -4900,7 +5048,7 @@ checksum = "22f968c5ea23d555e670b449c1c5e7b2fc399fdaec1d304a17cd48e288abc107" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -4978,7 +5126,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -4989,7 +5137,7 @@ checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5037,7 +5185,7 @@ dependencies = [ "darling 0.21.3", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5075,7 +5223,7 @@ checksum = "5d69265a08751de7844521fd15003ae0a888e035773ba05695c5c759a6f89eef" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5279,7 +5427,7 @@ dependencies = [ "futures-io", "futures-util", "hashbrown 0.15.4", - "hashlink", + "hashlink 0.10.0", "indexmap", "ipnetwork", "log", @@ -5310,7 +5458,7 @@ dependencies = [ "quote", "sqlx-core", "sqlx-macros-core", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5334,7 +5482,7 @@ dependencies = [ "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", - "syn 2.0.117", + "syn 2.0.103", "url", ] @@ -5485,13 +5633,35 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "strum" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cae14b91c7d11c9a851d3fbc80a963198998c2a64eec840477fa92d8ce9b70bb" +dependencies = [ + "strum_macros 0.23.1", +] + [[package]] name = "strum" version = "0.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "290d54ea6f91c969195bdbcd7442c8c2a2ba87da8bf60a7ee86a235d4bc1e125" dependencies = [ - "strum_macros", + "strum_macros 0.25.3", +] + +[[package]] +name = "strum_macros" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5bb0dc7ee9c15cea6199cde9a127fa16a4c5819af85395457ad72d68edc85a38" +dependencies = [ + "heck 0.3.3", + "proc-macro2", + "quote", + "rustversion", + "syn 1.0.109", ] [[package]] @@ -5504,7 +5674,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5536,9 +5706,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.117" +version = "2.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +checksum = "e4307e30089d6fd6aff212f2da3a1f9e32f3223b1f010fb09b7c95f90f3ca1e8" dependencies = [ "proc-macro2", "quote", @@ -5562,7 +5732,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5652,7 +5822,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5663,7 +5833,7 @@ checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5776,7 +5946,7 @@ checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -5877,6 +6047,17 @@ version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +[[package]] +name = "toml_edit" +version = "0.19.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b5bb770da30e5cbfde35a2d7b9b8a2c4b8ef89548a7a6aeab5c9a576e3e7421" +dependencies = [ + "indexmap", + "toml_datetime", + "winnow 0.5.40", +] + [[package]] name = "toml_edit" version = "0.22.27" @@ -5885,7 +6066,7 @@ checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" dependencies = [ "indexmap", "toml_datetime", - "winnow", + "winnow 0.7.11", ] [[package]] @@ -5997,7 +6178,7 @@ checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -6068,7 +6249,7 @@ checksum = "f03ca4cb38206e2bef0700092660bb74d696f808514dae47fa1467cbfe26e96e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -6079,7 +6260,7 @@ checksum = "0e48cea23f68d1f78eb7bc092881b6bb88d3d6b5b7e6234f6f9c911da1ffb221" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -6124,6 +6305,12 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e70f2a8b45122e719eb623c01822704c4e0907e7e426a05927e1a1cfff5b75d0" +[[package]] +name = "unicode-segmentation" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -6277,7 +6464,7 @@ dependencies = [ "log", "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", "wasm-bindgen-shared", ] @@ -6312,7 +6499,7 @@ checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -6457,7 +6644,7 @@ checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -6468,7 +6655,7 @@ checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -6731,6 +6918,15 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271414315aff87387382ec3d271b52d7ae78726f5d44ac98b4f4030c91880486" +[[package]] +name = "winnow" +version = "0.5.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f593a95398737aeed53e489c785df13f3618e41dbcd6718c6addbf1395aa6876" +dependencies = [ + "memchr", +] + [[package]] name = "winnow" version = "0.7.11" @@ -6800,7 +6996,7 @@ checksum = "38da3c9736e16c5d3c8c597a9aaa5d1fa565d0532ae05e27c24aa62fb32c0ab6" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", "synstructure", ] @@ -6821,7 +7017,7 @@ checksum = "9ecf5b4cc5364572d7f4c329661bcc82724222973f2cab6f050a4e5c22f75181" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", ] [[package]] @@ -6841,7 +7037,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", "synstructure", ] @@ -6881,7 +7077,32 @@ checksum = "5b96237efa0c878c64bd89c436f661be4e46b2f3eff1ebb976f7ef2321d2f58f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.117", + "syn 2.0.103", +] + +[[package]] +name = "zookeeper-client" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1dfa35623e95f8c27cc64ee268d338cb1a8ae8fedfdc7b2b7aa8deb83d2cde14" +dependencies = [ + "bytes", + "compact_str", + "const_format", + "derive-where", + "either", + "fastrand 2.3.0", + "hashbrown 0.12.3", + "hashlink 0.8.4", + "ignore-result", + "num_enum 0.5.11", + "static_assertions", + "strum 0.23.0", + "thiserror 1.0.69", + "tokio", + "tokio-rustls", + "tracing", + "uuid", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 82e250f4..3f6f0383 100644 --- a/Cargo.toml +++ b/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' diff --git a/dt-common/src/config/config_enums.rs b/dt-common/src/config/config_enums.rs index 9625a29e..2d45cdda 100644 --- a/dt-common/src/config/config_enums.rs +++ b/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)] diff --git a/dt-common/src/config/extractor_config.rs b/dt-common/src/config/extractor_config.rs index 6fd0fb0d..b85ad76f 100644 --- a/dt-common/src/config/extractor_config.rs +++ b/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, + scan_interval_secs: u64, + include_ephemeral: bool, + heartbeat_interval_secs: u64, + }, } #[derive(Clone, Debug, Hash)] diff --git a/dt-common/src/config/mod.rs b/dt-common/src/config/mod.rs index 899992f3..bc4ad398 100644 --- a/dt-common/src/config/mod.rs +++ b/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; diff --git a/dt-common/src/config/router_config.rs b/dt-common/src/config/router_config.rs index 021f5222..6533e34d 100644 --- a/dt-common/src/config/router_config.rs +++ b/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, + }, } diff --git a/dt-common/src/config/sinker_config.rs b/dt-common/src/config/sinker_config.rs index c0b49d7d..9e6c354e 100644 --- a/dt-common/src/config/sinker_config.rs +++ b/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)] diff --git a/dt-common/src/config/task_config.rs b/dt-common/src/config/task_config.rs index c6f9e817..ad923528 100644 --- a/dt-common/src/config/task_config.rs +++ b/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, pub data_marker: Option, pub processor: Option, + 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> { if !loader.ini.sections().contains(&DATA_MARKER.to_string()) { return Ok(None); diff --git a/dt-common/src/config/zk_filter_config.rs b/dt-common/src/config/zk_filter_config.rs new file mode 100644 index 00000000..382ad991 --- /dev/null +++ b/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, +} diff --git a/dt-common/src/lib.rs b/dt-common/src/lib.rs index 208dd78d..151624bd 100644 --- a/dt-common/src/lib.rs +++ b/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; diff --git a/dt-common/src/meta/dt_data.rs b/dt-common/src/meta/dt_data.rs index 83745752..b39ab50d 100644 --- a/dt-common/src/meta/dt_data.rs +++ b/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, } diff --git a/dt-common/src/meta/mod.rs b/dt-common/src/meta/mod.rs index 772f8d8a..bad49c48 100644 --- a/dt-common/src/meta/mod.rs +++ b/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; diff --git a/dt-common/src/meta/position.rs b/dt-common/src/meta/position.rs index 9b088c2c..032a0bc0 100644 --- a/dt-common/src/meta/position.rs +++ b/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, + 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, } } diff --git a/dt-common/src/meta/zk/mod.rs b/dt-common/src/meta/zk/mod.rs new file mode 100644 index 00000000..73d0a90b --- /dev/null +++ b/dt-common/src/meta/zk/mod.rs @@ -0,0 +1,3 @@ +pub mod zk_entry; +pub mod zk_event_type; +pub mod zk_stat; diff --git a/dt-common/src/meta/zk/zk_entry.rs b/dt-common/src/meta/zk/zk_entry.rs new file mode 100644 index 00000000..41228c9b --- /dev/null +++ b/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>, + 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 + } + } +} diff --git a/dt-common/src/meta/zk/zk_event_type.rs b/dt-common/src/meta/zk/zk_event_type.rs new file mode 100644 index 00000000..a7563dd9 --- /dev/null +++ b/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, +} diff --git a/dt-common/src/meta/zk/zk_stat.rs b/dt-common/src/meta/zk/zk_stat.rs new file mode 100644 index 00000000..b46c826d --- /dev/null +++ b/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, +} diff --git a/dt-common/src/zk_filter.rs b/dt-common/src/zk_filter.rs new file mode 100644 index 00000000..a3072c6e --- /dev/null +++ b/dt-common/src/zk_filter.rs @@ -0,0 +1,97 @@ +use std::collections::HashSet; + +use crate::config::zk_filter_config::ZkFilterConfig; + +#[derive(Debug, Clone)] +pub struct ZkFilter { + pub do_paths: HashSet, + pub ignore_paths: HashSet, + pub include_ephemeral: bool, +} + +impl ZkFilter { + pub const INTERNAL_PATHS: [&'static str; 3] = [ + "/__ape_dts_marker", + "/__ape_dts_shadow", + "/__ape_dts_heartbeat", + ]; + + pub fn from_config(config: &ZkFilterConfig) -> anyhow::Result { + let do_paths = Self::parse_paths(&config.do_paths); + let ignore_paths = Self::parse_paths(&config.ignore_paths); + Ok(Self { + do_paths, + ignore_paths, + include_ephemeral: config.include_ephemeral, + }) + } + + pub fn filter_path(&self, path: &str) -> bool { + if Self::INTERNAL_PATHS + .iter() + .any(|p| Self::path_matches_prefix(p, path)) + { + return true; + } + if self + .ignore_paths + .iter() + .any(|p| Self::path_matches_prefix(p, path)) + { + return true; + } + if self.do_paths.is_empty() { + return false; + } + !self + .do_paths + .iter() + .any(|p| Self::path_matches_prefix(p, path)) + } + + pub fn filter_ephemeral(&self, ephemeral: bool) -> bool { + ephemeral && !self.include_ephemeral + } + + fn parse_paths(config_str: &str) -> HashSet { + if config_str.trim().is_empty() { + return HashSet::new(); + } + config_str + .split(',') + .map(|s| s.trim().to_string()) + .collect() + } + + fn path_matches_prefix(prefix: &str, path: &str) -> bool { + prefix == "/" + || path == prefix + || path.starts_with(&format!("{}/", prefix.trim_end_matches('/'))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::zk_filter_config::ZkFilterConfig; + + fn filter(do_paths: &str, ignore_paths: &str) -> ZkFilter { + ZkFilter::from_config(&ZkFilterConfig { + do_paths: do_paths.to_string(), + ignore_paths: ignore_paths.to_string(), + include_ephemeral: false, + }) + .unwrap() + } + + #[test] + fn internal_paths_are_always_filtered() { + let filter = filter("/", ""); + + assert!(filter.filter_path("/__ape_dts_marker")); + assert!(filter.filter_path("/__ape_dts_shadow/app")); + assert!(filter.filter_path("/__ape_dts_heartbeat")); + assert!(!filter.filter_path("/__ape_dts_shadow2")); + assert!(!filter.filter_path("/app/service")); + } +} diff --git a/dt-connector/Cargo.toml b/dt-connector/Cargo.toml index 3e011ced..cfffd8ca 100644 --- a/dt-connector/Cargo.toml +++ b/dt-connector/Cargo.toml @@ -49,3 +49,4 @@ opendal = { workspace = true } dashmap = {workspace = true} indexmap = {workspace = true} openssl = { workspace = true } +zookeeper-client = { workspace = true } diff --git a/dt-connector/src/data_marker.rs b/dt-connector/src/data_marker.rs index 2cd78b18..2443e080 100644 --- a/dt-connector/src/data_marker.rs +++ b/dt-connector/src/data_marker.rs @@ -2,7 +2,7 @@ use std::collections::HashSet; use dt_common::{ config::{config_enums::DbType, data_marker_config::DataMarkerConfig}, - meta::{dt_data::DtData, redis::redis_entry::RedisEntry}, + meta::{dt_data::DtData, redis::redis_entry::RedisEntry, zk::zk_entry::ZkEntry}, }; #[derive(Debug, Clone, Default)] @@ -76,6 +76,7 @@ impl DataMarker { match dt_data { DtData::Dml { row_data } => self.is_rdb_marker_info(&row_data.schema, &row_data.tb), DtData::Redis { entry } => self.is_redis_marker_info(entry), + DtData::Zk { entry } => self.is_zk_marker_info(entry), _ => false, } } @@ -95,6 +96,10 @@ impl DataMarker { self.marker_schema == schema && self.marker_tb == tb } + pub fn is_zk_marker_info(&self, entry: &ZkEntry) -> bool { + entry.path == self.marker || entry.path.starts_with(&format!("{}/", self.marker)) + } + pub fn refresh(&mut self, dt_data: &DtData) { match dt_data { DtData::Dml { row_data } => { @@ -113,6 +118,18 @@ impl DataMarker { self.data_origin_node = entry.cmd.get_str_arg(2); } + DtData::Zk { entry } => { + if let Some(data) = &entry.data { + if let Ok(marker_value) = serde_json::from_slice::(data) { + if let Some(source_id) = + marker_value.get("source_id").and_then(|v| v.as_str()) + { + self.data_origin_node = source_id.to_string(); + } + } + } + } + _ => {} } diff --git a/dt-connector/src/extractor/mod.rs b/dt-connector/src/extractor/mod.rs index 180aa021..60e6e45b 100644 --- a/dt-connector/src/extractor/mod.rs +++ b/dt-connector/src/extractor/mod.rs @@ -12,6 +12,7 @@ pub mod resumer; pub mod snapshot_chunk_id_generator; pub mod snapshot_dispatcher; pub mod snapshot_types; +pub mod zk; fn estimated_sample_limit(sample_rate: Option, estimated_count: u64) -> Option { let sample_rate = sample_rate.filter(|rate| (1..100).contains(rate))?; diff --git a/dt-connector/src/extractor/zk/mod.rs b/dt-connector/src/extractor/zk/mod.rs new file mode 100644 index 00000000..88ed885d --- /dev/null +++ b/dt-connector/src/extractor/zk/mod.rs @@ -0,0 +1 @@ +pub mod zk_extractor; diff --git a/dt-connector/src/extractor/zk/zk_extractor.rs b/dt-connector/src/extractor/zk/zk_extractor.rs new file mode 100644 index 00000000..dd9789e5 --- /dev/null +++ b/dt-connector/src/extractor/zk/zk_extractor.rs @@ -0,0 +1,710 @@ +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + time::Duration, +}; + +use anyhow::bail; +use async_trait::async_trait; +use tokio::sync::{mpsc, Mutex}; +use zookeeper_client::{ + Acls, AddWatchMode, Client as ZkClient, CreateMode, Error as ZkError, EventType, SessionState, + Stat, WatchedEvent, +}; + +use dt_common::{ + log_error, log_info, log_warn, + meta::{ + dt_data::DtData, + position::Position, + syncer::Syncer, + zk::{ + zk_entry::{ZkEntry, ZkOrderOrigin}, + zk_event_type::ZkEventType, + zk_stat::ZkStat, + }, + }, + zk_filter::ZkFilter, +}; + +use crate::extractor::base_extractor::{BaseExtractor, ExtractState}; +use crate::extractor::resumer::recovery::Recovery; +use crate::Extractor; + +pub struct ZkExtractor { + pub url: String, + pub watch_paths: Vec, + pub scan_interval_secs: u64, + pub include_ephemeral: bool, + pub heartbeat_interval_secs: u64, + pub filter: ZkFilter, + pub base_extractor: BaseExtractor, + pub extract_state: ExtractState, + pub syncer: Arc>, + pub recovery: Option>, +} + +#[derive(Debug, Clone)] +struct ZkWatchEvent { + watch_path: String, + event: WatchedEvent, +} + +#[derive(Debug, Clone)] +enum ReconcileScope { + All, + Path(String), +} + +#[derive(Debug, Clone)] +struct ZkScanResult { + versions: HashMap, + high_water_zxid: i64, + entries: Vec, +} + +impl ZkExtractor { + fn stat_to_zk_stat(stat: &Stat) -> ZkStat { + ZkStat { + version: stat.version, + cversion: stat.cversion, + aversion: stat.aversion, + mtime: stat.mtime, + ctime: stat.ctime, + czxid: stat.czxid, + mzxid: stat.mzxid, + ephemeral_owner: stat.ephemeral_owner, + } + } + + fn is_ephemeral(stat: &Stat) -> bool { + stat.ephemeral_owner != 0 + } + + async fn recover_position(&mut self) -> anyhow::Result<(HashMap, i64)> { + if let Some(recovery) = &self.recovery { + if let Some(position) = recovery.get_cdc_resume_position().await { + if let Position::Zk { + path_versions, + high_water_zxid, + last_scan_timestamp, + total_paths, + } = &position + { + log_info!( + "ZkExtractor recovery: {} paths, high_water_zxid={}, last_scan={}", + total_paths, + high_water_zxid, + last_scan_timestamp + ); + self.base_extractor + .push_dt_data( + &mut self.extract_state, + DtData::Heartbeat {}, + position.clone(), + ) + .await?; + return Ok((path_versions.clone(), *high_water_zxid)); + } + } + } + Ok((HashMap::new(), 0)) + } + + async fn reconcile( + &mut self, + client: &ZkClient, + path_versions: &mut HashMap, + high_water_zxid: &mut i64, + source_id: &str, + scope: ReconcileScope, + reason: &str, + ) -> anyhow::Result<()> { + let old_versions = path_versions.clone(); + let mut scan_result = ZkScanResult { + versions: match scope { + ReconcileScope::All => HashMap::new(), + ReconcileScope::Path(_) => old_versions.clone(), + }, + high_water_zxid: *high_water_zxid, + entries: Vec::new(), + }; + let roots = self.reconcile_roots(&scope); + + if matches!(scope, ReconcileScope::Path(_)) { + scan_result + .versions + .retain(|path, _| !Self::scope_contains_path(&scope, path)); + } + + log_info!( + "ZkExtractor reconciliation start: reason={}, scope={:?}, roots={:?}", + reason, + scope, + roots + ); + + for root in roots { + if self.base_extractor.shut_down.load(Ordering::Relaxed) { + break; + } + self.scan_node_recursive(client, &root, &old_versions, &mut scan_result, source_id) + .await?; + } + + for (path, (old_version, old_mzxid)) in &old_versions { + if !Self::scope_contains_path(&scope, path) { + continue; + } + if scan_result.versions.contains_key(path) { + continue; + } + if self.filter.filter_path(path) { + continue; + } + let delete_order_millis = chrono::Utc::now().timestamp_millis(); + let delete_zxid = if *old_mzxid > 0 { + *old_mzxid + } else { + delete_order_millis + }; + let entry = ZkEntry { + path: path.clone(), + data: None, + stat: ZkStat { + version: *old_version, + mzxid: delete_zxid, + mtime: delete_order_millis, + ..Default::default() + }, + ephemeral: false, + event_type: ZkEventType::Deleted, + source_id: source_id.to_string(), + source_order_millis: delete_order_millis, + source_zxid: delete_zxid, + order_origin: ZkOrderOrigin::ReconcileObserved, + }; + scan_result.entries.push(entry); + } + + let final_position = + self.build_position(&scan_result.versions, scan_result.high_water_zxid); + let emitted_entries = scan_result.entries.len(); + for entry in scan_result.entries { + self.extract_state + .push_dt_data( + &self.base_extractor, + DtData::Zk { entry }, + Self::staged_entry_position(), + ) + .await?; + } + self.extract_state + .push_dt_data(&self.base_extractor, DtData::Heartbeat {}, final_position) + .await?; + + *path_versions = scan_result.versions; + *high_water_zxid = scan_result.high_water_zxid; + + log_info!( + "ZkExtractor reconciliation complete: reason={}, {} paths, high_water_zxid={}, emitted_entries={}", + reason, + path_versions.len(), + high_water_zxid, + emitted_entries + ); + Ok(()) + } + + fn reconcile_roots(&self, scope: &ReconcileScope) -> Vec { + match scope { + ReconcileScope::All => self.watch_paths.clone(), + ReconcileScope::Path(path) => { + if self.watch_paths.iter().any(|watch_path| { + Self::path_contains(watch_path, path) || Self::path_contains(path, watch_path) + }) { + vec![path.clone()] + } else { + Vec::new() + } + } + } + } + + fn scope_contains_path(scope: &ReconcileScope, path: &str) -> bool { + match scope { + ReconcileScope::All => true, + ReconcileScope::Path(scope_path) => Self::path_contains(scope_path, path), + } + } + + fn path_contains(root: &str, path: &str) -> bool { + root == "/" || path == root || path.starts_with(&format!("{}/", root.trim_end_matches('/'))) + } + + fn staged_entry_position() -> Position { + Position::None + } + + async fn full_scan( + &mut self, + client: &ZkClient, + path_versions: &mut HashMap, + high_water_zxid: &mut i64, + source_id: &str, + ) -> anyhow::Result<()> { + self.reconcile( + client, + path_versions, + high_water_zxid, + source_id, + ReconcileScope::All, + "initial", + ) + .await + } + + async fn scan_node_recursive( + &mut self, + client: &ZkClient, + path: &str, + old_versions: &HashMap, + scan_result: &mut ZkScanResult, + source_id: &str, + ) -> anyhow::Result<()> { + if self.base_extractor.shut_down.load(Ordering::Relaxed) { + return Ok(()); + } + + if self.filter.filter_path(path) { + return Ok(()); + } + + let (data, stat) = match client.get_data(path).await { + Ok(r) => r, + Err(ZkError::NoNode) => { + return Ok(()); + } + Err(e) => { + bail!("ZkExtractor: get_data failed for {}: {}", path, e); + } + }; + + let ephemeral = Self::is_ephemeral(&stat); + if self.filter.filter_ephemeral(ephemeral) { + return Ok(()); + } + + let children = match client.list_children(path).await { + Ok(c) => c, + Err(ZkError::NoNode) => { + return Ok(()); + } + Err(e) => { + bail!("ZkExtractor: list_children failed for {}: {}", path, e); + } + }; + + let (already_synced, is_update) = match old_versions.get(path) { + Some((v, old_mzxid)) => (*v >= stat.version && *old_mzxid >= stat.mzxid, true), + None => (false, false), + }; + + if !already_synced { + let event_type = if is_update { + ZkEventType::Updated + } else { + ZkEventType::Created + }; + let entry = ZkEntry { + path: path.to_string(), + data: Some(data), + stat: Self::stat_to_zk_stat(&stat), + ephemeral, + event_type, + source_id: source_id.to_string(), + source_order_millis: stat.mtime, + source_zxid: stat.mzxid, + order_origin: ZkOrderOrigin::SourceStatMtime, + }; + scan_result.entries.push(entry); + } + + scan_result + .versions + .insert(path.to_string(), (stat.version, stat.mzxid)); + if stat.mzxid > scan_result.high_water_zxid { + scan_result.high_water_zxid = stat.mzxid; + } + + for child in children { + let child_path = if path == "/" { + format!("/{}", child) + } else { + format!("{}/{}", path, child) + }; + Box::pin(self.scan_node_recursive( + client, + &child_path, + old_versions, + scan_result, + source_id, + )) + .await?; + } + + Ok(()) + } + + async fn incremental_rescan( + &mut self, + client: &ZkClient, + path_versions: &mut HashMap, + high_water_zxid: &mut i64, + source_id: &str, + ) -> anyhow::Result<()> { + self.reconcile( + client, + path_versions, + high_water_zxid, + source_id, + ReconcileScope::All, + "periodic", + ) + .await + } + + fn build_position( + &self, + path_versions: &HashMap, + high_water_zxid: i64, + ) -> Position { + Position::Zk { + path_versions: path_versions.clone(), + last_scan_timestamp: chrono::Utc::now().timestamp_millis(), + high_water_zxid, + total_paths: path_versions.len(), + } + } + + fn start_heartbeat( + &self, + shut_down: Arc, + syncer: Arc>, + url: String, + heartbeat_interval_secs: u64, + source_id: String, + ) { + if heartbeat_interval_secs == 0 { + return; + } + tokio::spawn(async move { + let hb_client = match ZkClient::connect(&url).await { + Ok(c) => c, + Err(e) => { + log_error!("ZkExtractor heartbeat: connect failed: {}", e); + return; + } + }; + + let marker_path = "/__ape_dts_heartbeat"; + let _ = hb_client + .create( + marker_path, + b"", + &CreateMode::Persistent.with_acls(Acls::anyone_all()), + ) + .await; + + loop { + if shut_down.load(Ordering::Relaxed) { + break; + } + tokio::time::sleep(tokio::time::Duration::from_secs(heartbeat_interval_secs)).await; + if shut_down.load(Ordering::Relaxed) { + break; + } + + let syncer_guard = syncer.lock().await; + let hb_data = serde_json::json!({ + "source_id": source_id, + "received_position": syncer_guard.received_position.to_string(), + "committed_position": syncer_guard.committed_position.to_string(), + "timestamp": chrono::Utc::now().to_rfc3339(), + }); + drop(syncer_guard); + + if let Err(e) = hb_client + .set_data(marker_path, hb_data.to_string().as_bytes(), None) + .await + { + log_warn!("ZkExtractor heartbeat write failed: {}", e); + } + } + }); + } + + async fn start_watchers( + &self, + client: &ZkClient, + ) -> anyhow::Result> { + if self.watch_paths.is_empty() { + bail!("ZkExtractor requires at least one watch_path"); + } + + let (tx, rx) = mpsc::unbounded_channel(); + for watch_path in self.watch_paths.clone() { + let mut watcher = client + .watch(&watch_path, AddWatchMode::PersistentRecursive) + .await + .map_err(|e| anyhow::anyhow!("ZK register watcher {} failed: {}", watch_path, e))?; + let tx = tx.clone(); + let shut_down = self.base_extractor.shut_down.clone(); + + log_info!( + "ZkExtractor registered persistent recursive watcher on {}", + watch_path + ); + tokio::spawn(async move { + loop { + if shut_down.load(Ordering::Relaxed) { + break; + } + + let event = watcher.changed().await; + let terminal = event.event_type == EventType::Session + && event.session_state.is_terminated(); + if tx + .send(ZkWatchEvent { + watch_path: watch_path.clone(), + event, + }) + .is_err() + { + break; + } + if terminal { + break; + } + } + }); + } + drop(tx); + Ok(rx) + } + + async fn handle_watch_event( + &mut self, + client: &ZkClient, + path_versions: &mut HashMap, + high_water_zxid: &mut i64, + source_id: &str, + watch_event: ZkWatchEvent, + session_uncertain: &mut bool, + ) -> anyhow::Result<()> { + let event = watch_event.event; + log_info!( + "ZkExtractor watcher event: watch_path={}, event_type={}, state={}, path={}, zxid={}", + watch_event.watch_path, + event.event_type, + event.session_state, + event.path, + event.zxid + ); + + match event.event_type { + EventType::Session => match event.session_state { + SessionState::Disconnected => { + *session_uncertain = true; + log_warn!( + "ZkExtractor watcher disconnected; next connected event will force reconciliation" + ); + Ok(()) + } + SessionState::SyncConnected | SessionState::ConnectedReadOnly => { + if *session_uncertain { + *session_uncertain = false; + self.reconcile( + client, + path_versions, + high_water_zxid, + source_id, + ReconcileScope::All, + "watch-reconnect", + ) + .await?; + } + Ok(()) + } + SessionState::AuthFailed | SessionState::Expired | SessionState::Closed => { + bail!( + "ZkExtractor watcher terminal session state: {}", + event.session_state + ) + } + }, + EventType::NodeCreated + | EventType::NodeDeleted + | EventType::NodeDataChanged + | EventType::NodeChildrenChanged => { + if event.path.is_empty() { + return Ok(()); + } + if self.filter.filter_path(&event.path) { + return Ok(()); + } + if *session_uncertain { + log_warn!( + "ZkExtractor treats watcher event {} as hint while session is uncertain", + event.path + ); + return Ok(()); + } + self.reconcile( + client, + path_versions, + high_water_zxid, + source_id, + ReconcileScope::Path(event.path.clone()), + "watch-event", + ) + .await + } + } + } +} + +#[async_trait] +impl Extractor for ZkExtractor { + async fn extract(&mut self) -> anyhow::Result<()> { + let source_id = self + .extract_state + .data_marker + .as_ref() + .map(|dm| dm.src_node.clone()) + .unwrap_or_else(|| format!("ape_dts_{}", self.url)); + + let (mut path_versions, mut high_water_zxid) = self.recover_position().await?; + + let client = ZkClient::connect(&self.url).await?; + log_info!("ZkExtractor connected to {}", self.url); + + self.start_heartbeat( + self.base_extractor.shut_down.clone(), + self.syncer.clone(), + self.url.clone(), + self.heartbeat_interval_secs, + source_id.clone(), + ); + + let mut watch_rx = self.start_watchers(&client).await?; + + self.full_scan( + &client, + &mut path_versions, + &mut high_water_zxid, + &source_id, + ) + .await?; + + let mut reconciliation_timer = + tokio::time::interval(Duration::from_secs(self.scan_interval_secs.max(1))); + reconciliation_timer.tick().await; + let mut session_uncertain = false; + let mut watch_rx_closed = false; + + while !self.base_extractor.shut_down.load(Ordering::Relaxed) { + tokio::select! { + maybe_event = watch_rx.recv(), if !watch_rx_closed => { + if let Some(watch_event) = maybe_event { + self.handle_watch_event( + &client, + &mut path_versions, + &mut high_water_zxid, + &source_id, + watch_event, + &mut session_uncertain, + ).await?; + } else { + log_warn!("ZkExtractor watcher channel closed; forcing periodic reconciliation only"); + watch_rx_closed = true; + } + } + _ = reconciliation_timer.tick() => { + if self.base_extractor.shut_down.load(Ordering::Relaxed) { + break; + } + if session_uncertain { + log_warn!( + "ZkExtractor skips periodic reconciliation while session is uncertain" + ); + continue; + } + self.incremental_rescan( + &client, + &mut path_versions, + &mut high_water_zxid, + &source_id, + ) + .await?; + } + } + } + + self.base_extractor + .wait_task_finish(&mut self.extract_state) + .await + } + + async fn close(&mut self) -> anyhow::Result<()> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn path_contains_matches_exact_and_descendants() { + assert!(ZkExtractor::path_contains("/", "/app")); + assert!(ZkExtractor::path_contains("/app", "/app")); + assert!(ZkExtractor::path_contains("/app", "/app/service")); + assert!(!ZkExtractor::path_contains("/app", "/application")); + } + + #[test] + fn scoped_reconcile_only_matches_path_subtree() { + let scope = ReconcileScope::Path("/app".to_string()); + + assert!(ZkExtractor::scope_contains_path(&scope, "/app")); + assert!(ZkExtractor::scope_contains_path(&scope, "/app/service")); + assert!(!ZkExtractor::scope_contains_path(&scope, "/config")); + } + + #[test] + fn aborted_scan_result_does_not_replace_old_versions_or_emit_deletes() { + let old_versions = HashMap::from([("/app/service".to_string(), (1, 10))]); + let scan_result = ZkScanResult { + versions: HashMap::new(), + high_water_zxid: 0, + entries: Vec::new(), + }; + let scan_error: anyhow::Result<()> = Err(anyhow::anyhow!( + "ZkExtractor: list_children failed for /app" + )); + + assert!(scan_error.is_err()); + assert_eq!(old_versions.get("/app/service"), Some(&(1, 10))); + assert!(scan_result.versions.is_empty()); + assert!(scan_result.entries.is_empty()); + } + + #[test] + fn staged_reconciliation_entries_do_not_carry_final_position() { + assert_eq!(ZkExtractor::staged_entry_position(), Position::None); + } +} diff --git a/dt-connector/src/lib.rs b/dt-connector/src/lib.rs index 3e959aaa..7f5d6dfd 100644 --- a/dt-connector/src/lib.rs +++ b/dt-connector/src/lib.rs @@ -10,6 +10,7 @@ pub mod meta_fetcher; pub mod rdb_query_builder; pub mod rdb_router; pub mod sinker; +pub mod zk_router; use async_trait::async_trait; use checker::check_log::CheckLog; diff --git a/dt-connector/src/rdb_router.rs b/dt-connector/src/rdb_router.rs index 96738a35..c15e3a27 100644 --- a/dt-connector/src/rdb_router.rs +++ b/dt-connector/src/rdb_router.rs @@ -187,6 +187,7 @@ impl RdbRouterInner { col_map, }) } + _ => Ok(Self::default()), } } @@ -469,6 +470,7 @@ impl RdbTopicRouterInner { RouterConfig::Rdb { topic_map, .. } => Ok(Self { topic_map: Self::parse_topic_map(topic_map, db_type)?, }), + _ => Ok(Self::default()), } } diff --git a/dt-connector/src/sinker/base_struct_sinker.rs b/dt-connector/src/sinker/base_struct_sinker.rs index ac1ae0c7..8c60f0c2 100644 --- a/dt-connector/src/sinker/base_struct_sinker.rs +++ b/dt-connector/src/sinker/base_struct_sinker.rs @@ -44,7 +44,9 @@ impl BaseStructSinker { Err(error) => { log_error!("ddl failed, error: {}", error); match conflict_policy { - ConflictPolicyEnum::Interrupt => bail! {error}, + ConflictPolicyEnum::Interrupt | ConflictPolicyEnum::LastWriteWins => { + bail! {error} + } ConflictPolicyEnum::Ignore => {} } } diff --git a/dt-connector/src/sinker/clickhouse/clickhouse_struct_sinker.rs b/dt-connector/src/sinker/clickhouse/clickhouse_struct_sinker.rs index 19473467..12227d03 100644 --- a/dt-connector/src/sinker/clickhouse/clickhouse_struct_sinker.rs +++ b/dt-connector/src/sinker/clickhouse/clickhouse_struct_sinker.rs @@ -287,7 +287,9 @@ impl ClickhouseStructSinker { Err(error) => { log_error!("ddl failed, error: {}", error); match self.conflict_policy { - ConflictPolicyEnum::Interrupt => bail! {error}, + ConflictPolicyEnum::Interrupt | ConflictPolicyEnum::LastWriteWins => { + bail! {error} + } ConflictPolicyEnum::Ignore => {} } } diff --git a/dt-connector/src/sinker/mod.rs b/dt-connector/src/sinker/mod.rs index db523070..4b405ddd 100644 --- a/dt-connector/src/sinker/mod.rs +++ b/dt-connector/src/sinker/mod.rs @@ -10,3 +10,4 @@ pub mod pg; pub mod redis; pub mod sql_sinker; pub mod starrocks; +pub mod zk; diff --git a/dt-connector/src/sinker/mongo/mongo_struct_sinker.rs b/dt-connector/src/sinker/mongo/mongo_struct_sinker.rs index 2d80d863..4caccc82 100644 --- a/dt-connector/src/sinker/mongo/mongo_struct_sinker.rs +++ b/dt-connector/src/sinker/mongo/mongo_struct_sinker.rs @@ -53,7 +53,9 @@ impl Sinker for MongoStructSinker { Err(error) => { log_error!("mongo struct failed, error: {}", error); match self.conflict_policy { - ConflictPolicyEnum::Interrupt => return Err(error), + ConflictPolicyEnum::Interrupt | ConflictPolicyEnum::LastWriteWins => { + return Err(error) + } ConflictPolicyEnum::Ignore => {} } } diff --git a/dt-connector/src/sinker/starrocks/starrocks_struct_sinker.rs b/dt-connector/src/sinker/starrocks/starrocks_struct_sinker.rs index 3347a245..777cfc3e 100644 --- a/dt-connector/src/sinker/starrocks/starrocks_struct_sinker.rs +++ b/dt-connector/src/sinker/starrocks/starrocks_struct_sinker.rs @@ -386,7 +386,9 @@ impl StarrocksStructSinker { Err(error) => { log_error!("ddl failed, error: {}", error); match self.conflict_policy { - ConflictPolicyEnum::Interrupt => bail! {error}, + ConflictPolicyEnum::Interrupt | ConflictPolicyEnum::LastWriteWins => { + bail! {error} + } ConflictPolicyEnum::Ignore => {} } } diff --git a/dt-connector/src/sinker/zk/mod.rs b/dt-connector/src/sinker/zk/mod.rs new file mode 100644 index 00000000..b46861cc --- /dev/null +++ b/dt-connector/src/sinker/zk/mod.rs @@ -0,0 +1 @@ +pub mod zk_sinker; diff --git a/dt-connector/src/sinker/zk/zk_sinker.rs b/dt-connector/src/sinker/zk/zk_sinker.rs new file mode 100644 index 00000000..213c10da --- /dev/null +++ b/dt-connector/src/sinker/zk/zk_sinker.rs @@ -0,0 +1,595 @@ +use std::sync::Arc; + +use anyhow::bail; +use async_trait::async_trait; +use tokio::sync::RwLock; +use zookeeper_client as zk; + +use dt_common::config::config_enums::ConflictPolicyEnum; +use dt_common::log_info; +use dt_common::log_warn; +use dt_common::meta::dt_data::{DtData, DtItem}; +use dt_common::meta::zk::zk_entry::{ZkEntry, ZkOrderOrigin}; +use dt_common::meta::zk::zk_event_type::ZkEventType; + +use crate::data_marker::DataMarker; +use crate::sinker::base_sinker::BaseSinker; +use crate::zk_router::ZkRouter; +use crate::Sinker; + +pub struct ZkSinker { + pub url: String, + pub batch_size: usize, + pub create_if_not_exists: bool, + pub sync_ephemeral_as_persistent: bool, + pub conflict_policy: ConflictPolicyEnum, + pub router: ZkRouter, + pub base_sinker: BaseSinker, + pub data_marker: Option>>, + pub client: Option, +} + +#[async_trait] +impl Sinker for ZkSinker { + async fn sink_raw(&mut self, data: Vec, _batch: bool) -> anyhow::Result<()> { + if data.is_empty() { + return Ok(()); + } + + let client = self.get_or_connect().await?; + self.serial_sink_raw(&client, data).await + } + + async fn close(&mut self) -> anyhow::Result<()> { + self.client = None; + Ok(()) + } +} + +impl ZkSinker { + async fn get_or_connect(&mut self) -> anyhow::Result { + if let Some(client) = &self.client { + return Ok(client.clone()); + } + let client = zk::Client::connect(&self.url) + .await + .map_err(|e| anyhow::anyhow!("ZK connect to {} failed: {}", self.url, e))?; + log_info!("ZkSinker connected to {}", self.url); + self.client = Some(client.clone()); + Ok(client) + } + + async fn serial_sink_raw( + &mut self, + client: &zk::Client, + data: Vec, + ) -> anyhow::Result<()> { + let mut data_size = 0u64; + let mut record_count = 0u64; + + for dt_item in data.iter() { + if let DtData::Zk { entry } = &dt_item.dt_data { + data_size += entry.get_data_size(); + let routed_path = self.router.route_path(&entry.path); + + if self.conflict_policy == ConflictPolicyEnum::LastWriteWins + && self.should_skip_by_lww(client, &routed_path, entry).await? + { + continue; + } + + self.apply_zk_event(client, &routed_path, entry).await?; + record_count += 1; + } + } + + self.write_data_marker(client).await?; + + self.base_sinker + .update_serial_monitor(record_count, data_size) + .await + } + + async fn should_skip_by_lww( + &self, + client: &zk::Client, + path: &str, + entry: &ZkEntry, + ) -> anyhow::Result { + let shadow = shadow_path(path); + if let Ok((data, _)) = client.get_data(&shadow).await { + match serde_json::from_slice::(&data) { + Ok(shadow_meta) => { + return Ok(Self::should_skip_by_shadow_meta(&shadow_meta, entry)); + } + Err(e) => { + log_warn!( + "ZK shadow {} has invalid metadata, ignoring for LWW: {}", + shadow, + e + ); + } + } + } + Ok(false) + } + + fn should_skip_by_shadow_meta(shadow: &ZkShadowMeta, entry: &ZkEntry) -> bool { + shadow.order_key() > Self::entry_order_key(entry) + } + + fn entry_order_key(entry: &ZkEntry) -> (i64, i64, &str) { + ( + entry.source_order_millis(), + entry.source_zxid, + entry.source_id.as_str(), + ) + } + + async fn apply_zk_event( + &self, + client: &zk::Client, + path: &str, + entry: &ZkEntry, + ) -> anyhow::Result<()> { + let data = entry.data.as_deref().unwrap_or(&[]); + + if matches!( + entry.event_type, + ZkEventType::Created | ZkEventType::Updated + ) { + if let Ok((existing_data, _)) = client.get_data(path).await { + if existing_data == data { + self.write_shadow(client, path, entry).await?; + return Ok(()); + } + } + } + + match entry.event_type { + ZkEventType::Created => { + let options = self.create_options(entry); + + if self.create_if_not_exists { + self.ensure_parent_paths(client, path).await?; + } + + match client.create(path, data, &options).await { + Ok(_) => {} + Err(ref e) if is_node_exists(e) => { + client.set_data(path, data, None).await.map_err(|e| { + anyhow::anyhow!("ZK set_data (upsert) {} failed: {}", path, e) + })?; + } + Err(e) => bail!("ZK create {} failed: {}", path, e), + } + self.write_shadow(client, path, entry).await?; + Ok(()) + } + + ZkEventType::Updated => { + match client.set_data(path, data, None).await { + Ok(_) => {} + Err(ref e) if is_no_node(e) && self.create_if_not_exists => { + self.ensure_parent_paths(client, path).await?; + let options = self.create_options(entry); + client.create(path, data, &options).await.map_err(|e| { + anyhow::anyhow!("ZK create (auto) {} failed: {}", path, e) + })?; + } + Err(e) => bail!("ZK set_data {} failed: {}", path, e), + } + self.write_shadow(client, path, entry).await?; + Ok(()) + } + + ZkEventType::Deleted => { + match client.delete(path, None).await { + Ok(()) => {} + Err(ref e) if is_no_node(e) => {} + Err(ref e) if is_not_empty(e) => { + log_warn!( + "ZK delete {} skipped: node has children (may have new data from peer)", + path + ); + return Ok(()); + } + Err(e) => bail!("ZK delete {} failed: {}", path, e), + } + self.write_tombstone_shadow(client, path, entry).await?; + Ok(()) + } + + ZkEventType::ChildrenChanged => Ok(()), + } + } + + fn create_options(&self, entry: &ZkEntry) -> zk::CreateOptions<'static> { + let mode = if entry.ephemeral && !self.sync_ephemeral_as_persistent { + zk::CreateMode::Ephemeral + } else { + zk::CreateMode::Persistent + }; + mode.with_acls(zk::Acls::anyone_all()) + } + + fn persistent_options(&self) -> zk::CreateOptions<'static> { + zk::CreateMode::Persistent.with_acls(zk::Acls::anyone_all()) + } + + async fn ensure_parent_paths(&self, client: &zk::Client, path: &str) -> anyhow::Result<()> { + let parts: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect(); + if parts.len() <= 1 { + return Ok(()); + } + let options = self.persistent_options(); + let mut current = String::new(); + for part in &parts[..parts.len() - 1] { + current.push('/'); + current.push_str(part); + match client.create(¤t, &[], &options).await { + Ok(_) => {} + Err(ref e) if is_node_exists(e) => {} + Err(e) => bail!("ZK create parent {} failed: {}", current, e), + } + } + Ok(()) + } + + async fn write_shadow( + &self, + client: &zk::Client, + path: &str, + entry: &ZkEntry, + ) -> anyhow::Result<()> { + let shadow = shadow_path(path); + let shadow_data = Self::shadow_data(entry, false)?; + match client.set_data(&shadow, &shadow_data, None).await { + Ok(_) => {} + Err(ref e) if is_no_node(e) => { + self.ensure_parent_paths(client, &shadow).await?; + let options = self.persistent_options(); + match client.create(&shadow, &shadow_data, &options).await { + Ok(_) => {} + Err(ref e) if is_node_exists(e) => { + client.set_data(&shadow, &shadow_data, None).await.ok(); + } + Err(e) => bail!("ZK create shadow {} failed: {}", shadow, e), + } + } + Err(e) => bail!("ZK write shadow {} failed: {}", shadow, e), + } + Ok(()) + } + + async fn write_tombstone_shadow( + &self, + client: &zk::Client, + path: &str, + entry: &ZkEntry, + ) -> anyhow::Result<()> { + let shadow = shadow_path(path); + let shadow_data = Self::shadow_data(entry, true)?; + match client.set_data(&shadow, &shadow_data, None).await { + Ok(_) => {} + Err(ref e) if is_no_node(e) => { + self.ensure_parent_paths(client, &shadow).await?; + let options = self.persistent_options(); + match client.create(&shadow, &shadow_data, &options).await { + Ok(_) => {} + Err(ref e) if is_node_exists(e) => { + client.set_data(&shadow, &shadow_data, None).await.ok(); + } + Err(e) => bail!("ZK create tombstone shadow {} failed: {}", shadow, e), + } + } + Err(e) => bail!("ZK write tombstone shadow {} failed: {}", shadow, e), + } + Ok(()) + } + + fn shadow_data(entry: &ZkEntry, deleted: bool) -> anyhow::Result> { + if deleted { + Self::validate_tombstone_entry(entry)?; + } + Ok(serde_json::to_vec(&ZkShadowMeta::from_entry( + entry, deleted, + ))?) + } + + fn validate_tombstone_entry(entry: &ZkEntry) -> anyhow::Result<()> { + if entry.source_order_millis() <= 0 { + bail!( + "ZK tombstone for {} must have positive source_order_millis", + entry.path + ); + } + if entry.source_zxid <= 0 { + bail!( + "ZK tombstone for {} must have positive source_zxid", + entry.path + ); + } + Ok(()) + } + + async fn write_data_marker(&self, client: &zk::Client) -> anyhow::Result<()> { + if let Some(data_marker) = &self.data_marker { + let data_marker = data_marker.read().await; + let marker_data = serde_json::to_vec(&serde_json::json!({ + "source_id": data_marker.src_node + }))?; + + match client + .set_data(&data_marker.marker, &marker_data, None) + .await + { + Ok(_) => {} + Err(ref e) if is_no_node(e) => { + self.ensure_parent_paths(client, &data_marker.marker) + .await?; + let options = self.persistent_options(); + client + .create(&data_marker.marker, &marker_data, &options) + .await + .map_err(|e| { + anyhow::anyhow!("ZK create marker {} failed: {}", data_marker.marker, e) + })?; + } + Err(e) => bail!("ZK write marker {} failed: {}", data_marker.marker, e), + } + } + Ok(()) + } +} + +const SHADOW_PREFIX: &str = "/__ape_dts_shadow"; + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] +struct ZkShadowMeta { + pub source_id: String, + #[serde(default)] + pub source_order_millis: i64, + #[serde(default)] + pub source_mtime: i64, + #[serde(default)] + pub source_zxid: Option, + #[serde(default)] + pub version: i32, + #[serde(default)] + pub deleted: bool, + #[serde(default)] + pub order_origin: ZkOrderOrigin, +} + +impl ZkShadowMeta { + fn from_entry(entry: &ZkEntry, deleted: bool) -> Self { + Self { + source_id: entry.source_id.clone(), + source_order_millis: entry.source_order_millis(), + source_mtime: entry.stat.mtime, + source_zxid: (entry.source_zxid != 0).then_some(entry.source_zxid), + version: entry.stat.version, + deleted, + order_origin: entry.order_origin.clone(), + } + } + + fn source_order_millis(&self) -> i64 { + if self.source_order_millis != 0 { + self.source_order_millis + } else { + self.source_mtime + } + } + + fn source_zxid(&self) -> i64 { + self.source_zxid.unwrap_or_default() + } + + fn order_key(&self) -> (i64, i64, &str) { + ( + self.source_order_millis(), + self.source_zxid(), + self.source_id.as_str(), + ) + } +} + +fn shadow_path(path: &str) -> String { + format!("{}{}", SHADOW_PREFIX, path) +} + +fn is_node_exists(e: &zk::Error) -> bool { + matches!(e, zk::Error::NodeExists) +} + +fn is_no_node(e: &zk::Error) -> bool { + matches!(e, zk::Error::NoNode) +} + +fn is_not_empty(e: &zk::Error) -> bool { + matches!(e, zk::Error::NotEmpty) +} + +#[cfg(test)] +mod tests { + use super::*; + use dt_common::meta::zk::zk_stat::ZkStat; + + fn entry(order: i64, source_id: &str, event_type: ZkEventType) -> ZkEntry { + ZkEntry { + path: "/app/service".to_string(), + data: Some(b"value".to_vec()), + stat: ZkStat { + version: 1, + mtime: order, + mzxid: order, + ..Default::default() + }, + ephemeral: false, + event_type, + source_id: source_id.to_string(), + source_order_millis: order, + source_zxid: order, + order_origin: ZkOrderOrigin::SourceStatMtime, + } + } + + #[test] + fn newer_tombstone_skips_older_upsert() { + let old_upsert = entry(100, "node-a", ZkEventType::Updated); + let tombstone = ZkShadowMeta { + source_id: "node-b".to_string(), + source_order_millis: 200, + source_mtime: 200, + source_zxid: Some(200), + version: 1, + deleted: true, + order_origin: ZkOrderOrigin::ReconcileObserved, + }; + + assert!(ZkSinker::should_skip_by_shadow_meta( + &tombstone, + &old_upsert + )); + } + + #[test] + fn newer_upsert_can_pass_older_tombstone() { + let new_upsert = entry(300, "node-a", ZkEventType::Updated); + let tombstone = ZkShadowMeta { + source_id: "node-b".to_string(), + source_order_millis: 200, + source_mtime: 200, + source_zxid: Some(200), + version: 1, + deleted: true, + order_origin: ZkOrderOrigin::ReconcileObserved, + }; + + assert!(!ZkSinker::should_skip_by_shadow_meta( + &tombstone, + &new_upsert + )); + } + + #[test] + fn equal_order_uses_source_id_tie_break() { + let entry_from_a = entry(100, "node-a", ZkEventType::Updated); + let shadow_from_b = ZkShadowMeta { + source_id: "node-b".to_string(), + source_order_millis: 100, + source_mtime: 100, + source_zxid: Some(100), + version: 1, + deleted: false, + order_origin: ZkOrderOrigin::SourceStatMtime, + }; + + assert!(ZkSinker::should_skip_by_shadow_meta( + &shadow_from_b, + &entry_from_a + )); + } + + #[test] + fn same_source_same_millis_shadow_zxid_newer_skips_old_upsert() { + let mut old_upsert = entry(100, "node-a", ZkEventType::Updated); + old_upsert.source_zxid = 10; + let newer_tombstone = ZkShadowMeta { + source_id: "node-a".to_string(), + source_order_millis: 100, + source_mtime: 100, + source_zxid: Some(11), + version: 1, + deleted: true, + order_origin: ZkOrderOrigin::ReconcileObserved, + }; + + assert!(ZkSinker::should_skip_by_shadow_meta( + &newer_tombstone, + &old_upsert + )); + } + + #[test] + fn same_source_same_millis_entry_zxid_newer_passes_tombstone() { + let mut new_upsert = entry(100, "node-a", ZkEventType::Updated); + new_upsert.source_zxid = 12; + let older_tombstone = ZkShadowMeta { + source_id: "node-a".to_string(), + source_order_millis: 100, + source_mtime: 100, + source_zxid: Some(11), + version: 1, + deleted: true, + order_origin: ZkOrderOrigin::ReconcileObserved, + }; + + assert!(!ZkSinker::should_skip_by_shadow_meta( + &older_tombstone, + &new_upsert + )); + } + + #[test] + fn delete_shadow_data_is_tombstone() { + let delete = ZkEntry { + data: None, + event_type: ZkEventType::Deleted, + order_origin: ZkOrderOrigin::ReconcileObserved, + ..entry(400, "node-a", ZkEventType::Deleted) + }; + + let bytes = ZkSinker::shadow_data(&delete, true).unwrap(); + let shadow: ZkShadowMeta = serde_json::from_slice(&bytes).unwrap(); + + assert!(shadow.deleted); + assert_eq!(shadow.source_order_millis, 400); + assert_eq!(shadow.source_mtime, 400); + assert_eq!(shadow.source_zxid, Some(400)); + assert_eq!(shadow.order_origin, ZkOrderOrigin::ReconcileObserved); + } + + #[test] + fn delete_shadow_rejects_zero_order() { + let delete = ZkEntry { + data: None, + event_type: ZkEventType::Deleted, + order_origin: ZkOrderOrigin::ReconcileObserved, + source_order_millis: 0, + stat: ZkStat { + mtime: 0, + ..Default::default() + }, + source_zxid: 1, + ..entry(400, "node-a", ZkEventType::Deleted) + }; + + assert!(ZkSinker::shadow_data(&delete, true).is_err()); + } + + #[test] + fn delete_shadow_rejects_zero_zxid() { + let delete = ZkEntry { + data: None, + event_type: ZkEventType::Deleted, + order_origin: ZkOrderOrigin::ReconcileObserved, + source_zxid: 0, + ..entry(400, "node-a", ZkEventType::Deleted) + }; + + assert!(ZkSinker::shadow_data(&delete, true).is_err()); + } + + #[test] + fn old_shadow_json_falls_back_to_source_mtime() { + let old_json = br#"{"source_id":"node-b","source_mtime":200,"version":1}"#; + let shadow: ZkShadowMeta = serde_json::from_slice(old_json).unwrap(); + let old_upsert = entry(100, "node-a", ZkEventType::Updated); + + assert_eq!(shadow.source_order_millis(), 200); + assert!(ZkSinker::should_skip_by_shadow_meta(&shadow, &old_upsert)); + } +} diff --git a/dt-connector/src/zk_router.rs b/dt-connector/src/zk_router.rs new file mode 100644 index 00000000..244f961e --- /dev/null +++ b/dt-connector/src/zk_router.rs @@ -0,0 +1,45 @@ +use dt_common::config::router_config::RouterConfig; + +#[derive(Debug, Clone, Default)] +pub struct ZkRouter { + // sorted by src prefix length descending for longest-prefix-match + pub path_prefix_map: Vec<(String, String)>, +} + +impl ZkRouter { + pub fn from_config(config: &RouterConfig) -> anyhow::Result { + match config { + RouterConfig::Zk { path_prefix_map } => { + let map = Self::parse_path_prefix_map(path_prefix_map)?; + Ok(Self { + path_prefix_map: map, + }) + } + _ => Ok(Self::default()), + } + } + + pub fn route_path(&self, path: &str) -> String { + for (src_prefix, dst_prefix) in &self.path_prefix_map { + if path.starts_with(src_prefix.as_str()) { + return format!("{}{}", dst_prefix, &path[src_prefix.len()..]); + } + } + path.to_string() + } + + fn parse_path_prefix_map(config_str: &str) -> anyhow::Result> { + let mut map = Vec::new(); + if config_str.trim().is_empty() { + return Ok(map); + } + for pair in config_str.split(',') { + let parts: Vec<&str> = pair.split(':').collect(); + if parts.len() == 2 { + map.push((parts[0].trim().to_string(), parts[1].trim().to_string())); + } + } + map.sort_by(|a, b| b.0.len().cmp(&a.0.len())); + Ok(map) + } +} diff --git a/dt-pipeline/src/base_pipeline.rs b/dt-pipeline/src/base_pipeline.rs index 3dd45c92..788b4429 100644 --- a/dt-pipeline/src/base_pipeline.rs +++ b/dt-pipeline/src/base_pipeline.rs @@ -498,7 +498,7 @@ impl BasePipeline { DtData::Ddl { .. } => return SinkMethod::Ddl, DtData::Dcl { .. } => return SinkMethod::Dcl, DtData::Dml { .. } => return SinkMethod::Dml, - DtData::Redis { .. } => return SinkMethod::Raw, + DtData::Redis { .. } | DtData::Zk { .. } => return SinkMethod::Raw, DtData::Begin {} | DtData::Commit { .. } | DtData::Heartbeat {} => continue, } } @@ -656,6 +656,7 @@ mod tests { dt_data::{DtData, DtItem}, position::Position, redis::redis_entry::RedisEntry, + zk::{zk_entry::ZkEntry, zk_event_type::ZkEventType, zk_stat::ZkStat}, }; use dt_connector::extractor::resumer::utils::ResumerUtil; @@ -683,6 +684,34 @@ mod tests { } } + fn zk_item(position: Position) -> DtItem { + DtItem { + dt_data: DtData::Zk { + entry: ZkEntry { + path: "/app/service".to_string(), + data: Some(b"value".to_vec()), + stat: ZkStat::default(), + ephemeral: false, + event_type: ZkEventType::Updated, + source_id: "node-a".to_string(), + source_order_millis: 1, + source_zxid: 1, + order_origin: Default::default(), + }, + }, + position, + data_origin_node: String::new(), + } + } + + fn heartbeat_item(position: Position) -> DtItem { + DtItem { + dt_data: DtData::Heartbeat {}, + position, + data_origin_node: String::new(), + } + } + #[test] fn fetch_raw_collects_latest_position_per_redis_node() { let mut pending_snapshot_finished = HashMap::new(); @@ -706,4 +735,31 @@ mod tests { assert_eq!(by_key.get("redis-node-node-1"), Some(&node_1_new)); assert_eq!(by_key.get("redis-node-node-2"), Some(&node_2)); } + + #[test] + fn fetch_raw_advances_zk_position_only_on_trailing_heartbeat() { + let mut pending_snapshot_finished = HashMap::new(); + let final_position = Position::Zk { + path_versions: HashMap::from([("/app/service".to_string(), (1, 10))]), + last_scan_timestamp: 100, + high_water_zxid: 10, + total_paths: 1, + }; + let partial_data = vec![zk_item(Position::None)]; + + let (_, partial_position, partial_commits) = + BasePipeline::fetch_raw(&partial_data, &mut pending_snapshot_finished); + assert_eq!(partial_position, Some(Position::None)); + assert!(partial_commits.is_empty()); + + let complete_data = vec![ + zk_item(Position::None), + heartbeat_item(final_position.clone()), + ]; + let (_, complete_position, complete_commits) = + BasePipeline::fetch_raw(&complete_data, &mut pending_snapshot_finished); + + assert_eq!(complete_position, Some(final_position.clone())); + assert_eq!(complete_commits, vec![final_position]); + } } diff --git a/dt-task/src/extractor_util.rs b/dt-task/src/extractor_util.rs index e194d08c..2110b0d8 100644 --- a/dt-task/src/extractor_util.rs +++ b/dt-task/src/extractor_util.rs @@ -56,6 +56,7 @@ use dt_connector::{ redis_snapshot_file_extractor::RedisSnapshotFileExtractor, }, resumer::recovery::Recovery, + zk::zk_extractor::ZkExtractor, }, rdb_router::RdbRouter, Extractor, @@ -724,6 +725,28 @@ impl ExtractorUtil { }; Box::new(extractor) } + ExtractorConfig::Zk { + url, + watch_paths, + scan_interval_secs, + include_ephemeral, + heartbeat_interval_secs, + } => { + let zk_filter = dt_common::zk_filter::ZkFilter::from_config(&config.zk_filter)?; + let extractor = ZkExtractor { + url, + watch_paths, + scan_interval_secs, + include_ephemeral, + heartbeat_interval_secs, + filter: zk_filter, + base_extractor, + extract_state, + syncer, + recovery, + }; + Box::new(extractor) + } }; Ok(extractor) } diff --git a/dt-task/src/sinker_util.rs b/dt-task/src/sinker_util.rs index e8e90a27..c7254a8d 100644 --- a/dt-task/src/sinker_util.rs +++ b/dt-task/src/sinker_util.rs @@ -45,7 +45,9 @@ use dt_connector::{ starrocks::{ starrocks_sinker::StarRocksSinker, starrocks_struct_sinker::StarrocksStructSinker, }, + zk::zk_sinker::ZkSinker, }, + zk_router::ZkRouter, Sinker, }; @@ -554,6 +556,29 @@ impl SinkerUtil { Self::push_sinker(&mut sub_sinkers, sinker); } } + SinkerConfig::Zk { + url, + batch_size, + create_if_not_exists, + sync_ephemeral_as_persistent, + conflict_policy, + } => { + let zk_router = ZkRouter::from_config(&config.router)?; + for _ in 0..parallel_size { + let sinker = ZkSinker { + url: url.clone(), + batch_size, + create_if_not_exists, + sync_ephemeral_as_persistent, + conflict_policy: conflict_policy.clone(), + router: zk_router.clone(), + base_sinker: BaseSinker::new(monitor.clone(), monitor_interval), + data_marker: data_marker.clone(), + client: None, + }; + Self::push_sinker(&mut sub_sinkers, sinker); + } + } }; Ok(sub_sinkers) } diff --git a/dt-tests/docker-compose.zk.yml b/dt-tests/docker-compose.zk.yml new file mode 100644 index 00000000..20d81726 --- /dev/null +++ b/dt-tests/docker-compose.zk.yml @@ -0,0 +1,30 @@ +services: + zk-src: + image: zookeeper:3.9 + container_name: zk-src-it + ports: + - "2181:2181" + environment: + ZOO_MY_ID: 1 + ZOO_STANDALONE_ENABLED: "true" + healthcheck: + test: ["CMD-SHELL", "echo srvr | nc localhost 2181 | grep standalone"] + timeout: 5s + retries: 5 + start_period: 10s + interval: 5s + + zk-dst: + image: zookeeper:3.9 + container_name: zk-dst-it + ports: + - "2182:2181" + environment: + ZOO_MY_ID: 1 + ZOO_STANDALONE_ENABLED: "true" + healthcheck: + test: ["CMD-SHELL", "echo srvr | nc localhost 2181 | grep standalone"] + timeout: 5s + retries: 5 + start_period: 10s + interval: 5s diff --git a/dt-tests/evidence/zk_to_zk/node1_to_node2_verify.log b/dt-tests/evidence/zk_to_zk/node1_to_node2_verify.log new file mode 100644 index 00000000..1b685746 --- /dev/null +++ b/dt-tests/evidence/zk_to_zk/node1_to_node2_verify.log @@ -0,0 +1,113 @@ +Connecting to localhost:2181 +2026-06-06 12:59:07,389 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:zookeeper.version=3.9.5-293c895a8d966a3ecb92872be4a1daf87d725da2, built on 2026-02-11 20:18 UTC +2026-06-06 12:59:07,390 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:host.name=56cf11fd6120 +2026-06-06 12:59:07,390 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.version=17.0.19 +2026-06-06 12:59:07,390 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.vendor=Eclipse Adoptium +2026-06-06 12:59:07,390 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.home=/opt/java/openjdk +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.class.path=/apache-zookeeper-3.9.5-bin/bin/../zookeeper-metrics-providers/zookeeper-prometheus-metrics/target/classes:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/target/classes:/apache-zookeeper-3.9.5-bin/bin/../build/classes:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-metrics-providers/zookeeper-prometheus-metrics/target/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/target/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../build/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-prometheus-metrics-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-jute-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/snappy-java-1.1.10.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/slf4j-api-2.0.13.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_servlet-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_hotspot-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_common-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-native-unix-common-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-native-epoll-4.1.130.Final-linux-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-classes-epoll-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-classes-2.0.74.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-windows-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-osx-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-osx-aarch_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-linux-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-linux-aarch_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-resolver-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-handler-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-common-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-codec-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-buffer-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/metrics-core-4.1.12.1.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/logback-core-1.3.15.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/logback-classic-1.3.15.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jline-3.25.1.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-util-ajax-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-util-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-servlet-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-server-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-security-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-io-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-http-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/javax.servlet-api-3.1.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-databind-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-core-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-annotations-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/commons-io-2.17.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/commons-cli-1.5.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/audience-annotations-0.12.0.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-*.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/src/main/resources/lib/*.jar:/conf: +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.library.path=/usr/java/packages/lib:/usr/lib64:/lib64:/lib:/usr/lib +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.io.tmpdir=/tmp +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.compiler= +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.name=Linux +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.arch=aarch64 +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.version=6.12.76-linuxkit +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.name=root +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.home=/root +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.dir=/apache-zookeeper-3.9.5-bin +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.free=114MB +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.max=256MB +2026-06-06 12:59:07,391 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.total=128MB +2026-06-06 12:59:07,392 [myid:] - INFO [main:o.a.z.ZooKeeper@635] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@25b485ba +2026-06-06 12:59:07,395 [myid:] - INFO [main:o.a.z.c.X509Util@93] - Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation +2026-06-06 12:59:07,460 [myid:] - INFO [main:o.a.z.c.X509Util@115] - Default TLS protocol is TLSv1.3, supported TLS protocols are [TLSv1.3, TLSv1.2, TLSv1.1, TLSv1, SSLv3, SSLv2Hello] +2026-06-06 12:59:07,463 [myid:] - INFO [main:o.a.z.ClientCnxnSocket@235] - jute.maxbuffer value is 1048575 Bytes +2026-06-06 12:59:07,466 [myid:] - INFO [main:o.a.z.ClientCnxn@1734] - zookeeper.request.timeout value is 0. feature enabled=false +2026-06-06 12:59:07,467 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1170] - Opening socket connection to server localhost/[0:0:0:0:0:0:0:1]:2181. +2026-06-06 12:59:07,467 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1172] - SASL config status: Will not attempt to authenticate using SASL (unknown error) +2026-06-06 12:59:07,470 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1006] - Socket connection established, initiating session, client: /[0:0:0:0:0:0:0:1]:44318, server: localhost/[0:0:0:0:0:0:0:1]:2181 +2026-06-06 12:59:07,472 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1438] - Session establishment complete on server localhost/[0:0:0:0:0:0:0:1]:2181, session id = 0x1002ba28b500009, negotiated timeout = 30000 + +WATCHER:: + +WatchedEvent state:SyncConnected type:None path:null zxid: -1 +/ +/__ape_dts_marker +/app +/zookeeper +/app/service-a +/app/service-b +/app/service1 +/app/service-a/config +/zookeeper/config +/zookeeper/quota +2026-06-06 12:59:07,478 [myid:] - INFO [main:o.a.z.u.ServiceUtils@45] - Exiting JVM with code 0 +---service-a--- +Connecting to localhost:2181 +2026-06-06 12:59:08,100 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:zookeeper.version=3.9.5-293c895a8d966a3ecb92872be4a1daf87d725da2, built on 2026-02-11 20:18 UTC +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:host.name=56cf11fd6120 +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.version=17.0.19 +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.vendor=Eclipse Adoptium +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.home=/opt/java/openjdk +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.class.path=/apache-zookeeper-3.9.5-bin/bin/../zookeeper-metrics-providers/zookeeper-prometheus-metrics/target/classes:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/target/classes:/apache-zookeeper-3.9.5-bin/bin/../build/classes:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-metrics-providers/zookeeper-prometheus-metrics/target/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/target/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../build/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-prometheus-metrics-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-jute-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/snappy-java-1.1.10.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/slf4j-api-2.0.13.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_servlet-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_hotspot-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_common-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-native-unix-common-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-native-epoll-4.1.130.Final-linux-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-classes-epoll-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-classes-2.0.74.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-windows-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-osx-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-osx-aarch_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-linux-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-linux-aarch_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-resolver-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-handler-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-common-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-codec-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-buffer-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/metrics-core-4.1.12.1.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/logback-core-1.3.15.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/logback-classic-1.3.15.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jline-3.25.1.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-util-ajax-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-util-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-servlet-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-server-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-security-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-io-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-http-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/javax.servlet-api-3.1.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-databind-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-core-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-annotations-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/commons-io-2.17.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/commons-cli-1.5.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/audience-annotations-0.12.0.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-*.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/src/main/resources/lib/*.jar:/conf: +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.library.path=/usr/java/packages/lib:/usr/lib64:/lib64:/lib:/usr/lib +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.io.tmpdir=/tmp +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.compiler= +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.name=Linux +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.arch=aarch64 +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.version=6.12.76-linuxkit +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.name=root +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.home=/root +2026-06-06 12:59:08,101 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.dir=/apache-zookeeper-3.9.5-bin +2026-06-06 12:59:08,102 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.free=114MB +2026-06-06 12:59:08,102 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.max=256MB +2026-06-06 12:59:08,102 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.total=128MB +2026-06-06 12:59:08,102 [myid:] - INFO [main:o.a.z.ZooKeeper@635] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@25b485ba +2026-06-06 12:59:08,105 [myid:] - INFO [main:o.a.z.c.X509Util@93] - Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation +2026-06-06 12:59:08,170 [myid:] - INFO [main:o.a.z.c.X509Util@115] - Default TLS protocol is TLSv1.3, supported TLS protocols are [TLSv1.3, TLSv1.2, TLSv1.1, TLSv1, SSLv3, SSLv2Hello] +2026-06-06 12:59:08,172 [myid:] - INFO [main:o.a.z.ClientCnxnSocket@235] - jute.maxbuffer value is 1048575 Bytes +2026-06-06 12:59:08,175 [myid:] - INFO [main:o.a.z.ClientCnxn@1734] - zookeeper.request.timeout value is 0. feature enabled=false +2026-06-06 12:59:08,177 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1170] - Opening socket connection to server localhost/[0:0:0:0:0:0:0:1]:2181. +2026-06-06 12:59:08,177 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1172] - SASL config status: Will not attempt to authenticate using SASL (unknown error) +2026-06-06 12:59:08,179 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1006] - Socket connection established, initiating session, client: /[0:0:0:0:0:0:0:1]:44328, server: localhost/[0:0:0:0:0:0:0:1]:2181 +2026-06-06 12:59:08,181 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1438] - Session establishment complete on server localhost/[0:0:0:0:0:0:0:1]:2181, session id = 0x1002ba28b50000a, negotiated timeout = 30000 + +WATCHER:: + +WatchedEvent state:SyncConnected type:None path:null zxid: -1 +host1:8080 +2026-06-06 12:59:08,183 [myid:] - INFO [main:o.a.z.u.ServiceUtils@45] - Exiting JVM with code 0 +---service-b--- +Connecting to localhost:2181 +2026-06-06 12:59:08,819 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:zookeeper.version=3.9.5-293c895a8d966a3ecb92872be4a1daf87d725da2, built on 2026-02-11 20:18 UTC +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:host.name=56cf11fd6120 +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.version=17.0.19 +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.vendor=Eclipse Adoptium +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.home=/opt/java/openjdk +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.class.path=/apache-zookeeper-3.9.5-bin/bin/../zookeeper-metrics-providers/zookeeper-prometheus-metrics/target/classes:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/target/classes:/apache-zookeeper-3.9.5-bin/bin/../build/classes:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-metrics-providers/zookeeper-prometheus-metrics/target/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/target/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../build/lib/*.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-prometheus-metrics-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-jute-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/zookeeper-3.9.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/snappy-java-1.1.10.5.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/slf4j-api-2.0.13.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_servlet-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_hotspot-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient_common-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/simpleclient-0.9.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-native-unix-common-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-native-epoll-4.1.130.Final-linux-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-classes-epoll-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-transport-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-classes-2.0.74.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-windows-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-osx-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-osx-aarch_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-linux-x86_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final-linux-aarch_64.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-tcnative-boringssl-static-2.0.74.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-resolver-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-handler-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-common-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-codec-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/netty-buffer-4.1.130.Final.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/metrics-core-4.1.12.1.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/logback-core-1.3.15.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/logback-classic-1.3.15.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jline-3.25.1.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-util-ajax-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-util-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-servlet-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-server-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-security-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-io-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jetty-http-9.4.58.v20250814.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/javax.servlet-api-3.1.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-databind-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-core-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/jackson-annotations-2.15.2.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/commons-io-2.17.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/commons-cli-1.5.0.jar:/apache-zookeeper-3.9.5-bin/bin/../lib/audience-annotations-0.12.0.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-*.jar:/apache-zookeeper-3.9.5-bin/bin/../zookeeper-server/src/main/resources/lib/*.jar:/conf: +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.library.path=/usr/java/packages/lib:/usr/lib64:/lib64:/lib:/usr/lib +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.io.tmpdir=/tmp +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:java.compiler= +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.name=Linux +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.arch=aarch64 +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.version=6.12.76-linuxkit +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.name=root +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.home=/root +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:user.dir=/apache-zookeeper-3.9.5-bin +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.free=114MB +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.max=256MB +2026-06-06 12:59:08,820 [myid:] - INFO [main:o.a.z.Environment@98] - Client environment:os.memory.total=128MB +2026-06-06 12:59:08,821 [myid:] - INFO [main:o.a.z.ZooKeeper@635] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@25b485ba +2026-06-06 12:59:08,824 [myid:] - INFO [main:o.a.z.c.X509Util@93] - Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation +2026-06-06 12:59:08,888 [myid:] - INFO [main:o.a.z.c.X509Util@115] - Default TLS protocol is TLSv1.3, supported TLS protocols are [TLSv1.3, TLSv1.2, TLSv1.1, TLSv1, SSLv3, SSLv2Hello] +2026-06-06 12:59:08,891 [myid:] - INFO [main:o.a.z.ClientCnxnSocket@235] - jute.maxbuffer value is 1048575 Bytes +2026-06-06 12:59:08,893 [myid:] - INFO [main:o.a.z.ClientCnxn@1734] - zookeeper.request.timeout value is 0. feature enabled=false +2026-06-06 12:59:08,895 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1170] - Opening socket connection to server localhost/127.0.0.1:2181. +2026-06-06 12:59:08,895 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1172] - SASL config status: Will not attempt to authenticate using SASL (unknown error) +2026-06-06 12:59:08,897 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1006] - Socket connection established, initiating session, client: /127.0.0.1:60954, server: localhost/127.0.0.1:2181 +2026-06-06 12:59:08,900 [myid:localhost:2181] - INFO [main-SendThread(localhost:2181):o.a.z.ClientCnxn$SendThread@1438] - Session establishment complete on server localhost/127.0.0.1:2181, session id = 0x1002ba28b50000b, negotiated timeout = 30000 + +WATCHER:: + +WatchedEvent state:SyncConnected type:None path:null zxid: -1 +host2:9090 +2026-06-06 12:59:08,902 [myid:] - INFO [main:o.a.z.u.ServiceUtils@45] - Exiting JVM with code 0 diff --git a/dt-tests/scripts/zk_integration_test.sh b/dt-tests/scripts/zk_integration_test.sh new file mode 100755 index 00000000..ddf5a9d8 --- /dev/null +++ b/dt-tests/scripts/zk_integration_test.sh @@ -0,0 +1,121 @@ +#!/bin/bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +DT_TESTS_DIR="$(dirname "$SCRIPT_DIR")" +PROJECT_ROOT="$(dirname "$DT_TESTS_DIR")" + +COMPOSE_FILE="$DT_TESTS_DIR/docker-compose.zk.yml" +EVIDENCE_DIR="$DT_TESTS_DIR/evidence/zk_to_zk" +TASK_DIR="$DT_TESTS_DIR/tests/zk_to_zk/cdc/basic_test" + +ZK1_HOST="localhost:2181" +ZK2_HOST="localhost:2182" + +mkdir -p "$EVIDENCE_DIR" + +log() { echo "[$(date '+%H:%M:%S')] $*"; } + +cleanup() { + log "Stopping containers..." + docker-compose -f "$COMPOSE_FILE" down -v 2>/dev/null || true +} + +wait_for_zk() { + local host=$1 name=$2 + log "Waiting for $name ($host)..." + for i in $(seq 1 30); do + if echo ruok | nc -w 2 "${host%%:*}" "${host##*:}" 2>/dev/null | grep -q imok; then + log "$name ready" + return 0 + fi + sleep 1 + done + log "ERROR: $name not ready after 30s" + return 1 +} + +phase_1_start_zk() { + log "=== Phase 1: Start dual ZK ensembles ===" + docker-compose -f "$COMPOSE_FILE" up -d + wait_for_zk "$ZK1_HOST" "zk-node1" + wait_for_zk "$ZK2_HOST" "zk-node2" + log "Phase 1 PASS" +} + +phase_2_seed_data() { + log "=== Phase 2: Seed test data on zk-node1 ===" + docker exec zk-node1 /apache-zookeeper-*/bin/zkCli.sh -server localhost:2181 <<'ZK_CMDS' +create /app "" +create /app/service-a "host1:8080" +create /app/service-b "host2:9090" +create /app/service-a/config "timeout=30" +quit +ZK_CMDS + log "Seeded: /app/service-a, /app/service-b, /app/service-a/config" + log "Phase 2 PASS" +} + +phase_3_run_sync() { + log "=== Phase 3: Run ape-dts sync (node1 → node2) ===" + log "NOTE: Build and run manually:" + log " cargo build --release -p dt-main" + log " ./target/release/dt-main --task_conf $TASK_DIR/topo1_node1_to_node2/task_config.ini" + log "" + log "After sync runs for ~10 seconds, Ctrl+C and proceed to phase 4." +} + +phase_4_verify() { + log "=== Phase 4: Verify sync results on zk-node2 ===" + local result_file="$EVIDENCE_DIR/verify_$(date +%Y%m%d_%H%M%S).log" + + { + echo "=== zk-node2 data after sync ===" + docker exec zk-node2 /apache-zookeeper-*/bin/zkCli.sh -server localhost:2181 <<'ZK_CMDS' +ls /app +get /app/service-a +get /app/service-b +get /app/service-a/config +quit +ZK_CMDS + } 2>&1 | tee "$result_file" + + log "Evidence saved to: $result_file" + + if grep -q "host1:8080" "$result_file" && grep -q "host2:9090" "$result_file"; then + log "Phase 4 PASS — data synced correctly" + else + log "Phase 4 FAIL — data not found on zk-node2" + fi +} + +phase_5_verify_bidirectional() { + log "=== Phase 5: Verify bidirectional sync ===" + log "1. Write new data on zk-node2:" + log " docker exec zk-node2 zkCli.sh create /app/service-c 'host3:7070'" + log "2. Run reverse sync task:" + log " ./target/release/dt-main --task_conf $TASK_DIR/topo1_node2_to_node1/task_config.ini" + log "3. Verify on zk-node1:" + log " docker exec zk-node1 zkCli.sh get /app/service-c" +} + +case "${1:-all}" in + start) phase_1_start_zk ;; + seed) phase_2_seed_data ;; + sync) phase_3_run_sync ;; + verify) phase_4_verify ;; + bidir) phase_5_verify_bidirectional ;; + cleanup) cleanup ;; + all) + trap cleanup EXIT + phase_1_start_zk + phase_2_seed_data + phase_3_run_sync + phase_4_verify + phase_5_verify_bidirectional + ;; + *) + echo "Usage: $0 {start|seed|sync|verify|bidir|cleanup|all}" + exit 1 + ;; +esac diff --git a/dt-tests/tests/zk_to_zk/cdc/basic_test/task_config.ini b/dt-tests/tests/zk_to_zk/cdc/basic_test/task_config.ini new file mode 100644 index 00000000..dac900d5 --- /dev/null +++ b/dt-tests/tests/zk_to_zk/cdc/basic_test/task_config.ini @@ -0,0 +1,38 @@ +[extractor] +db_type=zk +extract_type=cdc +url=localhost:2181 +watch_paths=/app,/config +scan_interval_secs=5 +include_ephemeral=false +heartbeat_interval_secs=10 + +[zk_filter] +do_paths=* +ignore_paths=/zookeeper,/__ape_dts_shadow +include_ephemeral=false + +[sinker] +db_type=zk +sink_type=write +url=localhost:2182 +batch_size=1 +create_if_not_exists=true +sync_ephemeral_as_persistent=true +conflict_policy=last_write_wins + +[router] +path_prefix_map= + +[pipeline] +buffer_size=4 +checkpoint_interval_secs=1 + +[parallelizer] +parallel_type=serial +parallel_size=1 + +[runtime] +log_level=info +log4rs_file=./log4rs.yaml +log_dir=./logs diff --git a/dt-tests/tests/zk_to_zk/cdc/basic_test/topo1_node1_to_node2/task_config.ini b/dt-tests/tests/zk_to_zk/cdc/basic_test/topo1_node1_to_node2/task_config.ini new file mode 100644 index 00000000..30457c00 --- /dev/null +++ b/dt-tests/tests/zk_to_zk/cdc/basic_test/topo1_node1_to_node2/task_config.ini @@ -0,0 +1,49 @@ +[extractor] +db_type=zk +extract_type=cdc +url=localhost:2181 +watch_paths=/app +scan_interval_secs=5 +include_ephemeral=false +heartbeat_interval_secs=10 + +[zk_filter] +do_paths=/app +ignore_paths=/zookeeper,/__ape_dts_heartbeat,/__ape_dts_marker,/__ape_dts_shadow +include_ephemeral=false + +[data_marker] +topo_name=topo1 +topo_nodes=node1,node2 +src_node=node1 +dst_node=node2 +do_nodes=node1 +ignore_nodes=node2 +marker=/__ape_dts_marker + +[sinker] +db_type=zk +sink_type=write +url=localhost:2182 +batch_size=200 +create_if_not_exists=true +sync_ephemeral_as_persistent=false +conflict_policy=last_write_wins + +[router] +db_map= +col_map= +tb_map= + +[parallelizer] +parallel_type=serial +parallel_size=1 + +[pipeline] +buffer_size=100 +checkpoint_interval_secs=1 + +[runtime] +log_level=info +log4rs_file=./log4rs.yaml +log_dir=./logs diff --git a/dt-tests/tests/zk_to_zk/cdc/basic_test/topo1_node2_to_node1/task_config.ini b/dt-tests/tests/zk_to_zk/cdc/basic_test/topo1_node2_to_node1/task_config.ini new file mode 100644 index 00000000..82011b42 --- /dev/null +++ b/dt-tests/tests/zk_to_zk/cdc/basic_test/topo1_node2_to_node1/task_config.ini @@ -0,0 +1,49 @@ +[extractor] +db_type=zk +extract_type=cdc +url=localhost:2182 +watch_paths=/app +scan_interval_secs=5 +include_ephemeral=false +heartbeat_interval_secs=10 + +[zk_filter] +do_paths=/app +ignore_paths=/zookeeper,/__ape_dts_heartbeat,/__ape_dts_marker,/__ape_dts_shadow +include_ephemeral=false + +[data_marker] +topo_name=topo1 +topo_nodes=node1,node2 +src_node=node2 +dst_node=node1 +do_nodes=node2 +ignore_nodes=node1 +marker=/__ape_dts_marker + +[sinker] +db_type=zk +sink_type=write +url=localhost:2181 +batch_size=200 +create_if_not_exists=true +sync_ephemeral_as_persistent=false +conflict_policy=last_write_wins + +[router] +db_map= +col_map= +tb_map= + +[parallelizer] +parallel_type=serial +parallel_size=1 + +[pipeline] +buffer_size=100 +checkpoint_interval_secs=1 + +[runtime] +log_level=info +log4rs_file=./log4rs.yaml +log_dir=./logs