You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@curator.apache.org by ra...@apache.org on 2014/02/05 19:41:56 UTC
[02/12] git commit: Continued re-work on the connection state
transitions involving background/async APIs.
Continued re-work on the connection state transitions involving background/async APIs.
ConnectionStateManager now uses synchronization. This shouldn't hurt performance but rationalizes state changes from foreground/background ops.
Background errors now go through same code as foreground errors. Transition to LOST is handled specifically instead of generally in logError().
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/75acb0d9
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/75acb0d9
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/75acb0d9
Branch: refs/heads/master
Commit: 75acb0d9222c3a54c9e15edff319acedcb26bbf5
Parents: a937dfa
Author: randgalt <ra...@apache.org>
Authored: Tue Dec 24 17:54:55 2013 -0500
Committer: randgalt <ra...@apache.org>
Committed: Tue Dec 24 17:54:55 2013 -0500
----------------------------------------------------------------------
.../java/org/apache/curator/HandleHolder.java | 2 +-
.../curator/framework/imps/Backgrounding.java | 5 +
.../framework/imps/CuratorFrameworkImpl.java | 128 +++++++++++------
.../framework/state/ConnectionStateManager.java | 44 +++++-
.../framework/client/TestBackgroundStates.java | 142 +++++++++++++++++++
5 files changed, 268 insertions(+), 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/75acb0d9/curator-client/src/main/java/org/apache/curator/HandleHolder.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/HandleHolder.java b/curator-client/src/main/java/org/apache/curator/HandleHolder.java
index 4922688..1f7cd91 100644
--- a/curator-client/src/main/java/org/apache/curator/HandleHolder.java
+++ b/curator-client/src/main/java/org/apache/curator/HandleHolder.java
@@ -52,7 +52,7 @@ class HandleHolder
ZooKeeper getZooKeeper() throws Exception
{
- return helper.getZooKeeper();
+ return (helper != null) ? helper.getZooKeeper() : null;
}
String getConnectionString()
http://git-wip-us.apache.org/repos/asf/curator/blob/75acb0d9/curator-framework/src/main/java/org/apache/curator/framework/imps/Backgrounding.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/Backgrounding.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/Backgrounding.java
index 6ae9151..262b2a8 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/Backgrounding.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/Backgrounding.java
@@ -21,6 +21,7 @@ package org.apache.curator.framework.imps;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.zookeeper.KeeperException;
import java.util.concurrent.Executor;
class Backgrounding
@@ -109,6 +110,10 @@ class Backgrounding
}
catch ( Exception e )
{
+ if ( e instanceof KeeperException )
+ {
+ client.validateConnection(client.codeToState(((KeeperException)e).code()));
+ }
client.logError("Background operation result handling threw exception", e);
}
}
http://git-wip-us.apache.org/repos/asf/curator/blob/75acb0d9/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
index 3aa1097..1b0ef3f 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java
@@ -512,24 +512,19 @@ public class CuratorFrameworkImpl implements CuratorFramework
log.error(reason, e);
}
- if ( e instanceof KeeperException.ConnectionLossException )
- {
- handleKeeperStateDisconnected();
- }
-
final String localReason = reason;
unhandledErrorListeners.forEach
- (
- new Function<UnhandledErrorListener, Void>()
- {
- @Override
- public Void apply(UnhandledErrorListener listener)
+ (
+ new Function<UnhandledErrorListener, Void>()
{
- listener.unhandledError(localReason, e);
- return null;
+ @Override
+ public Void apply(UnhandledErrorListener listener)
+ {
+ listener.unhandledError(localReason, e);
+ return null;
+ }
}
- }
- );
+ );
}
String unfixForNamespace(String path)
@@ -557,6 +552,73 @@ public class CuratorFrameworkImpl implements CuratorFramework
return namespaceWatcherMap;
}
+ void validateConnection(Watcher.Event.KeeperState state)
+ {
+ if ( state == Watcher.Event.KeeperState.Disconnected )
+ {
+ suspendConnection();
+ }
+ else if ( state == Watcher.Event.KeeperState.Expired )
+ {
+ connectionStateManager.addStateChange(ConnectionState.LOST);
+ }
+ else if ( state == Watcher.Event.KeeperState.SyncConnected )
+ {
+ connectionStateManager.addStateChange(ConnectionState.RECONNECTED);
+ }
+ else if ( state == Watcher.Event.KeeperState.ConnectedReadOnly )
+ {
+ connectionStateManager.addStateChange(ConnectionState.READ_ONLY);
+ }
+ }
+
+ Watcher.Event.KeeperState codeToState(KeeperException.Code code)
+ {
+ switch ( code )
+ {
+ case AUTHFAILED:
+ case NOAUTH:
+ {
+ return Watcher.Event.KeeperState.AuthFailed;
+ }
+
+ case CONNECTIONLOSS:
+ case OPERATIONTIMEOUT:
+ {
+ return Watcher.Event.KeeperState.Disconnected;
+ }
+
+ case SESSIONEXPIRED:
+ {
+ return Watcher.Event.KeeperState.Expired;
+ }
+
+ case OK:
+ case SESSIONMOVED:
+ {
+ return Watcher.Event.KeeperState.SyncConnected;
+ }
+ }
+ return Watcher.Event.KeeperState.fromInt(-1);
+ }
+
+ private void suspendConnection()
+ {
+ connectionStateManager.setToSuspended();
+
+ // we appear to have disconnected, force a new ZK event and see if we can connect to another server
+ BackgroundOperation<String> operation = new BackgroundSyncImpl(this, null);
+ OperationAndData.ErrorCallback<String> errorCallback = new OperationAndData.ErrorCallback<String>()
+ {
+ @Override
+ public void retriesExhausted(OperationAndData<String> operationAndData)
+ {
+ connectionStateManager.addStateChange(ConnectionState.LOST);
+ }
+ };
+ performBackgroundOperation(new OperationAndData<String>(operation, "/", null, errorCallback, null));
+ }
+
@SuppressWarnings({"ThrowableResultOfMethodCallIgnored"})
private <DATA_TYPE> boolean checkBackgroundRetry(OperationAndData<DATA_TYPE> operationAndData, CuratorEvent event)
{
@@ -588,8 +650,10 @@ public class CuratorFrameworkImpl implements CuratorFramework
}
if ( e == null )
{
- e = new Exception("Unknown result code: " + event.getResultCode());
+ e = new Exception("Unknown result codegetResultCode()");
}
+
+ validateConnection(codeToState(code));
logError("Background operation retry gave up", e);
}
return doRetry;
@@ -714,7 +778,10 @@ public class CuratorFrameworkImpl implements CuratorFramework
private void processEvent(final CuratorEvent curatorEvent)
{
- validateConnection(curatorEvent);
+ if ( curatorEvent.getType() == CuratorEventType.WATCHED )
+ {
+ validateConnection(curatorEvent.getWatchedEvent().getState());
+ }
listeners.forEach
(
@@ -738,33 +805,4 @@ public class CuratorFrameworkImpl implements CuratorFramework
}
);
}
-
- private void validateConnection(CuratorEvent curatorEvent)
- {
- if ( curatorEvent.getType() == CuratorEventType.WATCHED )
- {
- if ( curatorEvent.getWatchedEvent().getState() == Watcher.Event.KeeperState.Disconnected )
- {
- handleKeeperStateDisconnected();
- }
- else if ( curatorEvent.getWatchedEvent().getState() == Watcher.Event.KeeperState.Expired )
- {
- connectionStateManager.addStateChange(ConnectionState.LOST);
- }
- else if ( curatorEvent.getWatchedEvent().getState() == Watcher.Event.KeeperState.SyncConnected )
- {
- connectionStateManager.addStateChange(ConnectionState.RECONNECTED);
- }
- else if ( curatorEvent.getWatchedEvent().getState() == Watcher.Event.KeeperState.ConnectedReadOnly )
- {
- connectionStateManager.addStateChange(ConnectionState.READ_ONLY);
- }
- }
- }
-
- private void handleKeeperStateDisconnected()
- {
- connectionStateManager.addStateChange(ConnectionState.SUSPENDED);
- internalSync(this, "/", null); // we appear to have disconnected, force a new ZK event and see if we can connect to another server
- }
}
http://git-wip-us.apache.org/repos/asf/curator/blob/75acb0d9/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionStateManager.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionStateManager.java b/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionStateManager.java
index fe5f18a..a2cfa60 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionStateManager.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionStateManager.java
@@ -65,11 +65,13 @@ public class ConnectionStateManager implements Closeable
private final BlockingQueue<ConnectionState> eventQueue = new ArrayBlockingQueue<ConnectionState>(QUEUE_SIZE);
private final CuratorFramework client;
private final ListenerContainer<ConnectionStateListener> listeners = new ListenerContainer<ConnectionStateListener>();
- private final AtomicReference<ConnectionState> currentState = new AtomicReference<ConnectionState>();
private final AtomicBoolean initialConnectMessageSent = new AtomicBoolean(false);
private final ExecutorService service;
private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
+ // guarded by sync
+ private ConnectionState currentConnectionState;
+
private enum State
{
LATENT,
@@ -133,23 +135,44 @@ public class ConnectionStateManager implements Closeable
}
/**
+ * Change to {@link ConnectionState#SUSPENDED} only if not already suspended and not lost
+ */
+ public synchronized void setToSuspended()
+ {
+ if ( state.get() != State.STARTED )
+ {
+ return;
+ }
+
+ if ( (currentConnectionState == ConnectionState.LOST) || (currentConnectionState == ConnectionState.SUSPENDED) )
+ {
+ return;
+ }
+
+ currentConnectionState = ConnectionState.SUSPENDED;
+ postState(ConnectionState.SUSPENDED);
+ }
+
+ /**
* Post a state change. If the manager is already in that state the change
* is ignored. Otherwise the change is queued for listeners.
*
* @param newConnectionState new state
+ * @return true if the state actually changed, false if it was already at that state
*/
- public void addStateChange(ConnectionState newConnectionState)
+ public synchronized boolean addStateChange(ConnectionState newConnectionState)
{
if ( state.get() != State.STARTED )
{
- return;
+ return false;
}
- ConnectionState previousState = currentState.getAndSet(newConnectionState);
+ ConnectionState previousState = currentConnectionState;
if ( previousState == newConnectionState )
{
- return;
+ return false;
}
+ currentConnectionState = newConnectionState;
ConnectionState localState = newConnectionState;
boolean isNegativeMessage = ((newConnectionState == ConnectionState.LOST) || (newConnectionState == ConnectionState.SUSPENDED));
@@ -158,8 +181,15 @@ public class ConnectionStateManager implements Closeable
localState = ConnectionState.CONNECTED;
}
- log.info("State change: " + localState);
- while ( !eventQueue.offer(localState) )
+ postState(localState);
+
+ return true;
+ }
+
+ private void postState(ConnectionState state)
+ {
+ log.info("State change: " + state);
+ while ( !eventQueue.offer(state) )
{
eventQueue.poll();
log.warn("ConnectionStateManager queue full - dropping events to make room");
http://git-wip-us.apache.org/repos/asf/curator/blob/75acb0d9/curator-recipes/src/test/java/org/apache/curator/framework/client/TestBackgroundStates.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/client/TestBackgroundStates.java b/curator-recipes/src/test/java/org/apache/curator/framework/client/TestBackgroundStates.java
new file mode 100644
index 0000000..b1c382f
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/client/TestBackgroundStates.java
@@ -0,0 +1,142 @@
+/**
+ * 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.framework.client;
+
+import com.google.common.collect.Queues;
+import com.google.common.io.Closeables;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.recipes.BaseClassForTests;
+import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.TestingServer;
+import org.apache.curator.test.Timing;
+import org.apache.curator.utils.DebugUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+// NOTE: these tests are in Framework as they use the PersistentEphemeralNode recipe
+
+public class TestBackgroundStates extends BaseClassForTests
+{
+ @Test
+ public void testListenersReconnectedIsOK() throws Exception
+ {
+ server.close();
+
+ Timing timing = new Timing();
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ try
+ {
+ client.start();
+ PersistentEphemeralNode node = new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, "/abc/node", "hello".getBytes());
+ node.start();
+
+ final CountDownLatch connectedLatch = new CountDownLatch(1);
+ final CountDownLatch reconnectedLatch = new CountDownLatch(1);
+ final AtomicReference<ConnectionState> lastState = new AtomicReference<ConnectionState>();
+ ConnectionStateListener listener = new ConnectionStateListener()
+ {
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState)
+ {
+ lastState.set(newState);
+ if ( newState == ConnectionState.CONNECTED )
+ {
+ connectedLatch.countDown();
+ }
+ if ( newState == ConnectionState.RECONNECTED )
+ {
+ reconnectedLatch.countDown();
+ }
+ }
+ };
+ client.getConnectionStateListenable().addListener(listener);
+ timing.sleepABit();
+ server = new TestingServer(server.getPort());
+ Assert.assertTrue(timing.awaitLatch(connectedLatch));
+ timing.sleepABit();
+ Assert.assertTrue(node.waitForInitialCreate(timing.forWaiting().milliseconds(), TimeUnit.MILLISECONDS));
+ server.close();
+ timing.sleepABit();
+ server = new TestingServer(server.getPort());
+ timing.sleepABit();
+ Assert.assertTrue(timing.awaitLatch(reconnectedLatch));
+ timing.sleepABit();
+ Assert.assertEquals(lastState.get(), ConnectionState.RECONNECTED);
+ }
+ finally
+ {
+ Closeables.closeQuietly(client);
+ }
+ }
+
+ @Test
+ public void testConnectionStateListener() throws Exception
+ {
+ System.setProperty(DebugUtils.PROPERTY_LOG_EVENTS, "true");
+
+ server.close();
+
+ Timing timing = new Timing();
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(timing.milliseconds()));
+ try
+ {
+ client.start();
+ PersistentEphemeralNode node = new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, "/abc/node", "hello".getBytes());
+ node.start();
+
+ final BlockingQueue<ConnectionState> stateVector = Queues.newLinkedBlockingQueue(1);
+ ConnectionStateListener listener = new ConnectionStateListener()
+ {
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState)
+ {
+ stateVector.offer(newState);
+ }
+ };
+
+ Timing waitingTiming = timing.forWaiting();
+
+ client.getConnectionStateListenable().addListener(listener);
+ server = new TestingServer(server.getPort());
+ Assert.assertEquals(stateVector.poll(waitingTiming.milliseconds(), TimeUnit.MILLISECONDS), ConnectionState.CONNECTED);
+ server.close();
+ Assert.assertEquals(stateVector.poll(waitingTiming.milliseconds(), TimeUnit.MILLISECONDS), ConnectionState.SUSPENDED);
+ Assert.assertEquals(stateVector.poll(waitingTiming.milliseconds(), TimeUnit.MILLISECONDS), ConnectionState.LOST);
+ server = new TestingServer(server.getPort());
+ Assert.assertEquals(stateVector.poll(waitingTiming.milliseconds(), TimeUnit.MILLISECONDS), ConnectionState.RECONNECTED);
+ server.close();
+ Assert.assertEquals(stateVector.poll(waitingTiming.milliseconds(), TimeUnit.MILLISECONDS), ConnectionState.SUSPENDED);
+ Assert.assertEquals(stateVector.poll(waitingTiming.milliseconds(), TimeUnit.MILLISECONDS), ConnectionState.LOST);
+ }
+ finally
+ {
+ Closeables.closeQuietly(client);
+ }
+ }
+
+}
\ No newline at end of file