Skip to content

Commit

Permalink
Basic concept of zk 3.4.x compatibility proven. The Compatibility cla…
Browse files Browse the repository at this point in the history
…ss checks for a well-known 3.5 class and sets a static

that advertises whether the ZK lib is 3.4.x or 3.5.x. Then, the code "ifs" using this static. The major work was emulating
the kill session injection (that emulation is done using reflection) and testing. The curator-test-zk module runs
the framework and recipe tests but forces ZooKeeper 3.4.x and uses the Curator 2.x version of curator-test. This
requires a few tricks as the new code uses new methods/classes on the Curator 3.x version of curator-test. I'll write
a readme documenting how this is done.
  • Loading branch information
randgalt committed Jul 20, 2017
1 parent 0641243 commit 58bc969
Show file tree
Hide file tree
Showing 42 changed files with 822 additions and 347 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
/**
* 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;

import org.apache.zookeeper.ZooKeeper;
import org.slf4j.LoggerFactory;

/**
* Utils to help with ZK 3.4.x compatibility
*/
public class Compatibility
{
private static final boolean hasZooKeeperAdmin;
static
{
boolean hasIt;
try
{
Class.forName("org.apache.zookeeper.admin.ZooKeeperAdmin");
hasIt = true;
}
catch ( ClassNotFoundException e )
{
hasIt = false;
LoggerFactory.getLogger(Compatibility.class).info("Running in ZooKeeper 3.4.x compatibility mode");
}
hasZooKeeperAdmin = hasIt;
}

/**
* Return true if the classpath ZooKeeper library is 3.4.x
*
* @return true/false
*/
public static boolean isZK34()
{
return !hasZooKeeperAdmin;
}

/**
* For ZooKeeper 3.5.x, use the supported <code>zooKeeper.getTestable().injectSessionExpiration()</code>.
* For ZooKeeper 3.4.x do the equivalent via reflection
*
* @param zooKeeper client
*/
public static void injectSessionExpiration(ZooKeeper zooKeeper)
{
if ( isZK34() )
{
InjectSessionExpiration.injectSessionExpiration(zooKeeper);
}
else
{
// LOL - this method was proposed by me (JZ) in 2013 for totally unrelated reasons
// it got added to ZK 3.5 and now does exactly what we need
// https://issues.apache.org/jira/browse/ZOOKEEPER-1730
zooKeeper.getTestable().injectSessionExpiration();
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
/**
* 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;

import org.apache.zookeeper.ClientCnxn;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import java.lang.reflect.Field;
import java.lang.reflect.Method;

// reflective version of zooKeeper.getTestable().injectSessionExpiration();
@SuppressWarnings("JavaReflectionMemberAccess")
public class InjectSessionExpiration
{
private static final Field cnxnField;
private static final Field stateField;
private static final Field eventThreadField;
private static final Field sendThreadField;
private static final Method queueEventMethod;
private static final Method queueEventOfDeathMethod;
private static final Method getClientCnxnSocketMethod;
private static final Method wakeupCnxnMethod;
static
{
Field localCnxnField;
Field localStateField;
Field localEventThreadField;
Field localSendThreadField;
Method localQueueEventMethod;
Method localEventOfDeathMethod;
Method localGetClientCnxnSocketMethod;
Method localWakeupCnxnMethod;
try
{
Class<?> eventThreadClass = Class.forName("org.apache.zookeeper.ClientCnxn$EventThread");
Class<?> sendThreadClass = Class.forName("org.apache.zookeeper.ClientCnxn$SendThread");
Class<?> clientCnxnSocketClass = Class.forName("org.apache.zookeeper.ClientCnxnSocket");

localCnxnField = ZooKeeper.class.getDeclaredField("cnxn");
localCnxnField.setAccessible(true);
localStateField = ClientCnxn.class.getDeclaredField("state");
localStateField.setAccessible(true);
localEventThreadField = ClientCnxn.class.getDeclaredField("eventThread");
localEventThreadField.setAccessible(true);
localSendThreadField = ClientCnxn.class.getDeclaredField("sendThread");
localSendThreadField.setAccessible(true);
localQueueEventMethod = eventThreadClass.getDeclaredMethod("queueEvent", WatchedEvent.class);
localQueueEventMethod.setAccessible(true);
localEventOfDeathMethod = eventThreadClass.getDeclaredMethod("queueEventOfDeath");
localEventOfDeathMethod.setAccessible(true);
localGetClientCnxnSocketMethod = sendThreadClass.getDeclaredMethod("getClientCnxnSocket");
localGetClientCnxnSocketMethod.setAccessible(true);
localWakeupCnxnMethod = clientCnxnSocketClass.getDeclaredMethod("wakeupCnxn");
localWakeupCnxnMethod.setAccessible(true);
}
catch ( ReflectiveOperationException e )
{
throw new RuntimeException("Could not access internal ZooKeeper fields", e);
}
cnxnField = localCnxnField;
stateField = localStateField;
eventThreadField = localEventThreadField;
sendThreadField = localSendThreadField;
queueEventMethod = localQueueEventMethod;
queueEventOfDeathMethod = localEventOfDeathMethod;
getClientCnxnSocketMethod = localGetClientCnxnSocketMethod;
wakeupCnxnMethod = localWakeupCnxnMethod;
}

public static void injectSessionExpiration(ZooKeeper zooKeeper)
{
try
{
WatchedEvent event = new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.Expired, null);

ClientCnxn clientCnxn = (ClientCnxn)cnxnField.get(zooKeeper);
Object eventThread = eventThreadField.get(clientCnxn);
queueEventMethod.invoke(eventThread, event);
queueEventOfDeathMethod.invoke(eventThread);
stateField.set(clientCnxn, ZooKeeper.States.CLOSED);
Object sendThread = sendThreadField.get(clientCnxn);
Object clientCnxnSocket = getClientCnxnSocketMethod.invoke(sendThread);
wakeupCnxnMethod.invoke(clientCnxnSocket);
}
catch ( ReflectiveOperationException e )
{
throw new RuntimeException("Could not inject session expiration using reflection", e);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
import org.apache.curator.ensemble.fixed.FixedEnsembleProvider;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.BaseClassForTests;
import org.apache.curator.test.KillSession;
import org.apache.curator.test.KillSession2;
import org.apache.curator.test.Timing;
import org.apache.curator.utils.ZookeeperFactory;
import org.apache.zookeeper.CreateMode;
Expand Down Expand Up @@ -100,7 +100,7 @@ public Object call() throws Exception
// ignore
}

KillSession.kill(client.getZooKeeper(), server.getConnectString());
KillSession2.kill(client.getZooKeeper());

Assert.assertTrue(timing.awaitLatch(latch));
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,8 @@

import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.BaseClassForTests;
import org.apache.curator.test.KillSession2;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.KillSession;
import org.apache.curator.test.Timing;
import org.testng.Assert;
import org.testng.annotations.Test;
Expand Down Expand Up @@ -58,7 +57,7 @@ public Void call() throws Exception
if ( firstTime.compareAndSet(true, false) )
{
Assert.assertNull(client.getZooKeeper().exists("/foo/bar", false));
KillSession.kill(client.getZooKeeper(), server.getConnectString());
KillSession2.kill(client.getZooKeeper());
client.getZooKeeper();
client.blockUntilConnectedOrTimedOut();
}
Expand Down Expand Up @@ -132,7 +131,7 @@ public Void call() throws Exception
if ( firstTime.compareAndSet(true, false) )
{
Assert.assertNull(client.getZooKeeper().exists("/foo/bar", false));
KillSession.kill(client.getZooKeeper(), server.getConnectString());
KillSession2.kill(client.getZooKeeper());
client.getZooKeeper();
client.blockUntilConnectedOrTimedOut();
}
Expand Down Expand Up @@ -197,7 +196,7 @@ public void testBasic() throws Exception
public Void call() throws Exception
{
Assert.assertNull(client.getZooKeeper().exists("/foo/bar", false));
KillSession.kill(client.getZooKeeper(), server.getConnectString());
KillSession2.kill(client.getZooKeeper());

timing.sleepABit();

Expand Down Expand Up @@ -259,7 +258,7 @@ public Object call() throws Exception
public Void call() throws Exception
{
Assert.assertNull(client.getZooKeeper().exists("/foo/bar", false));
KillSession.kill(client.getZooKeeper(), server.getConnectString());
KillSession2.kill(client.getZooKeeper());

client.getZooKeeper();
client.blockUntilConnectedOrTimedOut();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,14 +44,15 @@
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.LoggerFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static org.apache.curator.utils.Compatibility.isZK34;

/**
* Factory methods for creating framework-style clients
*/
Expand All @@ -68,33 +69,6 @@ public class CuratorFrameworkFactory
private static final long DEFAULT_INACTIVE_THRESHOLD_MS = (int)TimeUnit.MINUTES.toMillis(3);
private static final int DEFAULT_CLOSE_WAIT_MS = (int)TimeUnit.SECONDS.toMillis(1);

private static final boolean hasZooKeeperAdmin;
static
{
boolean hasIt;
try
{
Class.forName("org.apache.zookeeper.admin.ZooKeeperAdmin");
hasIt = true;
}
catch ( ClassNotFoundException e )
{
hasIt = false;
LoggerFactory.getLogger(CuratorFrameworkFactory.class).info("Running in ZooKeeper 3.4.x compatibility mode");
}
hasZooKeeperAdmin = hasIt;
}

/**
* Return true if the classpath ZooKeeper library is 3.4.x
*
* @return true/false
*/
public static boolean isZK34()
{
return !hasZooKeeperAdmin;
}

/**
* Return a new builder that builds a CuratorFramework
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.curator.framework;

import org.apache.curator.utils.Compatibility;
import org.apache.zookeeper.CreateMode;

public class SafeIsTtlMode
Expand All @@ -34,7 +35,7 @@ public boolean isTtl(CreateMode mode)

public static boolean isTtl(CreateMode mode)
{
return !CuratorFrameworkFactory.isZK34() && Internal.instance.isTtl(mode);
return !Compatibility.isZK34() && Internal.instance.isTtl(mode);
}

private SafeIsTtlMode()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,14 @@ public T forPath(String path, byte[] data) throws Exception
}

String fixedPath = client.fixForNamespace(path);
transaction.add(Op.create(fixedPath, data, acling.getAclList(path), createMode, ttl), OperationType.CREATE, path);
if ( client.isZk34CompatibilityMode() )
{
transaction.add(Op.create(fixedPath, data, acling.getAclList(path), createMode), OperationType.CREATE, path);
}
else
{
transaction.add(Op.create(fixedPath, data, acling.getAclList(path), createMode, ttl), OperationType.CREATE, path);
}
return context;
}
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.curator.framework.schema.Schema;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.OpResult;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.proto.CreateRequest;
Expand Down Expand Up @@ -135,7 +136,22 @@ public List<CuratorTransactionResult> forOperations(List<CuratorOp> operations)
if ( (curatorOp.get().getType() == ZooDefs.OpCode.create) || (curatorOp.get().getType() == ZooDefs.OpCode.createContainer) )
{
CreateRequest createRequest = (CreateRequest)curatorOp.get().toRequestRecord();
CreateMode createMode = CreateMode.fromFlag(createRequest.getFlags(), CreateMode.PERSISTENT);
CreateMode createMode;
if ( client.isZk34CompatibilityMode() )
{
try
{
createMode = CreateMode.fromFlag(createRequest.getFlags());
}
catch ( KeeperException.BadArgumentsException dummy )
{
createMode = CreateMode.PERSISTENT;
}
}
else
{
createMode = CreateMode.fromFlag(createRequest.getFlags(), CreateMode.PERSISTENT);
}
schema.validateCreate(createMode, createRequest.getPath(), createRequest.getData(), createRequest.getAcl());
}
else if ( (curatorOp.get().getType() == ZooDefs.OpCode.delete) || (curatorOp.get().getType() == ZooDefs.OpCode.deleteContainer) )
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import com.google.common.base.Preconditions;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.listen.ListenerContainer;
import org.apache.curator.utils.Compatibility;
import org.apache.curator.utils.ThreadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -306,10 +307,7 @@ private void checkSessionExpiration()
log.warn(String.format("Session timeout has elapsed while SUSPENDED. Injecting a session expiration. Elapsed ms: %d. Adjusted session timeout ms: %d", elapsedMs, useSessionTimeoutMs));
try
{
// LOL - this method was proposed by me (JZ) in 2013 for totally unrelated reasons
// it got added to ZK 3.5 and now does exactly what we need
// https://issues.apache.org/jira/browse/ZOOKEEPER-1730
client.getZookeeperClient().getZooKeeper().getTestable().injectSessionExpiration();
Compatibility.injectSessionExpiration(client.getZookeeperClient().getZooKeeper());
}
catch ( Exception e )
{
Expand Down
Loading

0 comments on commit 58bc969

Please sign in to comment.