Fix ZenDiscoveryUnitTests to use listener properly

A consequence of elastic#63 is that a FailedToCommitClusterStateException is no longer
thrown by ZenDiscovery#publish(), it is just passed to the listener, so tests
that relied on catching it can no longer work.

Author: DaveCTurner

server/src/test/java/org/elasticsearch/discovery/zen/ZenDiscoveryUnitTests.java

@@ -19,9 +19,8 @@
 
 package org.elasticsearch.discovery.zen;
 
-import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.replication.ClusterStateCreationUtils;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterModule;
@@ -44,6 +43,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.zen.PublishClusterStateActionTests.AssertingAckListener;
 import org.elasticsearch.index.shard.ShardId;
@@ -85,12 +85,12 @@
 import static org.elasticsearch.cluster.service.MasterServiceTests.discoveryState;
 import static org.elasticsearch.discovery.zen.ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES_SETTING;
 import static org.elasticsearch.discovery.zen.ZenDiscovery.shouldIgnoreOrRejectNewClusterState;
-import static org.elasticsearch.test.ClusterServiceUtils.setState;
-import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.emptyArray;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasToString;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 
 public class ZenDiscoveryUnitTests extends ESTestCase {
 
@@ -228,16 +228,19 @@ public void testNodesUpdatedAfterClusterStatePublished() throws Exception {
                 DiscoveryNodes.builder(state.nodes()).add(otherNode).masterNodeId(masterNode.getId())
             ).build();
 
-            try {
-                // publishing a new cluster state
-                ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent("testing", newState, state);
-                AssertingAckListener listener = new AssertingAckListener(newState.nodes().getSize() - 1);
-                expectedFDNodes = masterZen.getFaultDetectionNodes();
-                masterZen.publish(clusterChangedEvent, getNoopPublishListener(), listener);
+            // publishing a new cluster state
+            ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent("testing", newState, state);
+            AssertingAckListener listener = new AssertingAckListener(newState.nodes().getSize() - 1);
+            expectedFDNodes = masterZen.getFaultDetectionNodes();
+            AwaitingPublishListener awaitingPublishListener = new AwaitingPublishListener();
+            masterZen.publish(clusterChangedEvent, awaitingPublishListener, listener);
+            awaitingPublishListener.await();
+            if (awaitingPublishListener.getException() == null) {
+                // publication succeeded, wait for acks
                 listener.await(10, TimeUnit.SECONDS);
                 // publish was a success, update expected FD nodes based on new cluster state
                 expectedFDNodes = fdNodesForState(newState, masterNode);
-            } catch (Discovery.FailedToCommitClusterStateException e) {
+            } else {
                 // not successful, so expectedFDNodes above should remain what it was originally assigned
                 assertEquals(3, minMasterNodes); // ensure min master nodes is the higher value, otherwise we shouldn't fail
             }
@@ -281,35 +284,48 @@ public void testPendingCSQueueIsClearedWhenClusterStatePublished() throws Except
                 DiscoveryNodes.builder(discoveryState(masterMasterService).nodes()).masterNodeId(masterNode.getId())
             ).build();
 
-
-            try {
-                // publishing a new cluster state
-                ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent("testing", newState, state);
-                AssertingAckListener listener = new AssertingAckListener(newState.nodes().getSize() - 1);
-                masterZen.publish(clusterChangedEvent, getNoopPublishListener(), listener);
+            // publishing a new cluster state
+            ClusterChangedEvent clusterChangedEvent = new ClusterChangedEvent("testing", newState, state);
+            AssertingAckListener listener = new AssertingAckListener(newState.nodes().getSize() - 1);
+            AwaitingPublishListener awaitingPublishListener = new AwaitingPublishListener();
+            masterZen.publish(clusterChangedEvent, awaitingPublishListener, listener);
+            awaitingPublishListener.await();
+            if (awaitingPublishListener.getException() == null) {
+                // publication succeeded, wait for acks
                 listener.await(1, TimeUnit.HOURS);
-                // publish was a success, check that queue as cleared
-                assertThat(masterZen.pendingClusterStates(), emptyArray());
-            } catch (Discovery.FailedToCommitClusterStateException e) {
-                // not successful, so the pending queue should be cleaned
-                assertThat(Arrays.toString(masterZen.pendingClusterStates()), masterZen.pendingClusterStates(), arrayWithSize(0));
             }
+            // queue should be cleared whether successful or not
+            assertThat(Arrays.toString(masterZen.pendingClusterStates()), masterZen.pendingClusterStates(), emptyArray());
         } finally {
             IOUtils.close(toClose);
             terminate(threadPool);
         }
     }
 
-    private ActionListener<Void> getNoopPublishListener() {
-        return new ActionListener<Void>() {
-           @Override
-           public void onResponse(Void aVoid) {
-           }
+    private class AwaitingPublishListener implements  ActionListener<Void> {
+        private final CountDownLatch countDownLatch = new CountDownLatch(1);
+        private Discovery.FailedToCommitClusterStateException exception;
 
-           @Override
-           public void onFailure(Exception e) {
-           }
-       };
+        @Override
+        public void onResponse(Void aVoid) {
+            assertThat(countDownLatch.getCount(), is(1L));
+            countDownLatch.countDown();
+        }
+
+        @Override
+        public void onFailure(Exception e) {
+            assertThat(e, notNullValue());
+            exception = (Discovery.FailedToCommitClusterStateException) e; // fail on other exception types
+            onResponse(null);
+        }
+
+        public void await() throws InterruptedException {
+            countDownLatch.await();
+        }
+
+        public Discovery.FailedToCommitClusterStateException getException() {
+            return exception;
+        }
     }
 
     private ZenDiscovery buildZenDiscovery(Settings settings, TransportService service, MasterService masterService,