The true session timeout is a negotiated value between client and server. The new...
authorrandgalt <randgalt@apache.org>
Mon, 24 Aug 2015 01:31:21 +0000 (20:31 -0500)
committerrandgalt <randgalt@apache.org>
Mon, 24 Aug 2015 01:31:21 +0000 (20:31 -0500)
curator-client/src/main/java/org/apache/curator/ConnectionState.java
curator-client/src/main/java/org/apache/curator/CuratorZookeeperClient.java
curator-client/src/main/java/org/apache/curator/HandleHolder.java
curator-framework/src/main/java/org/apache/curator/framework/state/ConnectionStateManager.java

index d6ddd33..555a52d 100644 (file)
@@ -36,6 +36,7 @@ import java.util.Queue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -46,6 +47,7 @@ class ConnectionState implements Watcher, Closeable
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final HandleHolder zooKeeper;
     private final AtomicBoolean isConnected = new AtomicBoolean(false);
+    private final AtomicInteger lastNegotiatedSessionTimeoutMs = new AtomicInteger(0);
     private final EnsembleProvider ensembleProvider;
     private final int sessionTimeoutMs;
     private final int connectionTimeoutMs;
@@ -141,6 +143,11 @@ class ConnectionState implements Watcher, Closeable
         return instanceIndex.get();
     }
 
+    int getLastNegotiatedSessionTimeoutMs()
+    {
+        return lastNegotiatedSessionTimeoutMs.get();
+    }
+
     @Override
     public void process(WatchedEvent event)
     {
@@ -167,6 +174,11 @@ class ConnectionState implements Watcher, Closeable
         {
             isConnected.set(newIsConnected);
             connectionStartMs = System.currentTimeMillis();
+            if ( newIsConnected )
+            {
+                lastNegotiatedSessionTimeoutMs.set(zooKeeper.getNegotiatedSessionTimeoutMs());
+                log.debug("Negotiated session timeout: " + lastNegotiatedSessionTimeoutMs.get());
+            }
         }
     }
 
index d6c2072..471adf0 100644 (file)
@@ -334,6 +334,16 @@ public class CuratorZookeeperClient implements Closeable
         return connectionHandlingPolicy;
     }
 
+    /**
+     * Return the most recent value of {@link ZooKeeper#getSessionTimeout()} or 0
+     *
+     * @return session timeout or 0
+     */
+    public int getLastNegotiatedSessionTimeoutMs()
+    {
+        return state.getLastNegotiatedSessionTimeoutMs();
+    }
+
     void addParentWatcher(Watcher watcher)
     {
         state.addParentWatcher(watcher);
index 1f7cd91..8652f0c 100644 (file)
@@ -39,6 +39,8 @@ class HandleHolder
         ZooKeeper getZooKeeper() throws Exception;
         
         String getConnectionString();
+
+        int getNegotiatedSessionTimeoutMs();
     }
 
     HandleHolder(ZookeeperFactory zookeeperFactory, Watcher watcher, EnsembleProvider ensembleProvider, int sessionTimeout, boolean canBeReadOnly)
@@ -55,6 +57,11 @@ class HandleHolder
         return (helper != null) ? helper.getZooKeeper() : null;
     }
 
+    int getNegotiatedSessionTimeoutMs()
+    {
+        return (helper != null) ? helper.getNegotiatedSessionTimeoutMs() : 0;
+    }
+
     String  getConnectionString()
     {
         return (helper != null) ? helper.getConnectionString() : null;
@@ -107,6 +114,12 @@ class HandleHolder
                         {
                             return connectionString;
                         }
+
+                        @Override
+                        public int getNegotiatedSessionTimeoutMs()
+                        {
+                            return (zooKeeperHandle != null) ? zooKeeperHandle.getSessionTimeout() : 0;
+                        }
                     };
 
                     return zooKeeperHandle;
@@ -118,6 +131,12 @@ class HandleHolder
             {
                 return connectionString;
             }
+
+            @Override
+            public int getNegotiatedSessionTimeoutMs()
+            {
+                return (zooKeeperHandle != null) ? zooKeeperHandle.getSessionTimeout() : 0;
+            }
         };
     }
 
index 44d511b..3da7534 100644 (file)
@@ -292,9 +292,10 @@ public class ConnectionStateManager implements Closeable
         if ( (currentConnectionState == ConnectionState.SUSPENDED) && (startOfSuspendedEpoch != 0) )
         {
             long elapsedMs = System.currentTimeMillis() - startOfSuspendedEpoch;
-            if ( elapsedMs >= sessionTimeoutMs )
+            int useSessionTimeoutMs = Math.max(client.getZookeeperClient().getLastNegotiatedSessionTimeoutMs(), sessionTimeoutMs);
+            if ( elapsedMs >= useSessionTimeoutMs )
             {
-                log.warn(String.format("Session timeout has elapsed while SUSPENDED. Posting LOST event and resetting the connection. Elapsed ms: %d", elapsedMs));
+                log.warn(String.format("Session timeout has elapsed while SUSPENDED. Posting LOST event and resetting the connection. Elapsed ms: %d. Session Timeout ms: %d", elapsedMs, useSessionTimeoutMs));
                 try
                 {
                     client.getZookeeperClient().reset();