CURATOR-704. Add server compatibility check support

curator-client/src/main/java/org/apache/curator/utils/ZookeeperCompatibility.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.curator.utils;
+
+/**
+ * Describe feature supports based on server compatibility (as opposed to
+ * {@code Compatibility} which represents client compatibility.
+ */
+public class ZookeeperCompatibility {
+    /**
+     * Represent latest version with all features enabled
+     */
+    public static final ZookeeperCompatibility LATEST =
+            builder().hasPersistentWatchers(true).build();
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        // List of features introduced by Zookeeper over time.
+        // All values are set to false by default for backward compatibility
+        private boolean hasPersistentWatchers = false;
+
+        public Builder hasPersistentWatchers(boolean value) {
+            this.hasPersistentWatchers = value;
+            return this;
+        }
+
+        public boolean hasPersistentWatchers() {
+            return this.hasPersistentWatchers;
+        }
+
+        public ZookeeperCompatibility build() {
+            return new ZookeeperCompatibility(this);
+        }
+    }
+
+    private final boolean hasPersistentWatchers;
+
+    private ZookeeperCompatibility(Builder builder) {
+        this.hasPersistentWatchers = builder.hasPersistentWatchers;
+    }
+
+    /**
+     * Check if both client and server support persistent watchers
+     *
+     * @return {@code true} if both the client library and the server version
+     *         support persistent watchers
+     */
+    public boolean hasPersistentWatchers() {
+        return this.hasPersistentWatchers && Compatibility.hasPersistentWatchers();
+    }
+}

curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java

@@ -34,6 +34,7 @@
 import org.apache.curator.framework.state.ConnectionStateErrorPolicy;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.utils.EnsurePath;
+import org.apache.curator.utils.ZookeeperCompatibility;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 
@@ -262,6 +263,13 @@ public interface CuratorFramework extends Closeable {
      */
     public CuratorZookeeperClient getZookeeperClient();
 
+    /**
+     * Return zookeeper server compatibility
+     *
+     * @return compatibility
+     */
+    public ZookeeperCompatibility getZookeeperCompatibility();
+
     /**
      * Allocates an ensure path instance that is namespace aware
      *

curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java

@@ -46,6 +46,7 @@
 import org.apache.curator.framework.state.ConnectionStateListenerManagerFactory;
 import org.apache.curator.framework.state.StandardConnectionStateErrorPolicy;
 import org.apache.curator.utils.DefaultZookeeperFactory;
+import org.apache.curator.utils.ZookeeperCompatibility;
 import org.apache.curator.utils.ZookeeperFactory;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.Watcher;
@@ -174,6 +175,7 @@ public static class Builder {
                 ConnectionStateListenerManagerFactory.standard;
         private int simulatedSessionExpirationPercent = 100;
         private ZKClientConfig zkClientConfig;
+        private ZookeeperCompatibility zookeeperCompatibility = ZookeeperCompatibility.LATEST;
 
         /**
          * Apply the current values and build a new CuratorFramework
@@ -519,6 +521,11 @@ public Builder connectionStateListenerManagerFactory(
             return this;
         }
 
+        public Builder zookeeperCompatibility(ZookeeperCompatibility zookeeperCompatibility) {
+            this.zookeeperCompatibility = zookeeperCompatibility;
+            return this;
+        }
+
         public Executor getRunSafeService() {
             return runSafeService;
         }
@@ -640,6 +647,10 @@ public ConnectionStateListenerManagerFactory getConnectionStateListenerManagerFa
             return connectionStateListenerManagerFactory;
         }
 
+        public ZookeeperCompatibility getZookeeperCompatibility() {
+            return zookeeperCompatibility;
+        }
+
         private Builder() {}
     }
 

curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java

@@ -77,11 +77,11 @@
 import org.apache.curator.framework.state.ConnectionStateErrorPolicy;
 import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.framework.state.ConnectionStateManager;
-import org.apache.curator.utils.Compatibility;
 import org.apache.curator.utils.DebugUtils;
 import org.apache.curator.utils.EnsurePath;
 import org.apache.curator.utils.ThreadUtils;
 import org.apache.curator.utils.ZKPaths;
+import org.apache.curator.utils.ZookeeperCompatibility;
 import org.apache.curator.utils.ZookeeperFactory;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
@@ -117,6 +117,7 @@ public class CuratorFrameworkImpl implements CuratorFramework {
     private final EnsembleTracker ensembleTracker;
     private final SchemaSet schemaSet;
     private final Executor runSafeService;
+    private final ZookeeperCompatibility zookeeperCompatibility;
 
     private volatile ExecutorService executorService;
     private final AtomicBoolean logAsErrorConnectionErrors = new AtomicBoolean(false);
@@ -204,6 +205,7 @@ public void process(WatchedEvent watchedEvent) {
                 builder.withEnsembleTracker() ? new EnsembleTracker(this, builder.getEnsembleProvider()) : null;
 
         runSafeService = makeRunSafeService(builder);
+        zookeeperCompatibility = builder.getZookeeperCompatibility();
     }
 
     private Executor makeRunSafeService(CuratorFrameworkFactory.Builder builder) {
@@ -292,6 +294,7 @@ protected CuratorFrameworkImpl(CuratorFrameworkImpl parent) {
         schemaSet = parent.schemaSet;
         ensembleTracker = parent.ensembleTracker;
         runSafeService = parent.runSafeService;
+        zookeeperCompatibility = parent.zookeeperCompatibility;
     }
 
     @Override
@@ -585,8 +588,8 @@ public RemoveWatchesBuilder watches() {
     @Override
     public WatchesBuilder watchers() {
         Preconditions.checkState(
-                Compatibility.hasPersistentWatchers(),
-                "watchers() is not supported in the ZooKeeper library being used. Use watches() instead.");
+                zookeeperCompatibility.hasPersistentWatchers(),
+                "watchers() is not supported in the ZooKeeper library and/or server being used. Use watches() instead.");
         return new WatchesBuilderImpl(this);
     }
 
@@ -600,6 +603,11 @@ public CuratorZookeeperClient getZookeeperClient() {
         return client;
     }
 
+    @Override
+    public ZookeeperCompatibility getZookeeperCompatibility() {
+        return zookeeperCompatibility;
+    }
+
     @Override
     public EnsurePath newNamespaceAwareEnsurePath(String path) {
         return namespace.newNamespaceAwareEnsurePath(path);

curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/CuratorCacheBridgeBuilderImpl.java

@@ -21,7 +21,6 @@
 
 import java.util.concurrent.ExecutorService;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.utils.Compatibility;
 import org.slf4j.LoggerFactory;
 
 class CuratorCacheBridgeBuilderImpl implements CuratorCacheBridgeBuilder {
@@ -57,7 +56,7 @@ public CuratorCacheBridgeBuilder withExecutorService(ExecutorService executorSer
 
     @Override
     public CuratorCacheBridge build() {
-        if (!forceTreeCache && Compatibility.hasPersistentWatchers()) {
+        if (!forceTreeCache && client.getZookeeperCompatibility().hasPersistentWatchers()) {
             if (executorService != null) {
                 LoggerFactory.getLogger(getClass()).warn("CuratorCache does not support custom ExecutorService");
             }

curator-test-zk35/src/test/java/org/apache/curator/zk35/TestIs35.java

@@ -23,6 +23,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import org.apache.curator.test.compatibility.CuratorTestBase;
 import org.apache.curator.utils.Compatibility;
+import org.apache.curator.utils.ZookeeperCompatibility;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
@@ -33,6 +34,12 @@ public void testIsZk35() {
         assertFalse(Compatibility.hasGetReachableOrOneMethod());
         assertTrue(Compatibility.hasAddrField());
         assertFalse(Compatibility.hasPersistentWatchers());
+        assertFalse(ZookeeperCompatibility.LATEST.hasPersistentWatchers());
+        assertFalse(ZookeeperCompatibility.builder().build().hasPersistentWatchers());
+        assertFalse(ZookeeperCompatibility.builder()
+                .hasPersistentWatchers(false)
+                .build()
+                .hasPersistentWatchers());
     }
 
     @Override

curator-test-zk36/src/test/java/org/apache/curator/zk36/TestIs36.java

@@ -19,10 +19,12 @@
 
 package org.apache.curator.zk36;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import org.apache.curator.test.compatibility.CuratorTestBase;
 import org.apache.curator.utils.Compatibility;
+import org.apache.curator.utils.ZookeeperCompatibility;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
@@ -33,6 +35,12 @@ public void testIsZk36() {
         assertTrue(Compatibility.hasGetReachableOrOneMethod());
         assertTrue(Compatibility.hasAddrField());
         assertTrue(Compatibility.hasPersistentWatchers());
+        assertTrue(ZookeeperCompatibility.LATEST.hasPersistentWatchers());
+        assertFalse(ZookeeperCompatibility.builder().build().hasPersistentWatchers());
+        assertFalse(ZookeeperCompatibility.builder()
+                .hasPersistentWatchers(false)
+                .build()
+                .hasPersistentWatchers());
         try {
             Class.forName("org.apache.zookeeper.proto.WhoAmIResponse");
             fail("WhoAmIResponse is introduced after ZooKeeper 3.7");

curator-x-async/src/main/java/org/apache/curator/x/async/details/AsyncCuratorFrameworkImpl.java

@@ -32,7 +32,6 @@
 import org.apache.curator.framework.imps.CuratorMultiTransactionImpl;
 import org.apache.curator.framework.imps.GetACLBuilderImpl;
 import org.apache.curator.framework.imps.SyncBuilderImpl;
-import org.apache.curator.utils.Compatibility;
 import org.apache.curator.x.async.AsyncCuratorFramework;
 import org.apache.curator.x.async.AsyncStage;
 import org.apache.curator.x.async.WatchMode;
@@ -140,8 +139,8 @@ public AsyncRemoveWatchesBuilder removeWatches() {
     @Override
     public AsyncWatchBuilder addWatch() {
         Preconditions.checkState(
-                Compatibility.hasPersistentWatchers(),
-                "addWatch() is not supported in the ZooKeeper library being used.");
+                client.getZookeeperCompatibility().hasPersistentWatchers(),
+                "addWatch() is not supported in the ZooKeeper library and/or server being used.");
         return new AsyncWatchBuilderImpl(client, filters);
     }