[2/2] incubator-twill git commit: Better state handling and more comments to explain some of the race conditions may arise during stopping of the ZKClient.

[ch...@apache.org]

Better state handling and more comments to explain some of the race conditions may arise during stopping of the ZKClient.

Project: http://git-wip-us.apache.org/repos/asf/incubator-twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-twill/commit/74720b80
Tree: http://git-wip-us.apache.org/repos/asf/incubator-twill/tree/74720b80
Diff: http://git-wip-us.apache.org/repos/asf/incubator-twill/diff/74720b80

Branch: refs/heads/feature/twill-133
Commit: 74720b804e5216e106b2d0f1e9cf1dbbb2d6282b
Parents: bda9908
Author: Terence Yim <ch...@apache.org>
Authored: Mon Jun 15 10:22:17 2015 -0700
Committer: Terence Yim <ch...@apache.org>
Committed: Mon Jun 15 10:22:17 2015 -0700

----------------------------------------------------------------------
 .../zookeeper/DefaultZKClientService.java       | 57 +++++++++++++++-----
 1 file changed, 45 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/74720b80/twill-zookeeper/src/main/java/org/apache/twill/internal/zookeeper/DefaultZKClientService.java
----------------------------------------------------------------------
diff --git a/twill-zookeeper/src/main/java/org/apache/twill/internal/zookeeper/DefaultZKClientService.java b/twill-zookeeper/src/main/java/org/apache/twill/internal/zookeeper/DefaultZKClientService.java
index 71a0118..6fae663 100644
--- a/twill-zookeeper/src/main/java/org/apache/twill/internal/zookeeper/DefaultZKClientService.java
+++ b/twill-zookeeper/src/main/java/org/apache/twill/internal/zookeeper/DefaultZKClientService.java
@@ -391,7 +391,13 @@ public final class DefaultZKClientService extends AbstractZKClient implements ZK
 
   private final class ServiceDelegate extends AbstractService implements Watcher {
 
+    private final Runnable stopTask;
+
     private ServiceDelegate() {
+      // Creates the stop task runnable in constructor so that if the stop() method is called from shutdown hook,
+      // it won't fail with class loading error due to failure to load inner class from the shutdown thread.
+      this.stopTask = createStopTask();
+
       // Add a listener for state changes so that we can terminate the service even it is in STARTING state upoon
       // stop is requested
       addListener(new Listener() {
@@ -420,6 +426,8 @@ public final class DefaultZKClientService extends AbstractZKClient implements ZK
 
         @Override
         public void failed(State from, Throwable failure) {
+          eventExecutor.shutdownNow();
+          // Close the ZK client if there is exception. It is needed because the stop task may not get executed
           closeZooKeeper(zooKeeper.getAndSet(null));
         }
       }, Threads.SAME_THREAD_EXECUTOR);
@@ -444,24 +452,21 @@ public final class DefaultZKClientService extends AbstractZKClient implements ZK
 
     @Override
     protected void doStop() {
-      eventExecutor.submit(new Runnable() {
-        @Override
-        public void run() {
-          try {
-            closeZooKeeper(zooKeeper.getAndSet(null));
-            notifyStopped();
-          } catch (Exception e) {
-            notifyFailed(e);
-          }
-        }
-      });
+      // Submit a task to the executor to make sure all pending events in the executor are fired before
+      // transiting this Service into STOPPED state
+      eventExecutor.submit(stopTask);
       eventExecutor.shutdown();
     }
 
     @Override
     public void process(WatchedEvent event) {
+      State state = state();
+      if (state == State.TERMINATED || state == State.FAILED) {
+        return;
+      }
+
       try {
-        if (event.getState() == Event.KeeperState.SyncConnected && state() == State.STARTING) {
+        if (event.getState() == Event.KeeperState.SyncConnected && state == State.STARTING) {
           LOG.debug("Connected to ZooKeeper: {}", zkStr);
           notifyStarted();
           return;
@@ -473,6 +478,7 @@ public final class DefaultZKClientService extends AbstractZKClient implements ZK
           eventExecutor.submit(new Runnable() {
             @Override
             public void run() {
+              // Only reconnect if the current state is running
               if (state() != State.RUNNING) {
                 return;
               }
@@ -494,6 +500,33 @@ public final class DefaultZKClientService extends AbstractZKClient implements ZK
       }
     }
 
+
+    /**
+     * Creates a {@link Runnable} task that will get executed in the event executor for transiting this
+     * Service into STOPPED state.
+     */
+    private Runnable createStopTask() {
+      return new Runnable() {
+        @Override
+        public void run() {
+          try {
+            // Close the ZK connection in this task will make sure if there is ZK connection created
+            // after doStop() was called but before this task has been executed is also closed.
+            // It is possible to happen when the following sequence happens:
+            //
+            // 1. session expired, hence the expired event is triggered
+            // 2. The reconnect task executed. With Service.state() == RUNNING, it creates a new ZK client
+            // 3. Service.stop() gets called, Service.state() changed to STOPPING
+            // 4. The new ZK client created from the reconnect thread update the zooKeeper with the new one
+            closeZooKeeper(zooKeeper.getAndSet(null));
+            notifyStopped();
+          } catch (Exception e) {
+            notifyFailed(e);
+          }
+        }
+      };
+    }
+
     /**
      * Creates a new ZooKeeper connection.
      */