Skip to content

Commit

Permalink
Issue 245: Auth info specified in the CuratorFrameworkFactory.Builder…
Browse files Browse the repository at this point in the history 
… was not being re-set in cases

where the internal ZooKeeper handle was recreated. i.e. if the cluster has issues auth info would be lost.
  • Loading branch information
@Randgalt
Randgalt committed Feb 6, 2013
1 parent ceb4a7d commit 0482093
Show file tree
Hide file tree
Showing 3 changed files with 130 additions and 43 deletions.
3 changes: 3 additions & 0 deletions CHANGES.txt
View file
Expand Up @@ -7,6 +7,9 @@ thanks to user barkbay for his persistence and help on this.

* Issue 247: POST_INITIALIZED_EVENT wasn't correctly handling an initially empty node.

* Issue 245: Auth info specified in the CuratorFrameworkFactory.Builder was not being re-set in cases
where the internal ZooKeeper handle was recreated. i.e. if the cluster has issues auth info would be lost.

1.3.1 - January 28, 2013
========================
* Tightened up a possible race deep inside the connection management.
Expand Down
46 changes: 31 additions & 15 deletions curator-framework/src/main/java/com/netflix/curator/framework/imps/CuratorFrameworkImpl.java
View file
Expand Up @@ -34,6 +34,7 @@
import com.netflix.curator.utils.DebugUtils;
import com.netflix.curator.utils.EnsurePath;
import com.netflix.curator.utils.ThreadUtils;
import com.netflix.curator.utils.ZookeeperFactory;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
Expand Down Expand Up @@ -87,13 +88,23 @@ private AuthInfo(String scheme, byte[] auth)
this.scheme = scheme;
this.auth = auth;
}

@Override
public String toString()
{
return "AuthInfo{" +
"scheme='" + scheme + '\'' +
", auth=" + Arrays.toString(auth) +
'}';
}
}

public CuratorFrameworkImpl(CuratorFrameworkFactory.Builder builder)
{
ZookeeperFactory localZookeeperFactory = makeZookeeperFactory(builder.getZookeeperFactory());
this.client = new CuratorZookeeperClient
(
builder.getZookeeperFactory(),
localZookeeperFactory,
builder.getEnsembleProvider(),
builder.getSessionTimeoutMs(),
builder.getConnectionTimeoutMs(),
Expand Down Expand Up @@ -145,6 +156,25 @@ public void process(WatchedEvent watchedEvent)
namespaceFacadeCache = new NamespaceFacadeCache(this);
}

private ZookeeperFactory makeZookeeperFactory(final ZookeeperFactory actualZookeeperFactory)
{
return new ZookeeperFactory()
{
@Override
public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly) throws Exception
{
ZooKeeper zooKeeper = actualZookeeperFactory.newZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly);
AuthInfo auth = authInfo.get();
if ( auth != null )
{
zooKeeper.addAuthInfo(auth.scheme, auth.auth);
}

return zooKeeper;
}
};
}

private ThreadFactory getThreadFactory(CuratorFrameworkFactory.Builder builder)
{
ThreadFactory threadFactory = builder.getThreadFactory();
Expand Down Expand Up @@ -591,20 +621,6 @@ private<DATA_TYPE> void handleBackgroundOperationException(OperationAndData<DATA

private void backgroundOperationsLoop()
{
AuthInfo auth = authInfo.getAndSet(null);
if ( auth != null )
{
try
{
client.getZooKeeper().addAuthInfo(auth.scheme, auth.auth);
}
catch ( Exception e )
{
logError("addAuthInfo for background operation threw exception", e);
return;
}
}

while ( !Thread.interrupted() )
{
OperationAndData<?> operationAndData;
Expand Down
124 changes: 96 additions & 28 deletions curator-framework/src/test/java/com/netflix/curator/framework/imps/TestFramework.java
View file
Expand Up @@ -17,6 +17,7 @@
*/
package com.netflix.curator.framework.imps;

import com.google.common.collect.Lists;
import com.google.common.io.Closeables;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
Expand All @@ -27,13 +28,16 @@
import com.netflix.curator.framework.state.ConnectionState;
import com.netflix.curator.framework.state.ConnectionStateListener;
import com.netflix.curator.retry.RetryOneTime;
import com.netflix.curator.test.TestingServer;
import com.netflix.curator.test.Timing;
import com.netflix.curator.utils.EnsurePath;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;
import org.testng.Assert;
import org.testng.annotations.Test;
Expand Down Expand Up @@ -75,27 +79,6 @@ public void stateChanged(CuratorFramework client, ConnectionState newState)
Closeables.closeQuietly(client);
}
}

@Test
public void testIt() throws Exception
{
CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
client.start();

Watcher watcher = new Watcher()
{
@Override
public void process(WatchedEvent event)
{
System.out.println("yep");
}
};
client.checkExists().usingWatcher(watcher).forPath("/hey");

server.close();

Thread.sleep(10000);
}

@Test
public void testNamespaceWithWatcher() throws Exception
Expand Down Expand Up @@ -182,25 +165,49 @@ public void processResult(CuratorFramework client, CuratorEvent event) throws Ex
}
}

@Test(enabled = false) // temp disable - there's a bug in ZK 3.4.3 with this
@Test
public void testCreateACL() throws Exception
{
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
CuratorFramework client = builder.connectString(server.getConnectString()).authorization("digest", "me:pass".getBytes()).retryPolicy(new RetryOneTime(1)).build();
CuratorFramework client = builder
.connectString(server.getConnectString())
.authorization("digest", "me:pass".getBytes())
.retryPolicy(new RetryOneTime(1))
.build();
client.start();
try
{
client.create().withACL(ZooDefs.Ids.CREATOR_ALL_ACL).forPath("/test");

client.setData().forPath("/test", "test".getBytes());
ACL acl = new ACL(ZooDefs.Perms.WRITE, ZooDefs.Ids.AUTH_IDS);
List<ACL> aclList = Lists.newArrayList(acl);
client.create().withACL(aclList).forPath("/test", "test".getBytes());
client.close();

client = builder
.connectString(server.getConnectString())
.authorization("digest", "me:pass".getBytes())
.retryPolicy(new RetryOneTime(1))
.build();
client.start();
try
{
client.setData().forPath("/test", "test".getBytes());
}
catch ( KeeperException.NoAuthException e )
{
Assert.fail("Auth failed");
}
client.close();
client = builder.connectString(server.getConnectString()).authorization("digest", "someone:else".getBytes()).retryPolicy(new RetryOneTime(1)).build();

client = builder
.connectString(server.getConnectString())
.authorization("digest", "something:else".getBytes())
.retryPolicy(new RetryOneTime(1))
.build();
client.start();
try
{
client.setData().forPath("/test", "test".getBytes());
Assert.fail("Should be prohibited due to auth");
Assert.fail("Should have failed with auth exception");
}
catch ( KeeperException.NoAuthException e )
{
Expand All @@ -213,6 +220,67 @@ public void testCreateACL() throws Exception
}
}

@Test
public void testCreateACLWithReset() throws Exception
{
Timing timing = new Timing();
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
CuratorFramework client = builder
.connectString(server.getConnectString())
.sessionTimeoutMs(timing.session())
.connectionTimeoutMs(timing.connection())
.authorization("digest", "me:pass".getBytes())
.retryPolicy(new RetryOneTime(1))
.build();
client.start();
try
{
final CountDownLatch lostLatch = new CountDownLatch(1);
ConnectionStateListener listener = new ConnectionStateListener()
{
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState)
{
if ( newState == ConnectionState.LOST )
{
lostLatch.countDown();
}
}
};
client.getConnectionStateListenable().addListener(listener);

ACL acl = new ACL(ZooDefs.Perms.WRITE, ZooDefs.Ids.AUTH_IDS);
List<ACL> aclList = Lists.newArrayList(acl);
client.create().withACL(aclList).forPath("/test", "test".getBytes());

server.stop();
Assert.assertTrue(timing.awaitLatch(lostLatch));
try
{
client.checkExists().forPath("/");
Assert.fail("Connection should be down");
}
catch ( KeeperException.ConnectionLossException e )
{
// expected
}

server = new TestingServer(server.getPort(), server.getTempDirectory());
try
{
client.setData().forPath("/test", "test".getBytes());
}
catch ( KeeperException.NoAuthException e )
{
Assert.fail("Auth failed");
}
}
finally
{
Closeables.closeQuietly(client);
}
}

@Test
public void testCreateParents() throws Exception
{
Expand Down

0 comments on commit 0482093

Please sign in to comment.