Notify listener, do not throw from async method (elastic#63)

Rethrowing a FailedToCommitClusterStateException is the wrong thing to do - it
should be passed to the listener for further processing.

server/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java

@@ -325,14 +325,15 @@ public void publish(ClusterChangedEvent clusterChangedEvent, ActionListener<Void
         ClusterState newState = clusterChangedEvent.state();
         assert newState.getNodes().isLocalNodeElectedMaster() : "Shouldn't publish state when not master " + clusterChangedEvent.source();
 
-        // state got changed locally (maybe because another master published to us)
-        if (clusterChangedEvent.previousState() != this.committedState.get()) {
-            throw new FailedToCommitClusterStateException("state was mutated while calculating new CS update");
-        }
+        try {
+
+            // state got changed locally (maybe because another master published to us)
+            if (clusterChangedEvent.previousState() != this.committedState.get()) {
+                throw new FailedToCommitClusterStateException("state was mutated while calculating new CS update");
+            }
 
-        pendingStatesQueue.addPending(newState);
+            pendingStatesQueue.addPending(newState);
 
-        try {
             publishClusterState.publish(clusterChangedEvent, electMaster.minimumMasterNodes(), ackListener);
         } catch (FailedToCommitClusterStateException t) {
             // cluster service logs a WARN message
@@ -345,7 +346,9 @@ public void publish(ClusterChangedEvent clusterChangedEvent, ActionListener<Void
 
                 rejoin("zen-disco-failed-to-publish");
             }
-            throw t;
+
+            publishListener.onFailure(t);
+            return;
         }
 
         final DiscoveryNode localNode = newState.getNodes().getLocalNode();