change broadcast support to be able to run on all shards replicas (in parallel) and not just one shard per replica group. Change flush and refresh to use broadcast and not replicaiton. Remove shards transport support since broadcast now does exactly the same, and refactor index status to use broadcast (across all shards).

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/action/TransportActionModule.java

@@ -30,15 +30,11 @@
 import org.elasticsearch.action.admin.indices.create.TransportCreateIndexAction;
 import org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction;
 import org.elasticsearch.action.admin.indices.flush.TransportFlushAction;
-import org.elasticsearch.action.admin.indices.flush.TransportIndexFlushAction;
-import org.elasticsearch.action.admin.indices.flush.TransportShardFlushAction;
 import org.elasticsearch.action.admin.indices.gateway.snapshot.TransportGatewaySnapshotAction;
 import org.elasticsearch.action.admin.indices.gateway.snapshot.TransportIndexGatewaySnapshotAction;
 import org.elasticsearch.action.admin.indices.gateway.snapshot.TransportShardGatewaySnapshotAction;
 import org.elasticsearch.action.admin.indices.mapping.create.TransportCreateMappingAction;
-import org.elasticsearch.action.admin.indices.refresh.TransportIndexRefreshAction;
 import org.elasticsearch.action.admin.indices.refresh.TransportRefreshAction;
-import org.elasticsearch.action.admin.indices.refresh.TransportShardRefreshAction;
 import org.elasticsearch.action.admin.indices.status.TransportIndicesStatusAction;
 import org.elasticsearch.action.count.TransportCountAction;
 import org.elasticsearch.action.delete.TransportDeleteAction;
@@ -76,18 +72,11 @@ public class TransportActionModule extends AbstractModule {
         bind(TransportIndexGatewaySnapshotAction.class).asEagerSingleton();
         bind(TransportGatewaySnapshotAction.class).asEagerSingleton();
 
-        bind(TransportShardRefreshAction.class).asEagerSingleton();
-        bind(TransportIndexRefreshAction.class).asEagerSingleton();
         bind(TransportRefreshAction.class).asEagerSingleton();
-
-        bind(TransportShardFlushAction.class).asEagerSingleton();
-        bind(TransportIndexFlushAction.class).asEagerSingleton();
         bind(TransportFlushAction.class).asEagerSingleton();
 
         bind(TransportIndexAction.class).asEagerSingleton();
-
         bind(TransportGetAction.class).asEagerSingleton();
-
         bind(TransportDeleteAction.class).asEagerSingleton();
 
         bind(TransportShardDeleteByQueryAction.class).asEagerSingleton();

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/TransportBroadcastPingAction.java

@@ -24,6 +24,8 @@
 import org.elasticsearch.action.TransportActions;
 import org.elasticsearch.action.support.broadcast.TransportBroadcastOperationAction;
 import org.elasticsearch.cluster.ClusterService;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.routing.GroupShardsIterator;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -32,6 +34,8 @@
 
 import java.util.concurrent.atomic.AtomicReferenceArray;
 
+import static org.elasticsearch.action.Actions.*;
+
 /**
  * @author kimchy (Shay Banon)
  */
@@ -53,7 +57,11 @@ public class TransportBroadcastPingAction extends TransportBroadcastOperationAct
         return new BroadcastPingRequest();
     }
 
-    @Override protected BroadcastPingResponse newResponse(BroadcastPingRequest broadcastPingRequest, AtomicReferenceArray shardsResponses) {
+    @Override protected GroupShardsIterator shards(BroadcastPingRequest request, ClusterState clusterState) {
+        return indicesService.searchShards(clusterState, processIndices(clusterState, request.indices()), request.queryHint());
+    }
+
+    @Override protected BroadcastPingResponse newResponse(BroadcastPingRequest request, AtomicReferenceArray shardsResponses, ClusterState clusterState) {
         int successfulShards = 0;
         int failedShards = 0;
         for (int i = 0; i < shardsResponses.length(); i++) {

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java

@@ -19,33 +19,35 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.support.replication.IndicesReplicationOperationRequest;
-import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
+import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
 
 /**
  * @author kimchy (Shay Banon)
  */
-public class FlushRequest extends IndicesReplicationOperationRequest {
+public class FlushRequest extends BroadcastOperationRequest {
+
+    FlushRequest() {
+
+    }
 
     public FlushRequest(String index) {
         this(new String[]{index});
     }
 
     public FlushRequest(String... indices) {
-        this.indices = indices;
+        super(indices, null);
+        // we want to do the refresh in parallel on local shards...
+        operationThreading(BroadcastOperationThreading.THREAD_PER_SHARD);
     }
 
     @Override public FlushRequest listenerThreaded(boolean threadedListener) {
         super.listenerThreaded(threadedListener);
         return this;
     }
 
-    public FlushRequest timeout(TimeValue timeout) {
-        this.timeout = timeout;
+    @Override public FlushRequest operationThreading(BroadcastOperationThreading operationThreading) {
+        super.operationThreading(operationThreading);
         return this;
     }
-
-    FlushRequest() {
-
-    }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushResponse.java

@@ -19,47 +19,30 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * @author kimchy (Shay Banon)
  */
-public class FlushResponse implements ActionResponse, Streamable {
-
-    private Map<String, IndexFlushResponse> indices = new HashMap<String, IndexFlushResponse>();
+public class FlushResponse extends BroadcastOperationResponse {
 
     FlushResponse() {
 
     }
 
-    public Map<String, IndexFlushResponse> indices() {
-        return indices;
-    }
-
-    public IndexFlushResponse index(String index) {
-        return indices.get(index);
+    FlushResponse(int successfulShards, int failedShards) {
+        super(successfulShards, failedShards);
     }
 
     @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
-        int size = in.readInt();
-        for (int i = 0; i < size; i++) {
-            IndexFlushResponse indexFlushResponse = new IndexFlushResponse();
-            indexFlushResponse.readFrom(in);
-            indices.put(indexFlushResponse.index(), indexFlushResponse);
-        }
+        super.readFrom(in);
     }
 
     @Override public void writeTo(DataOutput out) throws IOException {
-        out.writeInt(indices.size());
-        for (IndexFlushResponse indexFlushResponse : indices.values()) {
-            indexFlushResponse.writeTo(out);
-        }
+        super.writeTo(out);
     }
-}
+}

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/IndexFlushRequest.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.support.replication.ShardReplicationOperationRequest;
+import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
 
 import java.io.DataInput;
 import java.io.DataOutput;
@@ -28,34 +28,20 @@
 /**
  * @author kimchy (Shay Banon)
  */
-public class ShardFlushRequest extends ShardReplicationOperationRequest {
-
-    private int shardId;
-
-    public ShardFlushRequest(IndexFlushRequest indexFlushRequest, int shardId) {
-        this(indexFlushRequest.index(), shardId);
-        timeout = indexFlushRequest.timeout();
-    }
-
-    public ShardFlushRequest(String index, int shardId) {
-        this.index = index;
-        this.shardId = shardId;
-    }
+public class ShardFlushRequest extends BroadcastShardOperationRequest {
 
     ShardFlushRequest() {
     }
 
-    public int shardId() {
-        return this.shardId;
+    public ShardFlushRequest(String index, int shardId) {
+        super(index, shardId);
     }
 
     @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
         super.readFrom(in);
-        shardId = in.readInt();
     }
 
     @Override public void writeTo(DataOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(shardId);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/IndexFlushResponse.java

@@ -19,8 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
 
 import java.io.DataInput;
 import java.io.DataOutput;
@@ -29,15 +28,21 @@
 /**
  * @author kimchy (Shay Banon)
  */
-public class ShardFlushResponse implements ActionResponse, Streamable {
+public class ShardFlushResponse extends BroadcastShardOperationResponse {
 
     ShardFlushResponse() {
 
     }
 
+    public ShardFlushResponse(String index, int shardId) {
+        super(index, shardId);
+    }
+
     @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        super.readFrom(in);
     }
 
     @Override public void writeTo(DataOutput out) throws IOException {
+        super.writeTo(out);
     }
 }