fix: Client::create does not work on ZooKeeper 3.4.x (#19)

This commit adds client side assumed ZooKeeper version so that client
can fallback to `OpCode::create` instead of `OpCode::create2` on 3.4.x.

Resolves #18.

Author: kezhuw

src/client/mod.rs

@@ -199,7 +199,8 @@ impl CreateSequence {
 /// Client encapsulates ZooKeeper session to interact with ZooKeeper cluster.
 ///
 /// Besides semantic errors, node operations could also fail due to cluster availability and
-/// limitations, e.g. [Error::ConnectionLoss], [Error::QuotaExceeded] and so on.
+/// capabilities, e.g. [Error::ConnectionLoss], [Error::QuotaExceeded], [Error::Unimplemented] and
+/// so on.
 ///
 /// All remote operations will fail after session expired, failed or closed.
 ///
@@ -211,6 +212,7 @@ impl CreateSequence {
 #[derive(Clone, Debug)]
 pub struct Client {
     chroot: OwnedChroot,
+    version: Version,
     session: (SessionId, Vec<u8>),
     session_timeout: Duration,
     requester: mpsc::UnboundedSender<SessionOperation>,
@@ -232,13 +234,14 @@ impl Client {
 
     pub(crate) fn new(
         chroot: OwnedChroot,
+        version: Version,
         session: (SessionId, Vec<u8>),
         timeout: Duration,
         requester: mpsc::UnboundedSender<SessionOperation>,
         state_receiver: watch::Receiver<SessionState>,
     ) -> Client {
         let state_watcher = StateWatcher::new(state_receiver);
-        Client { chroot, session, session_timeout: timeout, requester, state_watcher }
+        Client { chroot, version, session, session_timeout: timeout, requester, state_watcher }
     }
 
     fn validate_path<'a>(&'a self, path: &'a str) -> Result<ChrootPath<'a>> {
@@ -422,6 +425,12 @@ impl Client {
     /// * [Error::NoNode] if parent node does not exist.
     /// * [Error::NoChildrenForEphemerals] if parent node is ephemeral.
     /// * [Error::InvalidAcl] if acl is invalid or empty.
+    ///
+    /// # Notable behaviors
+    /// The resulting [Stat] will be [Stat::is_invalid] if assumed server version is 3.4 series or
+    /// below. See [ClientBuilder::assume_server_version] and [ZOOKEEPER-1297][].
+    ///
+    /// [ZOOKEEPER-1297]: https://issues.apache.org/jira/browse/ZOOKEEPER-1297
     pub fn create<'a: 'f, 'b: 'f, 'f>(
         &'a self,
         path: &'b str,
@@ -449,8 +458,10 @@ impl Client {
             OpCode::CreateTtl
         } else if create_mode.is_container() {
             OpCode::CreateContainer
-        } else {
+        } else if self.version >= Version(3, 5, 0) {
             OpCode::Create2
+        } else {
+            OpCode::Create
         };
         let flags = create_mode.as_flags(ttl != 0);
         let request = CreateRequest { path: chroot_path, data, acls: options.acls, flags, ttl };
@@ -461,7 +472,8 @@ impl Client {
             let server_path = record::unmarshal_entity::<&str>(&"server path", &mut buf)?;
             let client_path = util::strip_root_path(server_path, self.chroot.root())?;
             let sequence = if sequential { Self::parse_sequence(client_path, path)? } else { CreateSequence(-1) };
-            let stat = record::unmarshal::<Stat>(&mut buf)?;
+            let stat =
+                if op_code == OpCode::Create { Stat::new_invalid() } else { record::unmarshal::<Stat>(&mut buf)? };
             Ok((stat, sequence))
         })
     }
@@ -1508,10 +1520,14 @@ impl Drop for OwnedLockClient {
     }
 }
 
+#[derive(Copy, Clone, Debug, PartialEq, PartialOrd)]
+pub(crate) struct Version(u32, u32, u32);
+
 /// Builder for [Client] with more options than [Client::connect].
 #[derive(Clone, Debug)]
 pub struct ClientBuilder {
     authes: Vec<AuthPacket>,
+    version: Version,
     session: Option<(SessionId, Vec<u8>)>,
     readonly: bool,
     detached: bool,
@@ -1523,6 +1539,7 @@ impl ClientBuilder {
     fn new() -> Self {
         Self {
             authes: Default::default(),
+            version: Version(u32::MAX, u32::MAX, u32::MAX),
             session: None,
             readonly: false,
             detached: false,
@@ -1565,6 +1582,20 @@ impl ClientBuilder {
         self
     }
 
+    /// Specifies client assumed server version of ZooKeeper cluster.
+    ///
+    /// Client will issue server compatible protocol to avoid [Error::Unimplemented] for some
+    /// operations. See [Client::create] for an example.
+    ///
+    /// See [ZOOKEEPER-1381][] and [ZOOKEEPER-3762][] for references.
+    ///
+    /// [ZOOKEEPER-1381]: https://issues.apache.org/jira/browse/ZOOKEEPER-1381
+    /// [ZOOKEEPER-3762]: https://issues.apache.org/jira/browse/ZOOKEEPER-3762
+    pub fn assume_server_version(&mut self, major: u32, minor: u32, patch: u32) -> &mut Self {
+        self.version = Version(major, minor, patch);
+        self
+    }
+
     /// Detaches creating session so it will not be closed after all client instances dropped.
     pub fn detach(&mut self) -> &mut Self {
         self.detached = true;
@@ -1608,7 +1639,8 @@ impl ClientBuilder {
         tokio::spawn(async move {
             session.serve(servers, sock, buf, connecting_depot, receiver).await;
         });
-        let client = Client::new(chroot.to_owned(), session_info, session_timeout, sender, state_receiver);
+        let client =
+            Client::new(chroot.to_owned(), self.version, session_info, session_timeout, sender, state_receiver);
         Ok(client)
     }
 }

src/session/types.rs

@@ -85,7 +85,7 @@ pub struct WatchedEvent {
     ///
     /// # Notable behaviors
     /// * This feature was shipped in 3.9.0, `zxid` wil be [WatchedEvent::NO_ZXID] for earlier versions. See [ZOOKEEPER-4655].
-    /// * It is possible to receive multiple events with same `zxid` and even same `path` due to [MultiWriter]. See [ZOOKEEPER-4695].
+    /// * It is possible to receive multiple events with same `zxid` and even same `path` due to [crate::MultiWriter]. See [ZOOKEEPER-4695].
     ///
     /// [ZOOKEEPER-4655]: https://issues.apache.org/jira/browse/ZOOKEEPER-4655
     /// [ZOOKEEPER-4695]: https://issues.apache.org/jira/browse/ZOOKEEPER-4695

tests/zookeeper.rs

@@ -23,24 +23,32 @@ fn random_data() -> Vec<u8> {
     rng.sample_iter(Standard).take(32).collect()
 }
 
-fn zookeeper_image_with_properties<'a>(mut properties: Vec<&'a str>) -> GenericImage {
+fn zookeeper_image_with_version_and_properties<'a>(version: &'a str, mut properties: Vec<&'a str>) -> GenericImage {
     properties.insert(0, "-Dzookeeper.DigestAuthenticationProvider.superDigest=super:D/InIHSb7yEEbrWz8b9l71RjZJU=");
     properties.insert(0, "-Dzookeeper.enableEagerACLCheck=true");
     let jvm_properties = properties.join(" ");
     let healthcheck = Healthcheck::default()
         .with_cmd(["./bin/zkServer.sh", "status"].iter())
         .with_interval(Duration::from_secs(2))
         .with_retries(60);
-    GenericImage::new("zookeeper", "3.9.0")
+    GenericImage::new("zookeeper", version)
         .with_env_var("SERVER_JVMFLAGS", jvm_properties)
         .with_healthcheck(healthcheck)
         .with_wait_for(WaitFor::Healthcheck)
 }
 
+fn zookeeper_image_with_properties<'a>(properties: Vec<&'a str>) -> GenericImage {
+    zookeeper_image_with_version_and_properties("3.9.0", properties)
+}
+
 fn zookeeper_image() -> GenericImage {
     zookeeper_image_with_properties(Vec::default())
 }
 
+fn zookeeper34_image() -> GenericImage {
+    zookeeper_image_with_version_and_properties("3.4", Vec::default())
+}
+
 async fn example() {
     let docker = DockerCli::default();
     let zookeeper = docker.run(zookeeper_image());
@@ -591,6 +599,44 @@ async fn test_create_container() {
     assert_that!(client.delete("/container", None).await.unwrap_err()).is_equal_to(zk::Error::NoNode);
 }
 
+#[test_log::test(tokio::test)]
+async fn test_zookeeper34() {
+    let docker = DockerCli::default();
+    let zookeeper = docker.run(zookeeper34_image());
+    let zk_port = zookeeper.get_host_port(2181);
+    let cluster = format!("127.0.0.1:{}", zk_port);
+    let client = zk::Client::builder().assume_server_version(3, 4, u32::MAX).connect(&cluster).await.unwrap();
+    let (stat, _sequence) = client.create("/a", b"a1", PERSISTENT_OPEN).await.unwrap();
+    assert_that!(stat.is_invalid()).is_true();
+
+    let (data, stat) = client.get_data("/a").await.unwrap();
+    assert_eq!(data, b"a1".to_vec());
+    assert_eq!(stat.version, 0);
+
+    let (data, stat, watcher) = client.get_and_watch_data("/a").await.unwrap();
+    assert_eq!(data, b"a1".to_vec());
+    assert_eq!(stat.version, 0);
+
+    let stat = client.set_data("/a", b"a2", Some(0)).await.unwrap();
+    assert_eq!(stat.version, 1);
+
+    let event = watcher.changed().await;
+    assert_eq!(event.path, "/a");
+    assert_eq!(event.zxid, -1);
+    assert_eq!(event.event_type, zk::EventType::NodeDataChanged);
+
+    let (data, stat, watcher) = client.get_and_watch_data("/a").await.unwrap();
+    assert_eq!(data, b"a2".to_vec());
+    assert_eq!(stat.version, 1);
+
+    client.delete("/a", Some(1)).await.unwrap();
+
+    let event = watcher.changed().await;
+    assert_eq!(event.path, "/a");
+    assert_eq!(event.zxid, -1);
+    assert_eq!(event.event_type, zk::EventType::NodeDeleted);
+}
+
 #[test_log::test(tokio::test)]
 async fn test_mkdir() {
     let docker = DockerCli::default();