Skip to content

Commit

Permalink
CURATOR-709. Add server compatibility check support
Browse files Browse the repository at this point in the history
Add new interface ZookeeperCompatibility to represent server
compatibility in addition to the existing Compatibility class (which
represents client compatibility).

Enhance CuratorFramework to accept ZookeeperCompatibility instance,
allowing user to specify which server version to target (default is
LATEST).
  • Loading branch information
laurentgo committed Mar 11, 2024
1 parent 972fffa commit 53d30e3
Show file tree
Hide file tree
Showing 7 changed files with 83 additions and 8 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;

import org.apache.zookeeper.server.quorum.QuorumPeer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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 interface ZookeeperCompatibility {
public enum Version implements ZookeeperCompatibility {
VERSION_3_5(false),
LATEST(true);

private final boolean hasPersistentWatchers;

private Version(boolean hasPersistentWatchers) {
this.hasPersistentWatchers = hasPersistentWatchers;
}

@Override
public boolean hasPersistentWatchers() {
return this.hasPersistentWatchers && Compatibility.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
*/
boolean hasPersistentWatchers();
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.apache.curator.framework;


import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import java.net.InetAddress;
Expand Down Expand Up @@ -46,6 +47,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;
Expand Down Expand Up @@ -174,6 +176,7 @@ public static class Builder {
ConnectionStateListenerManagerFactory.standard;
private int simulatedSessionExpirationPercent = 100;
private ZKClientConfig zkClientConfig;
private ZookeeperCompatibility zookeeperCompatibility = ZookeeperCompatibility.Version.LATEST;

/**
* Apply the current values and build a new CuratorFramework
Expand Down Expand Up @@ -519,6 +522,11 @@ public Builder connectionStateListenerManagerFactory(
return this;
}

public Builder zookeeperCompatibility(ZookeeperCompatibility zookeeperCompatibility) {
this.zookeeperCompatibility = zookeeperCompatibility;
return this;
}

public Executor getRunSafeService() {
return runSafeService;
}
Expand Down Expand Up @@ -640,6 +648,10 @@ public ConnectionStateListenerManagerFactory getConnectionStateListenerManagerFa
return connectionStateListenerManagerFactory;
}

public ZookeeperCompatibility getZookeeperCompatibility() {
return zookeeperCompatibility;
}

private Builder() {}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -292,6 +294,7 @@ protected CuratorFrameworkImpl(CuratorFrameworkImpl parent) {
schemaSet = parent.schemaSet;
ensembleTracker = parent.ensembleTracker;
runSafeService = parent.runSafeService;
zookeeperCompatibility = parent.zookeeperCompatibility;
}

@Override
Expand Down Expand Up @@ -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);
}

Expand All @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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");
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -41,6 +40,7 @@
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;


public class AsyncCuratorFrameworkImpl implements AsyncCuratorFramework {
private final CuratorFrameworkImpl client;
private final Filters filters;
Expand Down Expand Up @@ -140,8 +140,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);
}

Expand Down

0 comments on commit 53d30e3

Please sign in to comment.