Timeout all acknowledgements for now

Author: imotov

src/main/java/com/sonian/elasticsearch/zookeeper/discovery/ZooKeeperClusterState.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.zen.DiscoveryNodesProvider;
 
 import java.io.IOException;
@@ -83,7 +84,13 @@ public ZooKeeperClusterState(Settings settings, ZooKeeperEnvironment environment
      *
      * @throws InterruptedException
      */
-    public void publish(ClusterState state) throws ElasticSearchException, InterruptedException {
+    public void publish(ClusterState state, Discovery.AckListener ackListener) throws ElasticSearchException, InterruptedException {
+        // TODO: Add ack logic
+        publish(state/*, new AckClusterStatePublishResponseHandler(state.nodes().size() - 1, ackListener)*/ );
+        ackListener.onTimeout();
+    }
+
+    private void publish(ClusterState state/*, final ClusterStatePublishResponseHandler publishResponseHandler*/) throws ElasticSearchException, InterruptedException {
         publishingLock.lock();
         try {
             logger.trace("Publishing new cluster state version [{}]", state.version());

src/main/java/com/sonian/elasticsearch/zookeeper/discovery/ZooKeeperDiscovery.java

@@ -667,9 +667,9 @@ public ZooKeeperStatePublisher(Settings settings, ZooKeeperEnvironment environme
         @Override public void publish(ClusterState clusterState, AckListener ackListener) {
             try {
                 // ignore the ack. rely on zk to handle distribution.
-                zooKeeperClusterState.publish(clusterState);
+                zooKeeperClusterState.publish(clusterState, ackListener);
             } catch (InterruptedException ex) {
-                // Ignore
+                Thread.currentThread().interrupt();
             }
 
         }

src/test/java/com/sonian/elasticsearch/zookeeper/discovery/AbstractZooKeeperTests.java

@@ -23,13 +23,16 @@
 import com.sonian.elasticsearch.zookeeper.discovery.embedded.EmbeddedZooKeeperService;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.*;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.network.NetworkUtils;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.zen.DiscoveryNodesProvider;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.index.shard.ShardId;
@@ -251,4 +254,15 @@ protected String clusterStateVersion() {
         }
     }
 
+
+    protected static class NoOpAckListener implements Discovery.AckListener {
+        @Override
+        public void onNodeAck(DiscoveryNode node, @Nullable Throwable t) {
+        }
+
+        @Override
+        public void onTimeout() {
+        }
+    }
+
 }

src/test/java/com/sonian/elasticsearch/zookeeper/discovery/ZooKeeperClusterStateTests.java

@@ -60,7 +60,7 @@ public void testClusterStatePublishing() throws Exception {
 
         zkState.start();
 
-        zkState.publish(initialState);
+        zkState.publish(initialState, new NoOpAckListener());
 
         final CountDownLatch latch = new CountDownLatch(1);
 
@@ -79,7 +79,7 @@ public void onNewClusterState(ClusterState clusterState) {
                 .version(1235L)
                 .build();
 
-        zkState.publish(secondVersion);
+        zkState.publish(secondVersion, new NoOpAckListener());
 
         retrievedState = zkState.retrieve(null);
 
@@ -101,7 +101,7 @@ public void testClusterStatePublishingWithNewVersion() throws Exception {
 
         zkStateOld.start();
 
-        zkStateOld.publish(initialState);
+        zkStateOld.publish(initialState, new NoOpAckListener());
 
         zkStateOld.stop();
 
@@ -124,7 +124,7 @@ public void testClusterStatePublishingWithNewVersion() throws Exception {
         zkStateNew.syncClusterState();
 
         // Make sure that new start can be published now
-        zkStateNew.publish(initialState);
+        zkStateNew.publish(initialState, new NoOpAckListener());
 
         zkStateNew.stop();
 

src/test/java/com/sonian/elasticsearch/zookeeper/discovery/ZooKeeperDiscoveryTests.java

@@ -236,7 +236,7 @@ public void shutdownZooKeeper() {
         ClusterState initialState = testClusterState(routingTable, nodes);
         ZooKeeperClusterState zkStateOld = buildZooKeeperClusterState(nodes, "0.0.1");
         zkStateOld.start();
-        zkStateOld.publish(initialState);
+        zkStateOld.publish(initialState, new NoOpAckListener());
         zkStateOld.stop();
 
         // Create a client node