You need a working ElasticSearch node, a Maven built tool, and an optional Java IDE. The code of this recipe is available in the chapter12/rest_plugin directory.

In this recipe, we have seen that a REST call is converted to an internal cluster action.

To execute an internal cluster action, the following classes are required:

We need to perform the following steps:

As you have seen previously for executing a cluster action the following classes are required:

These classes must extend one of the supported kinds of action available:

In our example, we have defined an action that will be broadcast to every shard as follows:

public class SimpleRequest extends BroadcastOperationRequest<SimpleRequest>

All the Request/Response classes extend a Streamable class, so the two following methods for serializing their content must be provided:

In both the actions, super must be called to allow the correct serialization of parent classes.

To complete the request/response action, we must define an action that binds the request with the correct response and a builder to construct it. To do so, we need to define an Action class as follows:

public class SimpleAction extends Action<SimpleRequest, SimpleResponse, SimpleRequestBuilder>

This Action object is a singleton object. We obtain it by creating a default static instance and private constructors:

public static final SimpleAction INSTANCE = new SimpleAction();
public static final String NAME = "indices/simple";
private SimpleAction() {super(NAME); }

The static string, NAME is used to univocally define the action at cluster level.

To complete the Action definition, the following two methods must be defined:

When the action is executed, the request and the response are serialized and sent to the cluster. To execute our custom code at cluster level, a transport action is required.

The transport actions are usually defined as map and reduce jobs. The map part consists of executing the action on several shards (via the ShardRequest and ShardResponse classes) and the reduce part consists of collecting all the results from the shards in a response that must be sent back to the requester.

The transport action is a long class with many methods, but the most important ones are the ShardOperation (map part) and newResponse (reduce part).

The original request is converted into a distributed ShardRequest that is processed by the shardOperation method as follows:

@Override protected ShardSimpleResponse shardOperation(ShardSimpleRequest request) throws ElasticSearchException {

It is good practice to execute the shard operation using a lock to prevent concurrency problems.

synchronized (simpleMutex) {…}

To obtain the internal shard, we need to ask at the IndexService class to return a shard based on the wanted index.

The shard request contains the index and the ID of the shard that must be used to execute the action.

InternalIndexShard indexShard = (InternalIndexShard) indicesService.indexServiceSafe(request.index()).shardSafe(request.shardId());

The InternalIndexShard object allows executing every possible shard operation (search, get, index, and many others). In this method, we can execute every data shard manipulation that we want.

In the preceding example, we have created a simple set of values as follows:

Set<String> set = new HashSet<String>();
set.add(request.getField() + "_" + request.shardId());

The final step of our shard operation is to create a response to send back to the reduce step. While creating the shard response we need to return the result in addition to the information about the index and the shard that executed the action, as given in the following code:

return new ShardSimpleResponse(request.index(), request.shardId(), set);

The distributed shard operations are collected in the reduce step (the newResponse method). In this step, we need to aggregate all the shard results and produce the result to send back to the original Action.

@Override protected SimpleResponse newResponse(SimpleRequest request, AtomicReferenceArray shardsResponses, ClusterState clusterState) {

Other than the result, we also need to collect the information about the shard execution (if there are failures on them). This information is usually collected in three values: successfulShards, failedShards, and shardFailures.

int successfulShards = 0;
int failedShards = 0;
List<ShardOperationFailedException> shardFailures = null;

The request result is a set of collected strings.

Set<String> simple = new HashSet<String>();

To collect the results, we need to iterate on the shard responses as follows:

for (int i = 0; i < shardsResponses.length(); i++) {
    Object shardResponse = shardsResponses.get(i);

We need to skip the null shardResponse, mainly due to inactive shards.

if (shardResponse == null) {} 

If a failure is raised, we also need to collect them to inform the caller.

else if (shardResponse instanceof BroadcastShardOperationFailedException) {
   failedShards++;
   if (shardFailures == null) {
       shardFailures = newArrayList();
   }
   shardFailures.add(new DefaultShardOperationFailedException((BroadcastShardOperationFailedException) shardResponse));

We can aggregate the valid results as follows:

} else {
    successfulShards++;
     if (shardResponse instanceof ShardSimpleResponse) {
         ShardSimpleResponse resp = (ShardSimpleResponse) shardResponse;
         simple.addAll(resp.getTermList());
     }
}

The final step is to create the response collecting the previous result and response status.

return new SimpleResponse(shardsResponses.length(), successfulShards, failedShards, shardFailures, simple); 

Creating a cluster action is required when there are low-level operations that we want to execute very quickly, such as special facet or complex manipulation that requires the ElasticSearch call to be executed, but that can be easily written as a cluster action.