More consistent reset behavior for a zookeeper cluster.

lib-dempsyimpl/src/main/java/com/nokia/dempsy/mpcluster/zookeeper/ZookeeperSession.java

@@ -81,6 +81,14 @@ protected ZooKeeper makeZookeeperInstance(String connectString, int sessionTimeo
    {
       return new ZooKeeper(connectString, sessionTimeout, new ZkWatcher());
    }
+
+   /**
+    * This is defined here to be overridden in a test.
+    */
+   protected ZookeeperCluster makeZookeeperCluster(ClusterId clusterId) throws MpClusterException
+   {
+      return new ZookeeperCluster(clusterId);
+   }
 
    @Override
    public MpApplication<T, N> getApplication(String applicationId) throws MpClusterException
@@ -109,7 +117,7 @@ public MpCluster<T, N> getCluster(ClusterId clusterId) throws MpClusterException
          if (cluster == null)
          {
             // make sure the paths are set up
-            cluster = new ZookeeperCluster(clusterId);
+            cluster = makeZookeeperCluster(clusterId);
             initializeCluster(cluster,false);
             cachedClusters.put(clusterId, cluster);
          }
@@ -361,25 +369,32 @@ public void process(WatchedEvent event)
          // event = null means it was called explicitly
          if (logger.isDebugEnabled() && event != null)
             logger.debug("CALLBACK:MpContainerCluster for " + idForLogging + " Event:" + event);
-
+
+         boolean kickOffProcess = true;
          if (event != null)
          {
             if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged)
                clearState(event);
 
+            // when we're not connected we want to reset
             if (event.getState() != KeeperState.SyncConnected)
             {
+               kickOffProcess = false; // no reason to execute process if we're going to reset zookeeper.
+
                clearState(event);
 
                if (zkref != null)
                   resetZookeeper(zkref.get());
             }
          }
 
-         synchronized(processLock)
+         if (kickOffProcess)
          {
-            for(MpClusterWatcher watch: watchers)
-               watch.process();
+            synchronized(processLock)
+            {
+               for(MpClusterWatcher watch: watchers)
+                  watch.process();
+            }
          }
       }
    }
@@ -414,7 +429,7 @@ public class ZookeeperCluster extends WatcherManager implements MpCluster<T, N>
       private ZookeeperPath clusterPath;
       private ZookeeperPath appPath;
 
-      private ZookeeperCluster(ClusterId clusterId) throws MpClusterException
+      protected ZookeeperCluster(ClusterId clusterId) throws MpClusterException
       {
          super(clusterId.toString());
          this.clusterId = clusterId;