curator-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From randg...@apache.org
Subject [01/29] curator git commit: First pass at new (optional) definition of state LOST
Date Tue, 01 Sep 2015 13:02:25 GMT
Repository: curator
Updated Branches:
  refs/heads/CURATOR-3.0 667c97e7f -> d57aaeb77


First pass at new (optional) definition of state LOST


Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/344634ac
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/344634ac
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/344634ac

Branch: refs/heads/CURATOR-3.0
Commit: 344634ac6e34e61bc0cc7b41923a1df4089c7948
Parents: 7d97259
Author: randgalt <randgalt@apache.org>
Authored: Fri Aug 21 12:10:24 2015 -0500
Committer: randgalt <randgalt@apache.org>
Committed: Fri Aug 21 12:10:24 2015 -0500

----------------------------------------------------------------------
 .../framework/CuratorFrameworkFactory.java      | 19 +++++
 .../framework/api/UnhandledErrorListener.java   |  4 +-
 .../framework/imps/CuratorFrameworkImpl.java    | 10 ++-
 .../framework/state/ConnectionState.java        | 35 +++++++--
 .../framework/state/ConnectionStateManager.java | 75 +++++++++++++++-----
 5 files changed, 113 insertions(+), 30 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/344634ac/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
index dcb2ee6..6209b06 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
@@ -31,6 +31,7 @@ import org.apache.curator.framework.imps.CuratorFrameworkImpl;
 import org.apache.curator.framework.imps.CuratorTempFrameworkImpl;
 import org.apache.curator.framework.imps.DefaultACLProvider;
 import org.apache.curator.framework.imps.GzipCompressionProvider;
+import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.utils.DefaultZookeeperFactory;
 import org.apache.curator.utils.ZookeeperFactory;
 import org.apache.zookeeper.CreateMode;
@@ -116,6 +117,7 @@ public class CuratorFrameworkFactory
         private ACLProvider aclProvider = DEFAULT_ACL_PROVIDER;
         private boolean canBeReadOnly = false;
         private boolean useContainerParentsIfAvailable = true;
+        private boolean enableSessionExpiredState = false;
 
         /**
          * Apply the current values and build a new CuratorFramework
@@ -343,6 +345,18 @@ public class CuratorFrameworkFactory
             return this;
         }
 
+        /**
+         * Changes the meaning of {@link ConnectionState#LOST} from it's pre Curator 3.0.0
meaning
+         * to a true lost session state. See the {@link ConnectionState#LOST} doc for details.
+         *
+         * @return this
+         */
+        public Builder enableSessionExpiredState()
+        {
+            this.enableSessionExpiredState = true;
+            return this;
+        }
+
         public ACLProvider getAclProvider()
         {
             return aclProvider;
@@ -398,6 +412,11 @@ public class CuratorFrameworkFactory
             return useContainerParentsIfAvailable;
         }
 
+        public boolean getEnableSessionExpiredState()
+        {
+            return enableSessionExpiredState;
+        }
+
         @Deprecated
         public String getAuthScheme()
         {

http://git-wip-us.apache.org/repos/asf/curator/blob/344634ac/curator-framework/src/main/java/org/apache/curator/framework/api/UnhandledErrorListener.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/api/UnhandledErrorListener.java
b/curator-framework/src/main/java/org/apache/curator/framework/api/UnhandledErrorListener.java
index b463af2..3721d4b 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/api/UnhandledErrorListener.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/api/UnhandledErrorListener.java
@@ -24,9 +24,7 @@ import org.apache.curator.framework.state.ConnectionStateListener;
 public interface UnhandledErrorListener
 {
     /**
-     * Called when an exception is caught in a background thread, handler, etc. Before this
-     * listener is called, the error will have been logged and a {@link ConnectionState#LOST}
-     * event will have been queued for any {@link ConnectionStateListener}s.
+     * Called when an exception is caught in a background thread, handler, etc.
      *
      * @param message Source message
      * @param e exception

http://git-wip-us.apache.org/repos/asf/curator/blob/344634ac/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 41bb7cd..c64fb8f 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
@@ -83,6 +83,7 @@ public class CuratorFrameworkImpl implements CuratorFramework
     private final NamespaceFacadeCache namespaceFacadeCache;
     private final NamespaceWatcherMap namespaceWatcherMap = new NamespaceWatcherMap(this);
     private final boolean useContainerParentsIfAvailable;
+    private final boolean enableSessionExpiredState;
 
     private volatile ExecutorService executorService;
     private final AtomicBoolean logAsErrorConnectionErrors = new AtomicBoolean(false);
@@ -119,11 +120,12 @@ public class CuratorFrameworkImpl implements CuratorFramework
         namespace = new NamespaceImpl(this, builder.getNamespace());
         threadFactory = getThreadFactory(builder);
         maxCloseWaitMs = builder.getMaxCloseWaitMs();
-        connectionStateManager = new ConnectionStateManager(this, builder.getThreadFactory());
+        connectionStateManager = new ConnectionStateManager(this, builder.getThreadFactory(),
builder.getEnableSessionExpiredState(), builder.getSessionTimeoutMs());
         compressionProvider = builder.getCompressionProvider();
         aclProvider = builder.getAclProvider();
         state = new AtomicReference<CuratorFrameworkState>(CuratorFrameworkState.LATENT);
         useContainerParentsIfAvailable = builder.useContainerParentsIfAvailable();
+        enableSessionExpiredState = builder.getEnableSessionExpiredState();
 
         byte[] builderDefaultData = builder.getDefaultData();
         defaultData = (builderDefaultData != null) ? Arrays.copyOf(builderDefaultData, builderDefaultData.length)
: new byte[0];
@@ -197,6 +199,7 @@ public class CuratorFrameworkImpl implements CuratorFramework
         state = parent.state;
         authInfos = parent.authInfos;
         useContainerParentsIfAvailable = parent.useContainerParentsIfAvailable;
+        enableSessionExpiredState = parent.enableSessionExpiredState;
     }
 
     @Override
@@ -722,7 +725,10 @@ public class CuratorFrameworkImpl implements CuratorFramework
             return;
         }
 
-        doSyncForSuspendedConnection(client.getInstanceIndex());
+        if ( !enableSessionExpiredState )
+        {
+            doSyncForSuspendedConnection(client.getInstanceIndex());
+        }
     }
 
     private void doSyncForSuspendedConnection(final long instanceIndex)

http://git-wip-us.apache.org/repos/asf/curator/blob/344634ac/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionState.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionState.java
b/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionState.java
index 3ca1d66..49d0044 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionState.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionState.java
@@ -19,6 +19,8 @@
 package org.apache.curator.framework.state;
 
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
 
 /**
  * Represents state changes in the connection to ZK
@@ -39,8 +41,7 @@ public enum ConnectionState
 
     /**
      * There has been a loss of connection. Leaders, locks, etc. should suspend
-     * until the connection is re-established. If the connection times-out you will
-     * receive a {@link #LOST} notice
+     * until the connection is re-established.
      */
     SUSPENDED
     {
@@ -62,9 +63,29 @@ public enum ConnectionState
     },
 
     /**
-     * The connection is confirmed to be lost. Close any locks, leaders, etc. and
-     * attempt to re-create them. NOTE: it is possible to get a {@link #RECONNECTED}
-     * state after this but you should still consider any locks, etc. as dirty/unstable
+     * <p>
+     *     NOTE: the meaning of this state depends on how your CuratorFramework instance
+     *     is created.
+     * </p>
+     *
+     * <p>
+     *     The default meaning of LOST (and the only meaning prior to Curator 3.0.0) is:
+     *     The connection is confirmed to be lost (i.e. the retry policy has given up). Close
any locks, leaders, etc. and
+     *     attempt to re-create them. NOTE: it is possible to get a {@link #RECONNECTED}
+     *     state after this but you should still consider any locks, etc. as dirty/unstable
+     * </p>
+     *
+     * <p>
+     *     <strong>Since 3.0.0</strong>, you can alter the meaning of LOST by
calling
+     *     {@link CuratorFrameworkFactory.Builder#enableSessionExpiredState()}. In this mode,
+     *     Curator will set the LOST state only when it believes that the ZooKeeper session
+     *     has expired. ZooKeeper connections have a session. When the session expires, clients
must take appropriate
+     *     action. In Curator, this is complicated by the fact that Curator internally manages
the ZooKeeper
+     *     connection. In this mode, Curator will set the LOST state when any of the following
occurs:
+     *     a) ZooKeeper returns a {@link Watcher.Event.KeeperState#Expired} or {@link KeeperException.Code#SESSIONEXPIRED};
+     *     b) Curator closes the internally managed ZooKeeper instance; c) The configured
session timeout
+     *     elapses during a network partition.
+     * </p>
      */
     LOST
     {
@@ -87,7 +108,9 @@ public enum ConnectionState
         {
             return true;
         }
-    };
+    }
+
+    ;
     
     /**
      * Check if this state indicates that Curator has a connection to ZooKeeper

http://git-wip-us.apache.org/repos/asf/curator/blob/344634ac/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 67ff13d..c0feb84 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
@@ -35,6 +35,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -65,6 +66,8 @@ public class ConnectionStateManager implements Closeable
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final BlockingQueue<ConnectionState> eventQueue = new ArrayBlockingQueue<ConnectionState>(QUEUE_SIZE);
     private final CuratorFramework client;
+    private final boolean enableSessionExpiredState;
+    private final int sessionTimeoutMs;
     private final ListenerContainer<ConnectionStateListener> listeners = new ListenerContainer<ConnectionStateListener>();
     private final AtomicBoolean initialConnectMessageSent = new AtomicBoolean(false);
     private final ExecutorService service;
@@ -72,6 +75,8 @@ public class ConnectionStateManager implements Closeable
 
     // guarded by sync
     private ConnectionState currentConnectionState;
+    // guarded by sync
+    private long startOfSuspendedEpoch = 0;
 
     private enum State
     {
@@ -83,10 +88,14 @@ public class ConnectionStateManager implements Closeable
     /**
      * @param client        the client
      * @param threadFactory thread factory to use or null for a default
+     * @param enableSessionExpiredState if true, applies new meaning for LOST as described
here: {@link ConnectionState#LOST}
+     * @param sessionTimeoutMs the ZK session timeout in milliseconds
      */
-    public ConnectionStateManager(CuratorFramework client, ThreadFactory threadFactory)
+    public ConnectionStateManager(CuratorFramework client, ThreadFactory threadFactory, boolean
enableSessionExpiredState, int sessionTimeoutMs)
     {
         this.client = client;
+        this.enableSessionExpiredState = enableSessionExpiredState;
+        this.sessionTimeoutMs = sessionTimeoutMs;
         if ( threadFactory == null )
         {
             threadFactory = ThreadUtils.newThreadFactory("ConnectionStateManager");
@@ -137,7 +146,7 @@ public class ConnectionStateManager implements Closeable
 
     /**
      * Change to {@link ConnectionState#SUSPENDED} only if not already suspended and not
lost
-     * 
+     *
      * @return true if connection is set to SUSPENDED
      */
     public synchronized boolean setToSuspended()
@@ -152,7 +161,7 @@ public class ConnectionStateManager implements Closeable
             return false;
         }
 
-        currentConnectionState = ConnectionState.SUSPENDED;
+        setCurrentConnectionState(ConnectionState.SUSPENDED);
         postState(ConnectionState.SUSPENDED);
 
         return true;
@@ -177,7 +186,7 @@ public class ConnectionStateManager implements Closeable
         {
             return false;
         }
-        currentConnectionState = newConnectionState;
+        setCurrentConnectionState(newConnectionState);
 
         ConnectionState localState = newConnectionState;
         boolean isNegativeMessage = ((newConnectionState == ConnectionState.LOST) || (newConnectionState
== ConnectionState.SUSPENDED) || (newConnectionState == ConnectionState.READ_ONLY));
@@ -242,25 +251,34 @@ public class ConnectionStateManager implements Closeable
         {
             while ( !Thread.currentThread().isInterrupted() )
             {
-                final ConnectionState newState = eventQueue.take();
-
-                if ( listeners.size() == 0 )
+                final ConnectionState newState = eventQueue.poll(sessionTimeoutMs, TimeUnit.MILLISECONDS);
+                if ( newState != null )
                 {
-                    log.warn("There are no ConnectionStateListeners registered.");
-                }
+                    if ( listeners.size() == 0 )
+                    {
+                        log.warn("There are no ConnectionStateListeners registered.");
+                    }
 
-                listeners.forEach
-                    (
-                        new Function<ConnectionStateListener, Void>()
-                        {
-                            @Override
-                            public Void apply(ConnectionStateListener listener)
+                    listeners.forEach
+                        (
+                            new Function<ConnectionStateListener, Void>()
                             {
-                                listener.stateChanged(client, newState);
-                                return null;
+                                @Override
+                                public Void apply(ConnectionStateListener listener)
+                                {
+                                    listener.stateChanged(client, newState);
+                                    return null;
+                                }
                             }
-                        }
-                    );
+                        );
+                }
+                else if ( enableSessionExpiredState )
+                {
+                    synchronized(this)
+                    {
+                        checkSessionExpiration();
+                    }
+                }
             }
         }
         catch ( InterruptedException e )
@@ -268,4 +286,23 @@ public class ConnectionStateManager implements Closeable
             Thread.currentThread().interrupt();
         }
     }
+
+    private void checkSessionExpiration()
+    {
+        if ( (currentConnectionState == ConnectionState.SUSPENDED) && (startOfSuspendedEpoch
!= 0) )
+        {
+            long elapsedMs = System.currentTimeMillis() - startOfSuspendedEpoch;
+            if ( elapsedMs >= sessionTimeoutMs )
+            {
+                log.info(String.format("Session timeout has elapsed while SUSPENDED. Posting
LOST event. Elapsed ms: %d", elapsedMs));
+                addStateChange(ConnectionState.LOST);
+            }
+        }
+    }
+
+    private void setCurrentConnectionState(ConnectionState newConnectionState)
+    {
+        currentConnectionState = newConnectionState;
+        startOfSuspendedEpoch = (currentConnectionState == ConnectionState.SUSPENDED) ? System.currentTimeMillis()
: 0;
+    }
 }


Mime
View raw message