Added generic cluster state update ack mechanism

src/main/java/org/elasticsearch/action/admin/indices/delete/TransportDeleteIndexAction.java

@@ -31,11 +31,10 @@
 import org.elasticsearch.cluster.metadata.MetaDataDeleteIndexService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.util.concurrent.atomic.AtomicInteger;
-
 /**
  * Delete index action.
  */
@@ -102,9 +101,10 @@ protected ClusterBlockException checkBlock(DeleteIndexRequest request, ClusterSt
     protected void masterOperation(final DeleteIndexRequest request, final ClusterState state, final ActionListener<DeleteIndexResponse> listener) throws ElasticSearchException {
         if (request.indices().length == 0) {
             listener.onResponse(new DeleteIndexResponse(true));
+            return;
         }
         // TODO: this API should be improved, currently, if one delete index failed, we send a failure, we should send a response array that includes all the indices that were deleted
-        final AtomicInteger count = new AtomicInteger(request.indices().length);
+        final CountDown count = new CountDown(request.indices().length);
         for (final String index : request.indices()) {
             deleteIndexService.deleteIndex(new MetaDataDeleteIndexService.Request(index).timeout(request.timeout()).masterTimeout(request.masterNodeTimeout()), new MetaDataDeleteIndexService.Listener() {
 
@@ -116,7 +116,7 @@ public void onResponse(MetaDataDeleteIndexService.Response response) {
                     if (!response.acknowledged()) {
                         ack = false;
                     }
-                    if (count.decrementAndGet() == 0) {
+                    if (count.countDown()) {
                         if (lastFailure != null) {
                             listener.onFailure(lastFailure);
                         } else {
@@ -129,7 +129,7 @@ public void onResponse(MetaDataDeleteIndexService.Response response) {
                 public void onFailure(Throwable t) {
                     logger.debug("[{}] failed to delete index", t, index);
                     lastFailure = t;
-                    if (count.decrementAndGet() == 0) {
+                    if (count.countDown()) {
                         listener.onFailure(t);
                     }
                 }

src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequest.java

@@ -19,24 +19,33 @@
 
 package org.elasticsearch.action.admin.indices.warmer.delete;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.unit.TimeValue;
 
 import java.io.IOException;
 
+import static org.elasticsearch.common.unit.TimeValue.readTimeValue;
+import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
+
 /**
  * A request to delete an index warmer.
  */
-public class DeleteWarmerRequest extends MasterNodeOperationRequest<DeleteWarmerRequest> {
+public class DeleteWarmerRequest extends MasterNodeOperationRequest<DeleteWarmerRequest>
+        implements AcknowledgedRequest<DeleteWarmerRequest> {
 
     private String name;
 
     private String[] indices = Strings.EMPTY_ARRAY;
 
+    private TimeValue timeout = timeValueSeconds(10);
+
     DeleteWarmerRequest() {
     }
 
@@ -87,17 +96,40 @@ public String[] indices() {
         return indices;
     }
 
+    @Override
+    public DeleteWarmerRequest timeout(String timeout) {
+        this.timeout = TimeValue.parseTimeValue(timeout, this.timeout);
+        return this;
+    }
+
+    @Override
+    public DeleteWarmerRequest timeout(TimeValue timeout) {
+        this.timeout = timeout;
+        return this;
+    }
+
+    @Override
+    public TimeValue timeout() {
+        return timeout;
+    }
+
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         name = in.readOptionalString();
         indices = in.readStringArray();
+        if (in.getVersion().onOrAfter(Version.V_0_90_6)) {
+            timeout = readTimeValue(in);
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeOptionalString(name);
         out.writeStringArrayNullable(indices);
+        if (out.getVersion().onOrAfter(Version.V_0_90_6)) {
+            timeout.writeTo(out);
+        }
     }
 }

src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerRequestBuilder.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.support.master.MasterNodeOperationRequestBuilder;
 import org.elasticsearch.client.IndicesAdminClient;
 import org.elasticsearch.client.internal.InternalIndicesAdminClient;
+import org.elasticsearch.common.unit.TimeValue;
 
 /**
  *
@@ -47,6 +48,23 @@ public DeleteWarmerRequestBuilder setName(String name) {
         return this;
     }
 
+    /**
+     * Sets the maximum wait for acknowledgement from other nodes
+     */
+    public DeleteWarmerRequestBuilder setTimeout(TimeValue timeout) {
+        request.timeout(timeout);
+        return this;
+    }
+
+    /**
+     * Timeout to wait for the operation to be acknowledged by current cluster nodes. Defaults
+     * to <tt>10s</tt>.
+     */
+    public DeleteWarmerRequestBuilder setTimeout(String timeout) {
+        request.timeout(timeout);
+        return this;
+    }
+
     @Override
     protected void doExecute(ActionListener<DeleteWarmerResponse> listener) {
         ((IndicesAdminClient) client).deleteWarmer(request, listener);

src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/DeleteWarmerResponse.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.admin.indices.warmer.delete;
 
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 
@@ -28,7 +29,7 @@
 /**
  * A response for a delete warmer.
  */
-public class DeleteWarmerResponse extends ActionResponse {
+public class DeleteWarmerResponse extends ActionResponse implements AcknowledgedResponse {
 
     private boolean acknowledged;
 

src/main/java/org/elasticsearch/action/admin/indices/warmer/delete/TransportDeleteWarmerAction.java

@@ -23,13 +23,15 @@
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.master.TransportMasterNodeOperationAction;
+import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.TimeoutClusterStateUpdateTask;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
@@ -88,7 +90,27 @@ protected ClusterBlockException checkBlock(DeleteWarmerRequest request, ClusterS
 
     @Override
     protected void masterOperation(final DeleteWarmerRequest request, final ClusterState state, final ActionListener<DeleteWarmerResponse> listener) throws ElasticSearchException {
-        clusterService.submitStateUpdateTask("delete_warmer [" + request.name() + "]", new TimeoutClusterStateUpdateTask() {
+        clusterService.submitStateUpdateTask("delete_warmer [" + request.name() + "]", new AckedClusterStateUpdateTask() {
+
+            @Override
+            public boolean mustAck(DiscoveryNode discoveryNode) {
+                return true;
+            }
+
+            @Override
+            public void onAllNodesAcked(@Nullable Throwable t) {
+                listener.onResponse(new DeleteWarmerResponse(true));
+            }
+
+            @Override
+            public void onAckTimeout() {
+                listener.onResponse(new DeleteWarmerResponse(false));
+            }
+
+            @Override
+            public TimeValue ackTimeout() {
+                return request.timeout();
+            }
 
             @Override
             public TimeValue timeout() {
@@ -161,7 +183,7 @@ public ClusterState execute(ClusterState currentState) {
 
             @Override
             public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
-                listener.onResponse(new DeleteWarmerResponse(true));
+
             }
         });
     }

src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequest.java

@@ -19,27 +19,35 @@
 
 package org.elasticsearch.action.admin.indices.warmer.put;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.unit.TimeValue;
 
 import java.io.IOException;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
+import static org.elasticsearch.common.unit.TimeValue.readTimeValue;
+import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
 
 /**
  * A request to put a search warmer.
  */
-public class PutWarmerRequest extends MasterNodeOperationRequest<PutWarmerRequest> {
+public class PutWarmerRequest extends MasterNodeOperationRequest<PutWarmerRequest>
+        implements AcknowledgedRequest<PutWarmerRequest> {
 
     private String name;
 
     private SearchRequest searchRequest;
 
+    private TimeValue timeout = timeValueSeconds(10);
+
     PutWarmerRequest() {
 
     }
@@ -86,6 +94,22 @@ SearchRequest searchRequest() {
         return this.searchRequest;
     }
 
+    @Override
+    public PutWarmerRequest timeout(String timeout) {
+        return this;
+    }
+
+    @Override
+    public PutWarmerRequest timeout(TimeValue timeout) {
+        this.timeout = timeout;
+        return this;
+    }
+
+    @Override
+    public TimeValue timeout() {
+        return timeout;
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = searchRequest.validate();
@@ -103,6 +127,9 @@ public void readFrom(StreamInput in) throws IOException {
             searchRequest = new SearchRequest();
             searchRequest.readFrom(in);
         }
+        if (in.getVersion().onOrAfter(Version.V_0_90_6)) {
+            timeout = readTimeValue(in);
+        }
     }
 
     @Override
@@ -115,5 +142,8 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(true);
             searchRequest.writeTo(out);
         }
+        if (out.getVersion().onOrAfter(Version.V_0_90_6)) {
+            timeout.writeTo(out);
+        }
     }
 }

src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerRequestBuilder.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.action.support.master.MasterNodeOperationRequestBuilder;
 import org.elasticsearch.client.IndicesAdminClient;
 import org.elasticsearch.client.internal.InternalIndicesAdminClient;
+import org.elasticsearch.common.unit.TimeValue;
 
 /**
  *
@@ -63,6 +64,23 @@ public PutWarmerRequestBuilder setSearchRequest(SearchRequestBuilder searchReque
         return this;
     }
 
+    /**
+     * Sets the maximum wait for acknowledgement from other nodes
+     */
+    public PutWarmerRequestBuilder setTimeout(TimeValue timeout) {
+        request.timeout(timeout);
+        return this;
+    }
+
+    /**
+     * Timeout to wait for the operation to be acknowledged by current cluster nodes. Defaults
+     * to <tt>10s</tt>.
+     */
+    public PutWarmerRequestBuilder setTimeout(String timeout) {
+        request.timeout(timeout);
+        return this;
+    }
+
     @Override
     protected void doExecute(ActionListener<PutWarmerResponse> listener) {
         ((IndicesAdminClient) client).putWarmer(request, listener);

src/main/java/org/elasticsearch/action/admin/indices/warmer/put/PutWarmerResponse.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.admin.indices.warmer.put;
 
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 
@@ -28,7 +29,7 @@
 /**
  * The response of put warmer operation.
  */
-public class PutWarmerResponse extends ActionResponse {
+public class PutWarmerResponse extends ActionResponse implements AcknowledgedResponse {
 
     private boolean acknowledged;